src/components/etcd/etcd.go

package etcd
import (
	"context"
	"fmt"
	clientv3 "go.etcd.io/etcd/client/v3"
	"pro2d/conf"
	"pro2d/src/components/logger"
	"time"
)
type EtcdClient struct {
	etcd *clientv3.Client
}
func NewEtcdClient(conf *conf.Etcd) *EtcdClient {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:   conf.Endpoints,
		DialTimeout:  time.Duration(conf.DialTimeout) * time.Second,
	})
	if err != nil {
		logger.Error("etcd init err: %v", err)
		return nil
	}
	return &EtcdClient{
		etcd: cli,
	}
}
func (e *EtcdClient)PutWithPrefix(prefix, key, val string)  {
	_, err := e.etcd.Put(context.TODO(), fmt.Sprintf("/%s/%s/", prefix, key), val)
	if err != nil {
		logger.Error("PutWithPrefix err: %v", err)
		return
	}
}
func (e *EtcdClient)PutWithLeasePrefix(prefix, key, val string, ttl int64) error {
	lease := clientv3.NewLease(e.etcd)
	leaseResp, err := lease.Grant(context.TODO(), ttl)
	if err != nil {
		logger.Error("PutWithLeasePrefix 设置租约时间失败:%v\n", err)
		return err
	}
	_, err = e.etcd.Put(context.TODO(), fmt.Sprintf("/%s/%s/", prefix, key), val, clientv3.WithLease(leaseResp.ID))
	if err != nil {
		logger.Error("PutWithLeasePrefix err: %v", err)
		return err
	}
	keepRespChan, err := lease.KeepAlive(context.TODO(), leaseResp.ID)
	if err != nil {
		logger.Error("keepalive err: %v", err)
		return err
	}
	go func() {
		for {
			select {
			case _ = <-keepRespChan:
				if keepRespChan == nil {
					fmt.Println("租约已经失效")
					goto END
				} else { //每秒会续租一次，所以就会受到一次应答
					//fmt.Println("收到自动续租应答:", keepResp.ID)
				}
			}
		}
		END:
	}()
	return nil
}
func (e *EtcdClient)Get(key string) map[string]string {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	resp, err := e.etcd.Get(ctx, fmt.Sprintf("/%s/", key))
	cancel()
	if err != nil {
		logger.Error("etcd get key: %s, err: %v", key, err)
		return nil
	}
	m := make(map[string]string)
	for _, v := range resp.Kvs {
		m[string(v.Key)] = string(v.Value)
	}
	return m
}
func (e *EtcdClient)GetByPrefix(prefix string) map[string]string {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	resp, err := e.etcd.Get(ctx, fmt.Sprintf("/%s/", prefix), clientv3.WithPrefix())
	cancel()
	if err != nil {
		logger.Error("etcd get prefix: %s, err: %v", prefix, err)
		return nil
	}
	m := make(map[string]string)
	for _, v := range resp.Kvs {
		m[string(v.Key)] = string(v.Value)
	}
	return m
}
func (e *EtcdClient)Close()  {
	e.etcd.Close()
}