兰州网站排名推广,有赞微商城怎么收费,徐州做网站建设公司,电子商务网站建设的市场分析Redis实现服务注册与服务发现源码阅读
背景
近期在看开源项目CloudWeGo中看到目前GoLang微服务框架Hertz中支持通过Redis实现服务注册与服务发现功能。便想着阅读下源码
源码阅读
gut clone了hertz-contrib后看到在一级目录下有目前各种主流的服务注册与发现的实现方案。为…Redis实现服务注册与服务发现源码阅读
背景
近期在看开源项目CloudWeGo中看到目前GoLang微服务框架Hertz中支持通过Redis实现服务注册与服务发现功能。便想着阅读下源码
源码阅读
gut clone了hertz-contrib后看到在一级目录下有目前各种主流的服务注册与发现的实现方案。为了便于学习选择阅读redis
服务注册源码分析
看到redis/example/server/main.go中有服务注册的实现示例
func main() {r : redis.NewRedisRegistry(127.0.0.1:6379)addr : 127.0.0.1:8888h : server.Default(server.WithHostPorts(addr),server.WithRegistry(r, registry.Info{ServiceName: hertz.test.demo,Addr: utils.NewNetAddr(tcp, addr),Weight: 10,Tags: nil,}),)h.GET(/ping, func(_ context.Context, ctx *app.RequestContext) {ctx.JSON(consts.StatusOK, utils.H{ping: pong})})h.Spin()
}
代码主要逻辑是实现一个简单的webservice其中用到了服务注册机制。可以看到在hertz中服务注册可以通过配置engine的形式在webservice初始化时定义其中
r : redis.NewRedisRegistry(127.0.0.1:6379)定义了一个服务注册的地址即要把这个微服务注册到哪个主机中。而server.WithRegistry()使得服务初始化时引入了这个服务注册。Info即是服务注册的相关信息 进入redis/registry.go查看服务注册的定义可以看到redis服务注册是实现的registry.Registry接口
var _ registry.Registry (*redisRegistry)(nil)type redisRegistry struct {client *redis.Clientrctx *registryContextmu sync.Mutexwg sync.WaitGroup
}type registryContext struct {ctx context.Contextcancel context.CancelFunc
}// Registry is extension interface of service registry.
type Registry interface {Register(info *Info) errorDeregister(info *Info) error
}// Info is used for registry.
// The fields are just suggested, which is used depends on design.
type Info struct {// ServiceName will be set in hertz by defaultServiceName string// Addr will be set in hertz by defaultAddr net.Addr// Weight will be set in hertz by defaultWeight int// extend other infos with Tags.Tags map[string]string
}registry.Registry通过Register(info *Info)和Deregister(info *Info)描述服务注册与服务发现 接下来看如何创建一个redis服务注册
// NewRedisRegistry creates a redis registry
func NewRedisRegistry(addr string, opts ...Option) registry.Registry {redisOpts : redis.Options{Addr: addr,Password: ,DB: 0,}for _, opt : range opts {opt(redisOpts)}rdb : redis.NewClient(redisOpts)return redisRegistry{client: rdb,}
}我们已经可以猜到了配置redis客户端连接User Server的redis用redis来存储服务映射关系实现服务注册中心那么是不是这样呢我们接着往下看服务注册的实现源码
func (r *redisRegistry) Register(info *registry.Info) error {// 校验配置信息if err : validateRegistryInfo(info); err ! nil {return err}rctx : registryContext{}rctx.ctx, rctx.cancel context.WithCancel(context.Background())m : newMentor()r.wg.Add(1)// 并发监控redisgo m.subscribe(rctx.ctx, info, r)r.wg.Wait()rdb : r.client// 将注册信息hash化hash, err : prepareRegistryHash(info)if err ! nil {return err}// 上锁r.mu.Lock()r.rctx rctx// 注册信息写入到redis即我们的服务注册中心rdb.HSet(rctx.ctx, hash.key, hash.field, hash.value)rdb.Expire(rctx.ctx, hash.key, defaultExpireTime)// 生成服务相关信息和发送rdb.Publish(rctx.ctx, hash.key, generateMsg(register, info.ServiceName, info.Addr.String()))// 写完解锁r.mu.Unlock()go m.monitorTTL(rctx.ctx, hash, info, r)// 保持长连接go keepAlive(rctx.ctx, hash, r)return nil
}Register方法已经对服务注册的主要流程进行了描述下面来看一些细节
func validateRegistryInfo(info *registry.Info) error {if info nil {return fmt.Errorf(registry.Info can not be empty)}if info.ServiceName {return fmt.Errorf(registry.Info ServiceName can not be empty)}if info.Addr nil {return fmt.Errorf(registry.Info Addr can not be empty)}return nil
}校验服务注册时并不会对客户端是否连接上进行校验只会校验参数和结构体是否为空
func prepareRegistryHash(info *registry.Info) (*registryHash, error) {meta, err : json.Marshal(convertInfo(info))if err ! nil {return nil, err}return registryHash{key: generateKey(info.ServiceName, server),field: info.Addr.String(),value: string(meta),}, nil
}
服务注册信息hash即生成key-velue方便写入到redis中
func keepAlive(ctx context.Context, hash *registryHash, r *redisRegistry) {ticker : time.NewTicker(defaultTickerTime)defer ticker.Stop()for {select {case -ticker.C:r.client.Expire(ctx, hash.key, defaultKeepAliveTime)case -ctx.Done():break}}
}
最后再起一个协程在生命期内监听保持长连接这里用到的是多路复用
func keepAlive(ctx context.Context, hash *registryHash, r *redisRegistry) {ticker : time.NewTicker(defaultTickerTime)defer ticker.Stop()for {select {case -ticker.C:r.client.Expire(ctx, hash.key, defaultKeepAliveTime)case -ctx.Done():break}}
}
再来看服务注册退出
func (r *redisRegistry) Deregister(info *registry.Info) error {if err : validateRegistryInfo(info); err ! nil {return err}rctx : r.rctxrdb : r.clienthash, err : prepareRegistryHash(info)if err ! nil {return err}r.mu.Lock()// 删除redis中的注册信息rdb.HDel(rctx.ctx, hash.key, hash.field)rdb.Publish(rctx.ctx, hash.key, generateMsg(deregister, info.ServiceName, info.Addr.String()))rctx.cancel()r.mu.Unlock()return nil
}
整体逻辑和服务注册相似只是最后把注册信息删掉
服务发现源码分析
看到redis/example/client/main.go中有服务注册的实现示例
func main() {cli, err : client.NewClient()if err ! nil {panic(err)}r : redis.NewRedisResolver(127.0.0.1:6379)cli.Use(sd.Discovery(r))for i : 0; i 10; i {status, body, err : cli.Get(context.Background(), nil, http://hertz.test.demo/ping, config.WithSD(true))if err ! nil {hlog.Fatal(err)}hlog.Infof(HERTZ: code%d,body%s, status, string(body))}
}
config.WithSD(true)即通过中间件形式使得客户端发送请求时并非直接请求服务器而是请求注册中心通过服务发现再进一步转到服务器上 接前文中在redis里进行了服务注册这里客户端想要进行服务发现找到自己请求的微服务。这里服务发现还是通过复用接口实现的
var _ discovery.Resolver (*redisResolver)(nil)type redisResolver struct {client *redis.Client
}// NewRedisResolver creates a redis resolver
func NewRedisResolver(addr string, opts ...Option) discovery.Resolver {redisOpts : redis.Options{Addr: addr}for _, opt : range opts {opt(redisOpts)}rdb : redis.NewClient(redisOpts)return redisResolver{client: rdb,}
}服务发现开始和服务注册一样需要先连接上redis
func (r *redisResolver) Target(_ context.Context, target *discovery.TargetInfo) string {return target.Host
}func (r *redisResolver) Resolve(ctx context.Context, desc string) (discovery.Result, error) {rdb : r.client// 查询服务列表fvs : rdb.HGetAll(ctx, generateKey(desc, server)).Val()var its []discovery.Instancefor f, v : range fvs {// 反序列化获取服务信息var ri registryInfoerr : json.Unmarshal([]byte(v), ri)if err ! nil {hlog.Warnf(HERTZ: fail to unmarshal with err: %v, ignore instance Addr: %v, err, f)continue}// 负载均衡参数weight : ri.Weightif weight 0 {weight defaultWeight}its append(its, discovery.NewInstance(tcp, ri.Addr, weight, ri.Tags))}return discovery.Result{// 服务发现的结果CacheKey: desc,//redis表中的keyInstances: its,//服务表}, nil
}func (r *redisResolver) Name() string {return Redis
}
Target、Name、Resolve即为实现自方法的接口其中target和Name分别解出redis的地址和NameResolve方法用来在Redis中发现服务 我们还可以细扣一下服务发现中间件进一步是怎么实现的 /pkg/mod/github.com/cloudwego/hertzv0.6.0/pkg/common/config/request_option.go:58中WithSD如下
// WithSD set isSD in RequestOptions.
func WithSD(b bool) RequestOption {return RequestOption{F: func(o *RequestOptions) {o.isSD b}}
}可见这里是用来高速请求这个请求是有服务发现机制的。循着client.Get()方法一路往下找这项配置写入到了req中
func GetURL(ctx context.Context, dst []byte, url string, c Doer, requestOptions ...config.RequestOption) (statusCode int, body []byte, err error) {req : protocol.AcquireRequest()req.SetOptions(requestOptions...)statusCode, body, err doRequestFollowRedirectsBuffer(ctx, req, dst, url, c)protocol.ReleaseRequest(req)return statusCode, body, err
}在hertz中的Request定义中其实是包含有config定义里面就有sd的flag
type Request struct {noCopy nocopy.NoCopy //lint:ignore U1000 until noCopy is usedHeader RequestHeaderuri URIpostArgs ArgsbodyStream io.Readerw requestBodyWriterbody *bytebufferpool.ByteBufferbodyRaw []bytemaxKeepBodySize intmultipartForm *multipart.FormmultipartFormBoundary string// Group bool members in order to reduce Request object size.parsedURI boolparsedPostArgs boolisTLS boolmultipartFiles []*FilemultipartFields []*MultipartField// Request level options, service discovery options etc.options *config.RequestOptions
}
也就是会从req中解析出服务地址
