转载自：实现etcd服务注册与发现
0.1、目录结构

. ├── api │└── main.go ├── common │└── common.go ├── docker-compose.yml ├── etcd │└── Dockerfile ├── go.mod ├── go.sum ├── rpc │├── courseware ││├── courseware.pb.go ││└── courseware_grpc.pb.go │├── courseware.proto │└── main.go └── server ├── service_discovery.go └── service_registration.go

1、docker-compose部署一个3节点的集群
项目根目录下创建etcd目录，并在目录下新增Dockerfile文件

FROM bitnami/etcd:latestLABEL maintainer="liuyuede123 "

项目根目录下新增docker-compose.yml

version: '3.5' # 网络配置 networks: backend: driver: bridge# 服务容器配置 services: etcd1:# 自定义容器名称 build: context: etcd# 指定构建使用的 Dockerfile 文件 environment: - TZ=Asia/Shanghai - ALLOW_NONE_AUTHENTICATION=yes - ETCD_NAME=etcd1 - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380 - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380 - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379 - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379 - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 - ETCD_INITIAL_CLUSTER_STATE=new ports:# 设置端口映射 - "12379:2379" - "12380:2380" networks: - backend restart: alwaysetcd2: # 自定义容器名称 build: context: etcd# 指定构建使用的 Dockerfile 文件 environment: - TZ=Asia/Shanghai - ALLOW_NONE_AUTHENTICATION=yes - ETCD_NAME=etcd2 - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380 - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380 - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379 - ETCD_ADVERTISE_CLIENT_URLS=http://etcd2:2379 - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 - ETCD_INITIAL_CLUSTER_STATE=new ports: # 设置端口映射 - "22379:2379" - "22380:2380" networks: - backend restart: alwaysetcd3: # 自定义容器名称 build: context: etcd# 指定构建使用的 Dockerfile 文件 environment: - TZ=Asia/Shanghai - ALLOW_NONE_AUTHENTICATION=yes - ETCD_NAME=etcd3 - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd3:2380 - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380 - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379 - ETCD_ADVERTISE_CLIENT_URLS=http://etcd3:2379 - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380 - ETCD_INITIAL_CLUSTER_STATE=new ports: # 设置端口映射 - "32379:2379" - "32380:2380" networks: - backend restart: always

相关参数概念：

1. ETCD_INITIAL_ADVERTISE_PEER_URLS：该成员节点在整个集群中的通信地址列表，这个地址用来传输集群数据的地址。因此这个地址必须是可以连接集群中所有的成员的。
2. ETCD_LISTEN_PEER_URLS：该节点与其他节点通信时所监听的地址列表，多个地址使用逗号隔开，其格式可以划分为scheme://IP:PORT，这里的scheme可以是http、https
3. ETCD_LISTEN_CLIENT_URLS：该节点与客户端通信时监听的地址列表
4. ETCD_ADVERTISE_CLIENT_URLS：广播给集群中其他成员自己的客户端地址列表
5. ETCD_INITIAL_CLUSTER_TOKEN：初始化集群token
6. ETCD_INITIAL_CLUSTER：配置集群内部所有成员地址，其格式为：ETCD_NAME=ETCD_INITIAL_ADVERTISE_PEER_URLS，如果有多个使用逗号隔开
7. ETCD_INITIAL_CLUSTER_STATE：初始化集群状态，new表示新建

启动集群

docker-compose up -d Creating network "etcd_backend" with driver "bridge" Creating etcd_etcd1_1 ... done Creating etcd_etcd2_1 ... done Creating etcd_etcd3_1 ... done

测试集群可用性

# 登录其中一个节点 docker exec -it 5f97bf0b446f6e6514576fc1eb46c2f60d2c2b3e3f3ee3b1ad6219414fa915c8 /bin/sh # 写入一个键值 etcdctl put name "liuyuede" OK # 查看 etcdctl get name name liuyuede# 登录另外俩个节点 docker exec -it a6ccc9b6e5cc81ee7c779e2b9e7235cd6d814e92fbc66b7e4846798acff8ee2a /bin/sh etcdctl get name name liuyuededocker exec -it 6817fa89e3e9e422628e0049910b672df389c62d41bf2349a0f77e22c99e5270 /bin/sh etcdctl get name name liuyuede

etcd集群采用的是raft协议，一般至少为俩个集群，只有一个master，如果删除到只剩一个节点当前节点也不能提供服务
查看集群情况

etcdctl --endpoints=http://0.0.0.0:12379,0.0.0.0:22379,0.0.0.0:32379 endpoint status --write-out=table+----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+ |ENDPOINT|ID| VERSION | DB SIZE | IS LEADER | IS LEARNER | RAFT TERM | RAFT INDEX | RAFT APPLIED INDEX | ERRORS | +----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+ | http://0.0.0.0:12379 | ade526d28b1f92f7 |3.5.4 |20 kB |true |false |3 |13 |13 || |0.0.0.0:22379 | d282ac2ce600c1ce |3.5.4 |20 kB |false |false |3 |13 |13 || |0.0.0.0:32379 | bd388e7810915853 |3.5.4 |20 kB |false |false |3 |13 |13 || +----------------------+------------------+---------+---------+-----------+------------+-----------+------------+--------------------+--------+

2、增加服务注册功能
服务注册的流程

1. 向etcd新增一个包含rpc服务信息的键值对，并设置租约（比如5秒过期）
2. 利用保活函数KeepAlive不断续约

package serverimport ( "context" "encoding/json" "errors" clientv3 "go.etcd.io/etcd/client/v3" "time" )type ServiceInfo struct { Name string Ipstring }type Service struct { ServiceInfo ServiceInfo stopchan error leaseIdclientv3.LeaseID client*clientv3.Client }func NewService(serviceInfo ServiceInfo, endpoints []string) (service *Service, err error) { client, err := clientv3.New(clientv3.Config{ Endpoints:endpoints, DialTimeout: time.Second * 10, }) if err != nil { return nil, err }service = &Service{ ServiceInfo: serviceInfo, client:client, } return }func (s *Service) Start(ctx context.Context) (err error) { alive, err := s.KeepAlive(ctx) if err != nil { return }for { select { case err = <-s.stop: // 服务端关闭返回错误 return err case <-s.client.Ctx().Done(): // etcd关闭 return errors.New("server closed") case _, ok := <-alive: if !ok { // 保活通道关闭 return s.revoke(ctx) } } } }func (s *Service) KeepAlive(ctx context.Context) (<-chan *clientv3.LeaseKeepAliveResponse, error) { info := s.ServiceInfo key := s.getKey() val, _ := json.Marshal(info)// 创建租约 leaseResp, err := s.client.Grant(ctx, 5) if err != nil { return nil, err }// 写入etcd _, err = s.client.Put(ctx, key, string(val), clientv3.WithLease(leaseResp.ID)) if err != nil { return nil, err }s.leaseId = leaseResp.ID return s.client.KeepAlive(ctx, leaseResp.ID) }// 取消租约 func (s *Service) revoke(ctx context.Context) error { _, err := s.client.Revoke(ctx, s.leaseId) return err }func (s *Service) getKey() string { return s.ServiceInfo.Name + "/" + s.ServiceInfo.Ip }

3、增加服务发现
服务发现流程

1. 实现grpc中resolver.Builder接口的Build方法
2. 通过etcdclient获取并监听grpc服务（是否有新增或者删除）
3. 更新到resolver.State，State 包含与 ClientConn 相关的当前 Resolver 状态，包括grpc的地址resolver.Address

package serverimport ( "context" "encoding/json" "fmt" "go.etcd.io/etcd/api/v3/mvccpb" clientv3 "go.etcd.io/etcd/client/v3" "google.golang.org/grpc/resolver" )type Discovery struct { endpoints[]string servicestring client*clientv3.Client clientConn resolver.ClientConn }func NewDiscovery(endpoints []string, service string) resolver.Builder { return &Discovery{ endpoints: endpoints, service:service, } }func (d *Discovery) ResolveNow(rn resolver.ResolveNowOptions) {}func (d *Discovery) Close() {}func (d *Discovery) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) { var err error d.client, err = clientv3.New(clientv3.Config{ Endpoints: d.endpoints, }) if err != nil { return nil, err }d.clientConn = ccgo d.watch(d.service)return d, nil }func (d *Discovery) Scheme() string { return "etcd" }func (d *Discovery) watch(service string) { addrM := make(map[string]resolver.Address) state := resolver.State{}update := func() { addrList := make([]resolver.Address, 0, len(addrM)) for _, address := range addrM { addrList = append(addrList, address) } state.Addresses = addrList err := d.clientConn.UpdateState(state) if err != nil { fmt.Println("更新地址出错：", err) } } resp, err := d.client.Get(context.Background(), service, clientv3.WithPrefix()) if err != nil { fmt.Println("获取地址出错：", err) } else { for i, kv := range resp.Kvs { info := &ServiceInfo{} err = json.Unmarshal(kv.Value, info) if err != nil { fmt.Println("解析value失败：", err) } addrM[string(resp.Kvs[i].Key)] = resolver.Address{ Addr:info.Ip, ServerName: info.Name, } } }update()dch := d.client.Watch(context.Background(), service, clientv3.WithPrefix(), clientv3.WithPrevKV()) for response := range dch { for _, event := range response.Events { switch event.Type { case mvccpb.PUT: info := &ServiceInfo{} err = json.Unmarshal(event.Kv.Value, info) if err != nil { fmt.Println("监听时解析value报错：", err) } else { addrM[string(event.Kv.Key)] = resolver.Address{Addr: info.Ip} } fmt.Println(string(event.Kv.Key)) case mvccpb.DELETE: delete(addrM, string(event.Kv.Key)) fmt.Println(string(event.Kv.Key)) } } update() } }

4、grpc课件服务
common参数

package commonconst CoursewareRpc = "rpc.courseware"var Endpoints = []string{"127.0.0.1:12379", "127.0.0.1:22379", "127.0.0.1:32379"}

生成课件服务grpc

syntax = "proto3"; package rpc; option go_package = "./courseware"; message GetRequest { uint64 Id = 1; } message GetResponse { uint64 Id = 1; string Code = 2; string Name = 3; uint64 Type = 4; }service Courseware { rpc Get(GetRequest) returns(GetResponse); }

protoc --go_out=./ --go-grpc_out=./ courseware.proto

课件服务入口

package mainimport ( "context" "fmt" "go-demo/etcd/common" "go-demo/etcd/rpc/courseware" "go-demo/etcd/server" "google.golang.org/grpc" "google.golang.org/grpc/reflection" "net" "os" "strings" "time" )var Port stringtype service struct { courseware.UnsafeCoursewareServer }func (s *service) Get(ctx context.Context, req *courseware.GetRequest) (res *courseware.GetResponse, err error) { fmt.Println("获取课件详情 port:", Port, " time:", time.Now()) return &courseware.GetResponse{ Id:1, Code: "HD4544", Name: "多媒体课件", Type: 4, }, nil } func main() { args := os.Args[1:] if len(args) == 0 { panic("缺少port参数：port=8400") } for _, arg := range args { ports := strings.Split(arg, "=") if len(ports) < 2 || ports[0] != "port" { panic("port参数格式错误：port=8400") } Port = ports[1] } listen, err := net.Listen("tcp", ":"+Port) if err != nil { fmt.Println("failed to listen", err) return } s := grpc.NewServer() courseware.RegisterCoursewareServer(s, &service{})reflection.Register(s)// 注册到etcd newService, err := server.NewService(server.ServiceInfo{ Name: common.CoursewareRpc, Ip:"127.0.0.1:" + Port, }, common.Endpoints) if err != nil { fmt.Println("添加到etcd失败：", err) return }go func() { err = newService.Start(context.Background()) if err != nil { fmt.Println("开启服务注册失败：", err) } }()if err = s.Serve(listen); err != nil { fmt.Println("开启rpc服务失败：", err) } }

5、api服务

package mainimport ( "context" "fmt" "go-demo/etcd/common" "go-demo/etcd/rpc/courseware" "go-demo/etcd/server" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/resolver" "time" )func main() { d := server.NewDiscovery(common.Endpoints, common.CoursewareRpc) resolver.Register(d)for { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) // 通过etcd注册中心和grpc服务建立连接 conn, err := grpc.DialContext(ctx, fmt.Sprintf(d.Scheme()+":///"+common.CoursewareRpc), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`), grpc.WithBlock(), ) if err != nil { fmt.Println("和rpc建立连接失败：", err) return }client := courseware.NewCoursewareClient(conn) get, err := client.Get(ctx, &courseware.GetRequest{Id: 1}) if err != nil { fmt.Println("获取课件失败：", err) return }fmt.Println(get)time.Sleep(3 * time.Second) cancel() }}

6、测试
【实现etcd服务注册与发现】开启3个服务，可以看到客户端通过负载均衡随机到一个服务请求

go run main.go port=8400 获取课件详情 port: 8400time: 2022-08-25 18:47:43.784942 +0800 CST m=+78.228450885 获取课件详情 port: 8400time: 2022-08-25 18:47:52.925858 +0800 CST m=+87.369721731 获取课件详情 port: 8400time: 2022-08-25 18:48:02.001177 +0800 CST m=+96.445393312 获取课件详情 port: 8400time: 2022-08-25 18:48:05.060066 +0800 CST m=+99.504401028 获取课件详情 port: 8400time: 2022-08-25 18:48:14.154148 +0800 CST m=+108.598836458 go run main.go port=8500 获取课件详情 port: 8500time: 2022-08-25 18:47:46.832479 +0800 CST m=+62.822399701 获取课件详情 port: 8500time: 2022-08-25 18:47:49.844536 +0800 CST m=+65.834573960 获取课件详情 port: 8500time: 2022-08-25 18:47:55.955638 +0800 CST m=+71.945912584 获取课件详情 port: 8500time: 2022-08-25 18:48:17.168293 +0800 CST m=+93.159391485 获取课件详情 port: 8500time: 2022-08-25 18:48:20.182787 +0800 CST m=+96.174002796 go run main.go port=8600 获取课件详情 port: 8600time: 2022-08-25 18:47:58.968283 +0800 CST m=+1.317052360 获取课件详情 port: 8600time: 2022-08-25 18:48:08.106493 +0800 CST m=+10.455617422 获取课件详情 port: 8600time: 2022-08-25 18:48:11.125212 +0800 CST m=+13.474453269

推荐阅读

• 

  佳能750d可以换什么样的镜头
• 

  面料涤纶里料棉纶的衣服怎么洗 面料涤纶里料棉纶的衣服如何洗
• 

  快赔和保险有什么区别
• 

  mole|预算不超过2000元，要有5g、不卡，朋友们能否给点建议？
• 

  崇德老人八字例解（六）
• 

  痛风后果严重吗
• 

  stm32逻辑分析porta
• 

  秋葵焯水要切开不
• 

  万能看图王怎么卸载？万能看图王卸载方法
• 

  含有近义词的词语 含有近义词的词语 四字
• 

  利用手机QQ无限刷手机上网流量 三网通用
• 

  新冠疫苗接种后的注意事项 打新冠疫苗可以吃生鱼片吗
• 

  戒烟的过程中有点心慌是什么情况，要怎么办啊？
• 

  得了艾滋病应该怎么办 如果患上艾滋病怎么办
• 

  夏天如何培养天竺葵盆栽 夏天如何培养天竺葵，夏天如何培养天竺葵苗
• 

  wi-fi|联想发布 Yoga 7i 翻转本：可选 14/16 英寸屏
• 

  分享一下你们最喜欢的微信头像？
• 

  《雪鹰领主》进入薪火宫的条件 雪鹰领主招募
• 

  虚拟主机如何搭建shopex，虚拟主机如何搭建传奇游戏
• 

  考研申请书

• 服务器|Ajax跨域与封装
• Spring|Spring Boot 中实现跨域的 5 种方式
• 基于 Serverless+OSS 分分钟实现图片秒变素描
• 汽车出行|AITO问界M7发布51天后开启交付，营销服务网络超700家
• RabbitMQ|消息中间件RabbitMQ（五）——实现RPC调用
• 微服务|【黑马-SpringCloud技术栈】【09】消息队列—RabbitMQ_SpringAMQP
• k8s中如何实现pod自动扩缩容详解
• SpringBoot+JSON+AJAX+ECharts+Fiddler实现前后端分离开发可视化
• 一步步教你实现微信小程序自定义组件
• javascript动态分页的实现方法实例