介绍
下面介绍 jupiter-0.2.7 版本中 grpc 通过 etcd 实现服务发现与注册。
服务发现与注册的实现解析
服务注册
服务注册的流程图：
etcd的服务注册代码模块在 jupiter/pkg/registry/etcdv3 中。
下面让我们来看看实际的代码
// Registry register/unregister service
// registry impl should control rpc timeout
type Registry interface {
RegisterService(context.Context, *server.ServiceInfo) error
UnregisterService(context.Context, *server.ServiceInfo) error
ListServices(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 registry
func (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 `name`
func (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 list
func (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 {
// 将更新后的服务信息发送出去，接收方是 resolver
case 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 string
reg 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 "/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 servers
for _, s := range app.servers {
s := s
eg.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
}
复制代码
服务发现也是类型的将注册中心信息注入即可。
文章系列
如果觉得《java如何通过grpc连接etcd_grpc通过 etcd 实现服务发现与注册-源码分析》对你有帮助，请点赞、收藏，并留下你的观点哦！
