在系统中用etcd实现服务注册和发现

// 将服务注册到etcd上func RegisterServiceToETCD(ServiceTarget string, value string) {    dir = strings.TrimRight(ServiceTarget, "/") + "/"
    client, err := clientv3.New(clientv3.Config{        Endpoints:   []string{"localhost:2379"},        DialTimeout: 5 * time.Second,    })    if err != nil {    panic(err)    }
    kv := clientv3.NewKV(client)    lease := clientv3.NewLease(client)    var curLeaseId clientv3.LeaseID = 0
    for {        if curLeaseId == 0 {            leaseResp, err := lease.Grant(context.TODO(), 10)            if err != nil {              panic(err)            }
            key := ServiceTarget + fmt.Sprintf("%d", leaseResp.ID)            if _, err := kv.Put(context.TODO(), key, value, clientv3.WithLease(leaseResp.ID)); err != nil {                  panic(err)            }            curLeaseId = leaseResp.ID        } else {      // 续约租约，如果租约已经过期将curLeaseId复位到0重新走创建租约的逻辑            if _, err := lease.KeepAliveOnce(context.TODO(), curLeaseId); err == rpctypes.ErrLeaseNotFound {                curLeaseId = 0                continue            }        }        time.Sleep(time.Duration(1) * time.Second)    }}

type HelloService struct {}
func (p *HelloService) Hello(request string, reply *string) error {    *reply = "hello:" + request    return nil}
var serviceTarget = "Hello"var port = ":1234"var host = "remote_host"// 伪代码
func main() {    rpc.RegisterName("HelloService", new(HelloService))
    listener, err := net.Listen("tcp", port)    if err != nil {        log.Fatal("ListenTCP error:", err)    }
    conn, err := listener.Accept()    if err != nil {        log.Fatal("Accept error:", err)    }
    go RegisterServiceToETCD(serviceTarget,  host + port)    rpc.ServeConn(conn)}

var serviceTarget = "Hello"type remoteService struct {  name string  nodes map[string]string  mutex sync.Mutex}
// 获取服务目录下所有key初始化到服务的可用节点列表中func getService(etcdClient clientv3.Client) *remoteService {    service = &remoteService {      name: serviceTarget    }     kv := clientv3.NewKV(etcdClient)    rangeResp, err := kv.Get(context.TODO(), service.name, clientv3.WithPrefix())    if err != nil {       panic(err)    }
    service.mutex.Lock()    for _, kv := range rangeResp.Kvs {        service.nodes[string(kv.Key)] = string(kv.Value)    }    service.mutex.Unlock()
    go watchServiceUpdate(etcdClient, service)}
// 监控服务目录下的事件func watchServiceUpdate(etcdClient clientv3.Client, service *remoteService) {    watcher := clientv3.NewWatcher(client)    // Watch 服务目录下的更新    watchChan := watcher.Watch(context.TODO(), service.name, clientv3.WithPrefix())    for watchResp := range watchChan {          for _, event := range watchResp.Events {                service.mutex.Lock()                switch (event.Type) {                case mvccpb.PUT://PUT事件，目录下有了新key                      service.nodes[string(event.Kv.Key)] = string(event.Kv.Value)            case mvccpb.DELETE://DELETE事件，目录中有key被删掉(Lease过期，key 也会被删掉)                      delete(service.nodes, string(event.Kv.Key))                }                service.mutex.Unlock()          }    }}
func main () {  client, err := clientv3.New(clientv3.Config{        Endpoints:   []string{"remote_host:2379"},        DialTimeout: 5 * time.Second,    })  service := getService(client)// 获取服务的可用节点  ......  // 每次有请求过来从服务节点中选取一个连接，然后给节点发送请求  rpcClient, _ = rpc.Dial("tcp", service.nodes[i])  var reply string  rpcClient.Call("HelloService.hello", &reply)  ......
}