项目中使用etcd实现了grpc的服务户注册和服务发现,这里来看下如何实现的服务注册和服务发现node
先来看下使用的demo,demo中的代码discoverygit
package discovery import ( "context" "encoding/json" "errors" "net/http" "strconv" "strings" "time" clientv3 "go.etcd.io/etcd/client/v3" "go.uber.org/zap" ) // Register for grpc server type Register struct { EtcdAddrs []string DialTimeout int closeCh chan struct{} leasesID clientv3.LeaseID keepAliveCh <-chan *clientv3.LeaseKeepAliveResponse srvInfo Server srvTTL int64 cli *clientv3.Client logger *zap.Logger } // NewRegister create a register base on etcd func NewRegister(etcdAddrs []string, logger *zap.Logger) *Register { return &Register{ EtcdAddrs: etcdAddrs, DialTimeout: 3, logger: logger, } } // Register a service func (r *Register) Register(srvInfo Server, ttl int64) (chan<- struct{}, error) { var err error if strings.Split(srvInfo.Addr, ":")[0] == "" { return nil, errors.New("invalid ip") } if r.cli, err = clientv3.New(clientv3.Config{ Endpoints: r.EtcdAddrs, DialTimeout: time.Duration(r.DialTimeout) * time.Second, }); err != nil { return nil, err } r.srvInfo = srvInfo r.srvTTL = ttl if err = r.register(); err != nil { return nil, err } r.closeCh = make(chan struct{}) go r.keepAlive() return r.closeCh, nil } // Stop stop register func (r *Register) Stop() { r.closeCh <- struct{}{} } // register 注册节点 func (r *Register) register() error { leaseCtx, cancel := context.WithTimeout(context.Background(), time.Duration(r.DialTimeout)*time.Second) defer cancel() leaseResp, err := r.cli.Grant(leaseCtx, r.srvTTL) if err != nil { return err } r.leasesID = leaseResp.ID if r.keepAliveCh, err = r.cli.KeepAlive(context.Background(), leaseResp.ID); err != nil { return err } data, err := json.Marshal(r.srvInfo) if err != nil { return err } _, err = r.cli.Put(context.Background(), BuildRegPath(r.srvInfo), string(data), clientv3.WithLease(r.leasesID)) return err } // unregister 删除节点 func (r *Register) unregister() error { _, err := r.cli.Delete(context.Background(), BuildRegPath(r.srvInfo)) return err } // keepAlive func (r *Register) keepAlive() { ticker := time.NewTicker(time.Duration(r.srvTTL) * time.Second) for { select { case <-r.closeCh: if err := r.unregister(); err != nil { r.logger.Error("unregister failed", zap.Error(err)) } if _, err := r.cli.Revoke(context.Background(), r.leasesID); err != nil { r.logger.Error("revoke failed", zap.Error(err)) } return case res := <-r.keepAliveCh: if res == nil { if err := r.register(); err != nil { r.logger.Error("register failed", zap.Error(err)) } } case <-ticker.C: if r.keepAliveCh == nil { if err := r.register(); err != nil { r.logger.Error("register failed", zap.Error(err)) } } } } } // UpdateHandler return http handler func (r *Register) UpdateHandler() http.HandlerFunc { return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { wi := req.URL.Query().Get("weight") weight, err := strconv.Atoi(wi) if err != nil { w.WriteHeader(http.StatusBadRequest) w.Write([]byte(err.Error())) return } var update = func() error { r.srvInfo.Weight = int64(weight) data, err := json.Marshal(r.srvInfo) if err != nil { return err } _, err = r.cli.Put(context.Background(), BuildRegPath(r.srvInfo), string(data), clientv3.WithLease(r.leasesID)) return err } if err := update(); err != nil { w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(err.Error())) return } w.Write([]byte("update server weight success")) }) } func (r *Register) GetServerInfo() (Server, error) { resp, err := r.cli.Get(context.Background(), BuildRegPath(r.srvInfo)) if err != nil { return r.srvInfo, err } info := Server{} if resp.Count >= 1 { if err := json.Unmarshal(resp.Kvs[0].Value, &info); err != nil { return info, err } } return info, nil }
来分析下上面的代码实现github
当启动一个grpc的时候咱们注册到etcd中golang
etcdRegister := discovery.NewRegister(config.Etcd.Addrs, log.Logger) node := discovery.Server{ Name: app, Addr: utils.InternalIP() + config.Port.GRPC, } if _, err := etcdRegister.Register(node, 10); err != nil { panic(fmt.Sprintf("server register failed: %v", err)) }
调用服务注册的时候首先分配了一个租约json
func (l *lessor) Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error) { r := &pb.LeaseGrantRequest{TTL: ttl} resp, err := l.remote.LeaseGrant(ctx, r, l.callOpts...) if err == nil { gresp := &LeaseGrantResponse{ ResponseHeader: resp.GetHeader(), ID: LeaseID(resp.ID), TTL: resp.TTL, Error: resp.Error, } return gresp, nil } return nil, toErr(ctx, err) }
而后经过KeepAlive保活api
// KeepAlive尝试保持给定的租约永久alive func (l *lessor) KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error) { ch := make(chan *LeaseKeepAliveResponse, LeaseResponseChSize) l.mu.Lock() // ensure that recvKeepAliveLoop is still running select { case <-l.donec: err := l.loopErr l.mu.Unlock() close(ch) return ch, ErrKeepAliveHalted{Reason: err} default: } ka, ok := l.keepAlives[id] if !ok { // create fresh keep alive ka = &keepAlive{ chs: []chan<- *LeaseKeepAliveResponse{ch}, ctxs: []context.Context{ctx}, deadline: time.Now().Add(l.firstKeepAliveTimeout), nextKeepAlive: time.Now(), donec: make(chan struct{}), } l.keepAlives[id] = ka } else { // add channel and context to existing keep alive ka.ctxs = append(ka.ctxs, ctx) ka.chs = append(ka.chs, ch) } l.mu.Unlock() go l.keepAliveCtxCloser(ctx, id, ka.donec) // 使用once只在第一次调用 l.firstKeepAliveOnce.Do(func() { // 500毫秒一次,不断的发送保持活动请求 go l.recvKeepAliveLoop() // 删除等待过久没反馈的租约 go l.deadlineLoop() }) return ch, nil } // deadlineLoop获取在租约TTL中没有收到响应的任何保持活动的通道 func (l *lessor) deadlineLoop() { for { select { case <-time.After(time.Second): // donec 关闭,当 recvKeepAliveLoop 中止时设置 loopErr case <-l.donec: return } now := time.Now() l.mu.Lock() for id, ka := range l.keepAlives { if ka.deadline.Before(now) { // 等待响应过久;租约可能已过时 ka.close() delete(l.keepAlives, id) } } l.mu.Unlock() } } func (l *lessor) recvKeepAliveLoop() (gerr error) { defer func() { l.mu.Lock() close(l.donec) l.loopErr = gerr for _, ka := range l.keepAlives { ka.close() } l.keepAlives = make(map[LeaseID]*keepAlive) l.mu.Unlock() }() for { // resetRecv 打开一个新的lease stream并开始发送保持活动请求。 stream, err := l.resetRecv() if err != nil { if canceledByCaller(l.stopCtx, err) { return err } } else { for { // 接收lease stream的返回返回 resp, err := stream.Recv() if err != nil { if canceledByCaller(l.stopCtx, err) { return err } if toErr(l.stopCtx, err) == rpctypes.ErrNoLeader { l.closeRequireLeader() } break } // 根据LeaseKeepAliveResponse更新租约 // 若是租约过时删除全部alive channels l.recvKeepAlive(resp) } } select { case <-time.After(retryConnWait): continue case <-l.stopCtx.Done(): return l.stopCtx.Err() } } } // resetRecv 打开一个新的lease stream并开始发送保持活动请求。 func (l *lessor) resetRecv() (pb.Lease_LeaseKeepAliveClient, error) { sctx, cancel := context.WithCancel(l.stopCtx) // 创建服务端和客户端链接的lease stream stream, err := l.remote.LeaseKeepAlive(sctx, l.callOpts...) if err != nil { cancel() return nil, err } l.mu.Lock() defer l.mu.Unlock() if l.stream != nil && l.streamCancel != nil { l.streamCancel() } l.streamCancel = cancel l.stream = stream go l.sendKeepAliveLoop(stream) return stream, nil } // sendKeepAliveLoop 在给定流的生命周期内发送保持活动请求 func (l *lessor) sendKeepAliveLoop(stream pb.Lease_LeaseKeepAliveClient) { for { var tosend []LeaseID now := time.Now() l.mu.Lock() for id, ka := range l.keepAlives { if ka.nextKeepAlive.Before(now) { tosend = append(tosend, id) } } l.mu.Unlock() for _, id := range tosend { r := &pb.LeaseKeepAliveRequest{ID: int64(id)} if err := stream.Send(r); err != nil { // TODO do something with this error? return } } select { // 每500毫秒执行一次 case <-time.After(500 * time.Millisecond): case <-stream.Context().Done(): return case <-l.donec: return case <-l.stopCtx.Done(): return } } } // 撤销给定的租约,全部附加到租约的key将过时并被删除 func (l *lessor) Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse, error) { r := &pb.LeaseRevokeRequest{ID: int64(id)} resp, err := l.remote.LeaseRevoke(ctx, r, l.callOpts...) if err == nil { return (*LeaseRevokeResponse)(resp), nil } return nil, toErr(ctx, err) }
总结:缓存
一、每次注册一个服务的分配一个租约;服务器
二、KeepAlive经过从客户端到服务器端的流化的keep alive
请求和从服务器端到客户端的流化的keep alive
应答来维持租约;mvc
三、KeepAlive会500毫秒进行一次lease stream的发送;app
四、而后接收到KeepAlive发送信息回执,处理更新租约,服务处于活动状态;
五、若是在租约TTL中没有收到响应的任何保持活动的请求,删除租约;
六、Revoke撤销一个租约,全部附加到租约的key将过时并被删除。
咱们只需实现grpc在resolver中提供了Builder和Resolver接口,就能完成gRPC客户端的服务发现和负载均衡
// 建立一个resolver用于监视名称解析更新 type Builder interface { Build(target Target, cc ClientConn, opts BuildOption) (Resolver, error) Scheme() string }
Build方法:为给定目标建立一个新的resolver,当调用grpc.Dial()时执行;
Scheme方法:返回此resolver支持的方案,可参考Scheme定义
// 监视指定目标的更新,包括地址更新和服务配置更新 type Resolver interface { ResolveNow(ResolveNowOption) Close() }
ResolveNow方法:被 gRPC 调用,以尝试再次解析目标名称。只用于提示,可忽略该方法;
Close方法:关闭resolver。
接下来看下具体的实现
package discovery import ( "context" "time" "go.uber.org/zap" "go.etcd.io/etcd/api/v3/mvccpb" clientv3 "go.etcd.io/etcd/client/v3" "google.golang.org/grpc/resolver" ) const ( schema = "etcd" ) // Resolver for grpc client type Resolver struct { schema string EtcdAddrs []string DialTimeout int closeCh chan struct{} watchCh clientv3.WatchChan cli *clientv3.Client keyPrifix string srvAddrsList []resolver.Address cc resolver.ClientConn logger *zap.Logger } // NewResolver create a new resolver.Builder base on etcd func NewResolver(etcdAddrs []string, logger *zap.Logger) *Resolver { return &Resolver{ schema: schema, EtcdAddrs: etcdAddrs, DialTimeout: 3, logger: logger, } } // Scheme returns the scheme supported by this resolver. func (r *Resolver) Scheme() string { return r.schema } // Build creates a new resolver.Resolver for the given target func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) { r.cc = cc r.keyPrifix = BuildPrefix(Server{Name: target.Endpoint, Version: target.Authority}) if _, err := r.start(); err != nil { return nil, err } return r, nil } // ResolveNow resolver.Resolver interface func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {} // Close resolver.Resolver interface func (r *Resolver) Close() { r.closeCh <- struct{}{} } // start func (r *Resolver) start() (chan<- struct{}, error) { var err error r.cli, err = clientv3.New(clientv3.Config{ Endpoints: r.EtcdAddrs, DialTimeout: time.Duration(r.DialTimeout) * time.Second, }) if err != nil { return nil, err } resolver.Register(r) r.closeCh = make(chan struct{}) if err = r.sync(); err != nil { return nil, err } go r.watch() return r.closeCh, nil } // watch update events func (r *Resolver) watch() { ticker := time.NewTicker(time.Minute) r.watchCh = r.cli.Watch(context.Background(), r.keyPrifix, clientv3.WithPrefix()) for { select { case <-r.closeCh: return case res, ok := <-r.watchCh: if ok { r.update(res.Events) } case <-ticker.C: if err := r.sync(); err != nil { r.logger.Error("sync failed", zap.Error(err)) } } } } // update func (r *Resolver) update(events []*clientv3.Event) { for _, ev := range events { var info Server var err error switch ev.Type { case mvccpb.PUT: info, err = ParseValue(ev.Kv.Value) if err != nil { continue } addr := resolver.Address{Addr: info.Addr, Metadata: info.Weight} if !Exist(r.srvAddrsList, addr) { r.srvAddrsList = append(r.srvAddrsList, addr) r.cc.UpdateState(resolver.State{Addresses: r.srvAddrsList}) } case mvccpb.DELETE: info, err = SplitPath(string(ev.Kv.Key)) if err != nil { continue } addr := resolver.Address{Addr: info.Addr} if s, ok := Remove(r.srvAddrsList, addr); ok { r.srvAddrsList = s r.cc.UpdateState(resolver.State{Addresses: r.srvAddrsList}) } } } } // sync 同步获取全部地址信息 func (r *Resolver) sync() error { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() res, err := r.cli.Get(ctx, r.keyPrifix, clientv3.WithPrefix()) if err != nil { return err } r.srvAddrsList = []resolver.Address{} for _, v := range res.Kvs { info, err := ParseValue(v.Value) if err != nil { continue } addr := resolver.Address{Addr: info.Addr, Metadata: info.Weight} r.srvAddrsList = append(r.srvAddrsList, addr) } r.cc.UpdateState(resolver.State{Addresses: r.srvAddrsList}) return nil }
总结:
一、watch会监听前缀的信息变动,有变动的通知,及时更新srvAddrsList的地址信息;
二、sync会定时的同步etcd中的可用的服务地址到srvAddrsList中;
三、使用UpdateState更新ClientConn的Addresses;
四、而后grpc客户端就能根据配置的具体策略发送请求到grpc的server中。
这里使用gRPC内置的负载均衡策略round_robin
,根据负载均衡地址,以轮询的方式进行调用服务,来测试下服务的发现和简单的服务负载
package discovery import ( "context" "fmt" "log" "net" "testing" "time" "go.uber.org/zap" "google.golang.org/grpc/balancer/roundrobin" "google.golang.org/grpc/resolver" "etcd-learning/discovery/helloworld" "google.golang.org/grpc" ) var etcdAddrs = []string{"127.0.0.1:2379"} func TestResolver(t *testing.T) { r := NewResolver(etcdAddrs, zap.NewNop()) resolver.Register(r) // etcd中注册5个服务 go newServer(t, ":1001", "1.0.0", 1) go newServer(t, ":1002", "1.0.0", 1) go newServer(t, ":1003", "1.0.0", 1) go newServer(t, ":1004", "1.0.0", 1) go newServer(t, ":1006", "1.0.0", 10) conn, err := grpc.Dial("etcd:///hello", grpc.WithInsecure(), grpc.WithBalancerName(roundrobin.Name)) if err != nil { t.Fatalf("failed to dial %v", err) } defer conn.Close() c := helloworld.NewGreeterClient(conn) // 进行十次数据请求 for i := 0; i < 10; i++ { resp, err := c.SayHello(context.Background(), &helloworld.HelloRequest{Name: "abc"}) if err != nil { t.Fatalf("say hello failed %v", err) } log.Println(resp.Message) time.Sleep(100 * time.Millisecond) } time.Sleep(10 * time.Second) } type server struct { Port string } // SayHello implements helloworld.GreeterServer func (s *server) SayHello(ctx context.Context, in *helloworld.HelloRequest) (*helloworld.HelloReply, error) { return &helloworld.HelloReply{Message: fmt.Sprintf("Hello From %s", s.Port)}, nil } func newServer(t *testing.T, port string, version string, weight int64) { register := NewRegister(etcdAddrs, zap.NewNop()) defer register.Stop() listen, err := net.Listen("tcp", port) if err != nil { log.Fatalf("failed to listen %v", err) } s := grpc.NewServer() helloworld.RegisterGreeterServer(s, &server{Port: port}) info := Server{ Name: "hello", Addr: fmt.Sprintf("127.0.0.1%s", port), Version: version, Weight: weight, } register.Register(info, 10) if err := s.Serve(listen); err != nil { log.Fatalf("failed to server %v", err) } }
这里注册了5个服务,端口号是1001到1006,循环调用10次
=== RUN TestResolver 2021/07/24 22:44:52 Hello From :1001 2021/07/24 22:44:52 Hello From :1006 2021/07/24 22:44:53 Hello From :1001 2021/07/24 22:44:53 Hello From :1002 2021/07/24 22:44:53 Hello From :1003 2021/07/24 22:44:53 Hello From :1004 2021/07/24 22:44:53 Hello From :1006 2021/07/24 22:44:53 Hello From :1001 2021/07/24 22:44:53 Hello From :1002 2021/07/24 22:44:53 Hello From :1003
发现每次的请求会发送到不一样的服务中
在服务消费者和服务提供者之间有一个独立的LB,一般是专门的硬件设备如 F5,或者基于软件如LVS
,HAproxy
等实现。LB上有全部服务的地址映射表,一般由运维配置注册,当服务消费方调用某个目标服务时,它向LB发起请求,由LB以某种策略,好比轮询(Round-Robin)
作负载均衡后将请求转发到目标服务。LB通常具有健康检查能力,能自动摘除不健康的服务实例。
该方案主要问题:
一、单点问题,全部服务调用流量都通过LB,当服务数量和调用量大的时候,LB容易成为瓶颈,且一旦LB发生故障影响整个系统;
二、服务消费方、提供方之间增长了一级,有必定性能开销。
针对第一个方案的不足,此方案将LB的功能集成到服务消费方进程里,也被称为软负载或者客户端负载方案。服务提供方启动时,首先将服务地址注册到服务注册表,同时按期报心跳到服务注册表以代表服务的存活状态,至关于健康检查,服务消费方要访问某个服务时,它经过内置的LB组件向服务注册表查询,同时缓存并按期刷新目标服务地址列表,而后以某种负载均衡策略选择一个目标服务地址,最后向目标服务发起请求。LB和服务发现能力被分散到每个服务消费者的进程内部,同时服务消费方和服务提供方之间是直接调用,没有额外开销,性能比较好。
该方案主要问题:
一、开发成本,该方案将服务调用方集成到客户端的进程里头,若是有多种不一样的语言栈,就要配合开发多种不一样的客户端,有必定的研发和维护成本;
二、另外生产环境中,后续若是要对客户库进行升级,势必要求服务调用方修改代码并从新发布,升级较复杂。
该方案是针对第二种方案的不足而提出的一种折中方案,原理和第二种方案基本相似。
不一样之处是将LB和服务发现功能从进程内移出来,变成主机上的一个独立进程。主机上的一个或者多个服务要访问目标服务时,他们都经过同一主机上的独立LB进程作服务发现和负载均衡。该方案也是一种分布式方案没有单点问题,一个LB进程挂了只影响该主机上的服务调用方,服务调用方和LB之间是进程内调用性能好,同时该方案还简化了服务调用方,不须要为不一样语言开发客户库,LB的升级不须要服务调用方改代码。
该方案主要问题:部署较复杂,环节多,出错调试排查问题不方便。
上面经过etcd实现服务发现,使用的及时第二种 进程内LB(Balancing-aware Client)。
【Load Balancing in gRPC】https://github.com/grpc/grpc/blob/master/doc/load-balancing.md
【文中的代码示例】https://github.com/boilingfrog/etcd-learning/tree/main/discovery