主要使用ETCD来实现服务注册和服务发现
ETCD集群搭建参考https://www.jianshu.com/p/690d8adcb8c6
为什么使用grpc前要实现服务发现、服务注册?
考虑微服务水平拓展较为容易,grpc包本身没有提供服务发现和服务注册,那么在使用过程中,对应的rpc只能通过nginx等服务去进行负载均衡,而当微服务需要水平拓展时,因为没有服务发现,因此要修改nginx配置才能将该节点加入进行使用,因此需要更加智能的动态添加和删除节点,所以grpc也不必使用nginx进行负载均衡,而是在client端进行负载均衡的实现。
首先下载ETCD官方golang驱动包https://github.com/etcd-io/etcd/
1 | go get -u -v github.com/etcd-io/etcd |
但是好像github服务器的问题,不太好下载,速度太慢,去码云找了一个镜像https://gitee.com/mirrors/etcd
下载之后记得在$GOPATH路径下新建文件夹go.etcd.io,将etcd包放入该目录下。否则会出现获取etcd 的response时候,缺少Type,Kv,PrevKv等变量,放入go.etcd.io之后则可以正常使用。
image.png
废话不多说,直接上代码
服务注册
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | package discovery import ( "context" "go.etcd.io/etcd/clientv3" "time" "log" "encoding/json" "errors" ) //the detail of service 定义服务结构,唯一id加ip地址 type ServiceInfo struct{ ID uint64 IP string } type Service struct { Name string Info ServiceInfo stop chan error leaseid clientv3.LeaseID client *clientv3.Client } func NewService(name string, info ServiceInfo, endpoints []string) (*Service, error) { cli, err := clientv3.New(clientv3.Config{ Endpoints: endpoints, DialTimeout: 2 * time.Second, }) if err != nil { log.Fatal(err) return nil, err } return &Service { Name: name, Info: info, stop: make (chan error), client: cli, },err } func (s *Service) Start() error { ch, err := s.keepAlive() if err != nil { log.Fatal(err) return err } for { select { case err := <- s.stop: s.revoke() return err case <- s.client.Ctx().Done(): return errors.New("server closed") case ka, ok := <-ch: if !ok { log.Println("keep alive channel closed") s.revoke() return nil } else { log.Printf("Recv reply from service: %s, ttl:%d", s.Name, ka.TTL) } } } } func (s *Service) Stop() { s.stop <- nil } func (s *Service) keepAlive() (<-chan *clientv3.LeaseKeepAliveResponse, error){ info := &s.Info key := "services/" + s.Name value, _ := json.Marshal(info) // minimum lease TTL is 5-second resp, err := s.client.Grant(context.TODO(), 5) if err != nil { log.Fatal(err) return nil, err } _, err = s.client.Put(context.TODO(), key, string(value), clientv3.WithLease(resp.ID)) if err != nil { log.Fatal(err) return nil, err } s.leaseid = resp.ID return s.client.KeepAlive(context.TODO(), resp.ID) } func (s *Service) revoke() error { _, err := s.client.Revoke(context.TODO(), s.leaseid) if err != nil { log.Fatal(err) } log.Printf("servide:%s stop ", s.Name) return err } |
服务发现
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | package discovery import ( "context" "go.etcd.io/etcd/clientv3" "log" "time" "encoding/json" "fmt" ) type Master struct { Path string Nodes map[string] *Node Client *clientv3.Client } //node is a client type Node struct { State bool Key string Info map[uint64]*ServiceInfo } func NewMaster(endpoints []string, watchPath string) (*Master,error) { cli, err := clientv3.New(clientv3.Config{ Endpoints: endpoints, DialTimeout: time.Second, }) if err != nil { log.Fatal(err) return nil,err } master := &Master { Path: watchPath, Nodes: make(map[string]*Node), Client: cli, } go master.WatchNodes() return master,err } func (m *Master) AddNode(key string, info *ServiceInfo) { node, ok := m.Nodes[key] if !ok { node = &Node{ State: true, Key: key, Info: map[uint64]*ServiceInfo{info.ID:info}, } m.Nodes[node.Key] = node } else { node.Info[info.ID] = info } } func (m *Master) DeleteNode(key string, info *ServiceInfo) { node, ok := m.Nodes[key] if !ok { return } else { delete(node.Info, info.ID) } } func GetServiceInfo(ev *clientv3.Event) *ServiceInfo { info := &ServiceInfo{} err := json.Unmarshal([]byte(ev.Kv.Value), info) if err != nil { log.Println(err) } return info } func (m *Master) WatchNodes() { rch := m.Client.Watch(context.Background(), m.Path, clientv3.WithPrefix()) for wresp := range rch { for _, ev := range wresp.Events { switch ev.Type { case clientv3.EventTypePut: fmt.Printf("[%s] %q : %q ", ev.Type, ev.Kv.Key, ev.Kv.Value) info := GetServiceInfo(ev) m.AddNode(string(ev.Kv.Key), info) case clientv3.EventTypeDelete: fmt.Printf("[%s] %q : %q ", ev.Type, ev.Kv.Key, ev.Kv.Value) info := GetServiceInfo(ev) m.DeleteNode(string(ev.Kv.Key), info) } } } } |