七，服务注册与发现

原创李明 coding到灯火阑珊

Consul 介绍

服务注册与发现采用 consul 组件，是google开源的一个使用go语言开发的服务发现、配置管理中心服务。内置了服务注册与发现框架、分布一致性协议实现、健康检查、Key/Value存储、多数据中心方案等。

Consul使用gossip协议管理成员关系、广播消息到整个集群，他有两个gossip pool（LAN pool和WAN pool），LAN pool是同一个数据中心内部通信的，WAN pool是多个数据中心通信的，LAN pool有多个，WAN pool只有一个。

项目中应用

完整代码：

https://github.com/Justin02180218/micro-kit

配置文件

consul:  addr: "http://consul-server:8500"  interval: "10s"  timeout: "1s"  grpc:  retrymax: 3  retrytimeout: 500   name: "book-rpc-service"

在 /etc/hosts 中配置 consul-server 域名

config.go

type ConsulConfig struct {    Addr     string `json:"addr" yaml:"addr"`    Interval string `json:"interval" yaml:"interval"`    Timeout  string `json:"timeout" yaml:"timeout"`    Client   struct {        RetryMax     int `json:"retrymax" yaml:"retrymax"`        RetryTimeout int `json:"retrytimeout" yaml:"retrytimeout"`    }}
type GRPCConfig struct {    RetryMax     int    `json:"retrymax" yaml:"retrymax"`    RetryTimeout int    `json:"retrytimeout" yaml:"retrytimeout"`    Name         string `json:"name" yaml:"name"`}

pkg/registers

在 pkg 下新建目录 registers，创建 consul.go 文件：

代码如下：

// 链接 consulfunc connectConsul(consulAddr string) (client consul.Client) {    consulConfig := api.DefaultConfig()    consulConfig.Address = consulAddr    consulClient, err := api.NewClient(consulConfig)    if err != nil {        panic(err)    }    client = consul.NewClient(consulClient)    return}

// 向 consul 注册服务func InitRegister(cfg *configs.AppConfig, check api.AgentServiceCheck, logger log.Logger) (registrar sd.Registrar) {    rand.Seed(time.Now().UnixNano())    name := cfg.ServerConfig.Name    addr := utils.LocalIP()    port := cfg.ServerConfig.Port    consulAddr := cfg.ConsulConfig.Addr    client := connectConsul(consulAddr)    num := rand.Intn(100)    asr := api.AgentServiceRegistration{        ID:      name + "-" + strconv.Itoa(num),        Name:    name,        Address: addr,        Port:    port,        Tags:    []string{name},        Check:   &check,    }    registrar = consul.NewRegistrar(client, &asr, logger)    return}

// restful 服务检测func HttpCheck(port int, interval, timeout string) api.AgentServiceCheck {    return api.AgentServiceCheck{        HTTP:     "http://" + utils.LocalIP() + ":" + strconv.Itoa(port) + "/health",        Interval: interval,        Timeout:  timeout,        Notes:    "Http Health Check",    }}// gRPC 服务检测func GRPCCheck(port int, interval, timeout string) api.AgentServiceCheck {    return api.AgentServiceCheck{        GRPC:     utils.LocalIP() + ":" + strconv.Itoa(port) + "/health",        Interval: interval,        Timeout:  timeout,        Notes:    "GRPC Health Check",    }}

service层

以 library-user-service 为例，其他微服务相同。

// 增加函数HealthCheck() bool// 实现func (u *UserServiceImpl) HealthCheck() bool {    return true}

endpoint层

以 library-user-service 为例，其他微服务相同。

// 增加属性HealthEndpoint            endpoint.Endpoint
func MakeHealthEndpoint(svc service.UserService) endpoint.Endpoint {    return func(ctx context.Context, request interface{}) (response interface{}, err error) {        status := svc.HealthCheck()        return status, nil    }}

transport层

以 library-user-service 为例，其他微服务相同。

r.GET("/health", func(c *gin.Context) {    kithttp.NewServer(        endpoints.HealthEndpoint,        decodeHealthRequest,        utils.EncodeJsonResponse,    ).ServeHTTP(c.Writer, c.Request)})

main.go

以 library-user-service 为例，其他微服务相同。

// 配置日志var logger log.Logger{    logger = log.NewLogfmtLogger(os.Stderr)    logger = log.With(logger, "ts", log.DefaultTimestampUTC)    logger = log.With(logger, "caller", log.DefaultCaller)}
// consul 注册check := registers.HttpCheck(configs.Conf.ServerConfig.Port, configs.Conf.ConsulConfig.Interval, configs.Conf.ConsulConfig.Timeout)registrar := registers.InitRegister(configs.Conf, check, logger)
go func() {    registrar.Register()    errChan <- r.Run(fmt.Sprintf(":%s", strconv.Itoa(configs.Conf.ServerConfig.Port)))}()
registrar.Deregister()

启动 consul

使用 Docker 启动 consul

docker pull progrium/consul

docker run --rm -p 8400:8400 -p 8500:8500 -p 8600:53/udp -h node1 progrium/consul -server -bootstrap -ui-dir /ui

启动服务，然后在浏览器地址栏输入：http://consul-server:8500/

可以看到三个微服务都以注册到 consul 上，健康检测也都已经通过。

调用

以前 user-service 调用 book-rpc-service 的接口，使用的是gPRC Client的方式。现在服务注册的 consul 上，user-service 通过 consul 获取已经注册的 book-rpc-service 的一个实例，然后调用这个实例上的接口。

在 consul.go 文件中增加如下代码：

func GRPCClient(cfg *configs.AppConfig, makeEndpoint func(string) endpoint.Endpoint, logger log.Logger) endpoint.Endpoint {    consulAddr := cfg.ConsulConfig.Addr    retryMax := cfg.GRPCConfig.RetryMax    retryTimeout := cfg.GRPCConfig.RetryTimeout    name := cfg.GRPCConfig.Name    client := connectConsul(consulAddr)    instance := consul.NewInstancer(client, logger, name, []string{name}, true)    factory := factoryFor(makeEndpoint)    endpointer := sd.NewEndpointer(instance, factory, logger)    // 负载均衡采用轮询策略    balancer := lb.NewRoundRobin(endpointer)    retry := lb.Retry(retryMax, time.Millisecond*time.Duration(retryTimeout), balancer)    return retry}
func factoryFor(makeEndpoint func(string,) endpoint.Endpoint) sd.Factory {    return func(instance string) (endpoint.Endpoint, io.Closer, error) {        endpoint := makeEndpoint(instance)        return endpoint, nil, nil    }}

在 endpoint 目录下增加文件 book_rpc_endpoint.go，代码如下：

type BookRPCEndpoints struct {    FindBooksEndpoint endpoint.Endpoint}
func MakeFindBooksEndpoint(instance string) endpoint.Endpoint {    conn, err := grpc.Dial(instance, grpc.WithInsecure())    if err != nil {        fmt.Println(err)        return nil    }    findBooksEndpoint := grpctransport.NewClient(        conn, "book.Book", "FindBooksByUserID",        encodeGRPCFindBooksRequest,        decodeGRPCFindBooksResponse,        pbbook.BooksResponse{},    ).Endpoint()    return findBooksEndpoint}

修改 service.go

type UserServiceImpl struct {    userDao    dao.UserDao    grpcClient kitendpoint.Endpoint}
func NewUserServiceImpl(userDao dao.UserDao, grpcClient kitendpoint.Endpoint) UserService {    return &UserServiceImpl{        userDao:    userDao,        grpcClient: grpcClient,    }}
func (u *UserServiceImpl) FindBooksByUserID(ctx context.Context, id uint64) (interface{}, error) {    res, err := u.grpcClient(ctx, id)    if err != nil {        return nil, err    }    return res, nil}

修改 main.go

findBooksEndpoint := endpoint.MakeFindBooksEndpointgrpcClient := registers.GRPCClient(configs.Conf, findBooksEndpoint, tracer, logger)
userService := service.NewUserServiceImpl(userDao, grpcClient)

接口测试同《五，微服务library-book-grpc-service》

下一篇文章，我们在微服务中加入熔断、降级功能。

完整代码：

https://github.com/Justin02180218/micro-kit

更多【分布式专辑】【架构实战专辑】系列文章，请关注公众号

继续滑动看下一个