兄弟网站制作,qq刷赞网站怎么做的,北京网站设计联系方式,h5模版网站介绍下面介绍 jupiter-0.2.7 版本中 grpc 通过 etcd 实现服务发现与注册。服务发现与注册的实现解析服务注册服务注册的流程图#xff1a;etcd的服务注册代码模块在 jupiter/pkg/registry/etcdv3 中。下面让我们来看看实际的代码// Registry register/unregister service// re…介绍下面介绍 jupiter-0.2.7 版本中 grpc 通过 etcd 实现服务发现与注册。服务发现与注册的实现解析服务注册服务注册的流程图etcd的服务注册代码模块在 jupiter/pkg/registry/etcdv3 中。下面让我们来看看实际的代码// Registry register/unregister service// registry impl should control rpc timeouttype Registry interface {RegisterService(context.Context, *server.ServiceInfo) errorUnregisterService(context.Context, *server.ServiceInfo) errorListServices(context.Context, string, string) ([]*server.ServiceInfo, error)WatchServices(context.Context, string, string) (chan Endpoints, error)io.Closer}复制代码在 pkg/registry/registry.go 中定义了注册服务对象的接口。不同的服务只要实现了这些接口jupiter 就能使用。首先我们来看看注册方法// RegisterService register service to registryfunc (reg *etcdv3Registry) RegisterService(ctx context.Context, info *server.ServiceInfo) error {err : reg.registerBiz(ctx, info)...}// 业务信息注册func (reg *etcdv3Registry) registerBiz(ctx context.Context, info *server.ServiceInfo) error {...// 提交信息到 etcd_, err : reg.client.Put(readCtx, key, val, opOptions...)...}复制代码这里主要的部分是 reg.client.Put() 将服务信息提交到 etcd 中。其中的租约机制我会在之后单独写一篇文章介绍。这里主要还是关注如何注册。源码中还有个 registerMetric() 方法这个方法的目的是将服务信息在提交到etcd的 prometheus 前缀目录下用于服务监控用的也是 client.Put() 方法。这里具体就不展示代码了感兴趣的同学可以去源码库中查看。服务退出// 删除服务func (reg *etcdv3Registry) unregister(ctx context.Context, key string) error {...// 删除服务信息_, err : reg.client.Delete(ctx, key)...}复制代码这里通过 client.Delete() 方法将服务信息从 etcd 中删除掉。获取服务列表// ListServices list service registered in registry with name namefunc (reg *etcdv3Registry) ListServices(ctx context.Context, name string, scheme string) (services []*server.ServiceInfo, err error) {// 服务信息key的前缀target : fmt.Sprintf(/%s/%s/providers/%s://, reg.Prefix, name, scheme)// 获取相关前缀的所有信息getResp, getErr : reg.client.Get(ctx, target, clientv3.WithPrefix())...}复制代码通过 client.Get() 方法获取到相同前缀的服务信息。服务信息变动监控// WatchServices watch service change event, then return address listfunc (reg *etcdv3Registry) WatchServices(ctx context.Context, name string, scheme string) (chan registry.Endpoints, error) {prefix : fmt.Sprintf(/%s/%s/, reg.Prefix, name)// 通过etcd客户端创建一个监控通道watch, err : reg.client.WatchPrefix(context.Background(), prefix)if err ! nil {return nil, err}...xgo.Go(func() {// 不断接收etcd发送过来的变动事件for event : range watch.C() {switch event.Type {case mvccpb.PUT:updateAddrList(al, prefix, scheme, event.Kv)case mvccpb.DELETE:deleteAddrList(al, prefix, scheme, event.Kv)}out : al.DeepCopy()fmt.Printf(al %p\n, al.Nodes)fmt.Printf(snapshot %p\n, out.Nodes)select {// 将更新后的服务信息发送出去接收方是 resolvercase addresses default:xlog.Warnf(invalid)}}})// 返回一个地址通道用于传递return addresses, nil}复制代码WatchServices() 方法主要是监控信息的变动事件并将变动后的服务信息重新返回给 resolver。具体思路是通过 etcdClient.Watch() 方法创建一个监控通道然后放入一个 goroutine来不断接收 etcd 推送过来的事件维护本地的服务信息并通过 resolver 最终返回到 grpclb 负载均衡器进行服务地址信息的更新。服务发现服务发现流程图grpc 的 resolver 模块定义了两个接口// Builder creates a resolver that will be used to watch name resolution updates.type Builder interface {Build(target Target, cc ClientConn, opts BuildOptions) (Resolver, error)Scheme() string}// Resolver watches for the updates on the specified target.// Updates include address updates and service config updates.type Resolver interface {ResolveNow(ResolveNowOptions)Close()}复制代码首先我们来看看 Builder 接口的具体实现type baseBuilder struct {name stringreg registry.Registry}// Build ...func (b *baseBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {endpoints, err : b.reg.WatchServices(context.Background(), target.Endpoint, grpc)if err ! nil {return nil, err}var stop make(chan struct{})xgo.Go(func() {for {select {case endpoint : var state resolver.State{Addresses: make([]resolver.Address, 0),...}for _, node : range endpoint.Nodes {...state.Addresses append(state.Addresses, address)}cc.UpdateState(state)case return}}})return baseResolver{stop: stop,}, nil}复制代码这里Build 方法主要是通过 Registry 模块获得监控服务通道然后将更新的服务信息再更新到 grpcClient 中去保证 grpcClient 的负载均衡器的服务地址永远都是最新的。如何将Builder的具体实现注册到 grpc 中import google.golang.org/grpc/resolver// Register ...func Register(name string, reg registry.Registry) {resolver.Register(baseBuilder{name: name,reg: reg,})}复制代码将 Registry模块注入到 Builder 对象中然后注入到 grpc 的 resolver 模块中去。这样 grpcClient 在实际运行中就会调用 etcd 的服务发现功能了。grpc 如何使用服务与发现的源码解析这里在介绍一下jupiter框架在实际项目中如何使用服务发现与注册。服务注册func (app *Application) startServers() error {var eg errgroup.Group// start multi serversfor _, s : range app.servers {s : seg.Go(func() (err error) {_ app.registerer.RegisterService(context.TODO(), s.Info())defer app.registerer.UnregisterService(context.TODO(), s.Info())...})}return eg.Wait()}eng : engine.NewEngine()eng.SetRegistry(compound_registry.New(etcdv3_registry.StdConfig(default).Build(),))复制代码在框架的 Application 模块中已经实现了服务的自动注册与删除。一般使用框架时不需要再调用。项目使用中只需要在创建 Application 对象时将注册中心信息注入即可。服务发现// 服务发现需要初始化拿到etcd中服务的信息func (eng *Engine) initResolver() error {resolver.Register(etcd, etcdv3.StdConfig(default).Build())return nil}复制代码服务发现也是类型的将注册中心信息注入即可。文章系列
