基于consul实现服务注册与发现

git clone https://github.com/hashicorp/learn-consul-docker.git

cd datacenter-deploy-service-discovery

❯ docker-compose up -d
Creating network "datacenter-deploy-service-discovery_consul" with driver "bridge"
Creating consul-server ... done
Creating consul-client ... done

import "github.com/hashicorp/consul/api"

// consul 定义一个consul结构体，其内部有一个`*api.Client`字段。
type consul struct {
	client *api.Client
}

// NewConsul 连接至consul服务返回一个consul对象
func NewConsul(addr string) (*consul, error) {
	cfg := api.DefaultConfig()
	cfg.Address = addr
	c, err := api.NewClient(cfg)
	if err != nil {
		return nil, err
	}
	return &consul{c}, nil
}

// GetOutboundIP 获取本机的出口IP
func GetOutboundIP() (net.IP, error) {
	conn, err := net.Dial("udp", "8.8.8.8:80")
	if err != nil {
		return nil, err
	}
	defer conn.Close()
	localAddr := conn.LocalAddr().(*net.UDPAddr)
	return localAddr.IP, nil
}

// RegisterService 将gRPC服务注册到consul
func (c *consul) RegisterService(serviceName string, ip string, port int) error {
	srv := &api.AgentServiceRegistration{
		ID:      fmt.Sprintf("%s-%s-%d", serviceName, ip, port), // 服务唯一ID
		Name:    serviceName,                                    // 服务名称
		Tags:    []string{"q1mi", "hello"},                      // 为服务打标签
		Address: ip,
		Port:    port,
	}
	return c.client.Agent().ServiceRegister(srv)
}

import "google.golang.org/grpc/health"
import healthpb "google.golang.org/grpc/health/grpc_health_v1"

s := grpc.NewServer() // 创建gRPC服务器
healthcheck := health.NewServer()
healthpb.RegisterHealthServer(s, healthcheck)

// RegisterService 将gRPC服务注册到consul
func (c *consul) RegisterService(serviceName string, ip string, port int) error {
	// 健康检查
	check := &api.AgentServiceCheck{
		GRPC:                           fmt.Sprintf("%s:%d", ip, port), // 这里一定是外部可以访问的地址
		Timeout:                        "10s",  // 超时时间
		Interval:                       "10s",  // 运行检查的频率
		// 指定时间后自动注销不健康的服务节点
		// 最小超时时间为1分钟，收获不健康服务的进程每30秒运行一次，因此触发注销的时间可能略长于配置的超时时间。
		DeregisterCriticalServiceAfter: "1m",
	}
	srv := &api.AgentServiceRegistration{
		ID:      fmt.Sprintf("%s-%s-%d", serviceName, ip, port), // 服务唯一ID
		Name:    serviceName,                                    // 服务名称
		Tags:    []string{"q1mi", "hello"},                      // 为服务打标签
		Address: ip,
		Port:    port,
		Check:   check,
	}
	return c.client.Agent().ServiceRegister(srv)
}

// 连接consul
cc, err := api.NewClient(api.DefaultConfig())
if err != nil {
	fmt.Printf("api.NewClient failed, err:%v\n", err)
	return
}
// 返回的是一个 map[string]*api.AgentService
// 其中key是服务ID，值是注册的服务信息
serviceMap, err := cc.Agent().ServicesWithFilter("Service==`hello`")
if err != nil {
	fmt.Printf("query service from consul failed, err:%v\n", err)
	return
}
// 选一个服务机（这里选最后一个）
var addr string
for k, v := range serviceMap {
	fmt.Printf("%s:%#v\n", k, v)
	addr = v.Address + ":" + strconv.Itoa(v.Port)
}

// 建立RPC连接
conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
	log.Fatalf("grpc.Dial failed,err:%v", err)
	return
}
defer conn.Close()

import _ "github.com/mbobakov/grpc-consul-resolver"

conn, err := grpc.Dial(
	// consul服务
	"consul://localhost:8500/hello?healthy=true",
	grpc.WithTransportCredentials(insecure.NewCredentials()),
)
defer conn.Close()

c := pb.NewGreeterClient(conn)
resp, err := c.SayHello(ctx, &pb.HelloRequest{Name: *name})
if err != nil {
	fmt.Printf("c.SayHello failed, err:%v\n", err)
	return
}
fmt.Printf("resp:%v\n", resp.GetReply())

// Deregister 注销服务
func (c *consul) Deregister(serviceID string) error {
	return c.client.Agent().ServiceDeregister(serviceID)
}

lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
if err != nil {
	fmt.Printf("failed to listen: %v", err)
	return
}
s := grpc.NewServer() // 创建gRPC服务器
// 开启健康检查
healthcheck := health.NewServer()
healthpb.RegisterHealthServer(s, healthcheck)
pb.RegisterGreeterServer(s, &server{}) // 在gRPC服务端注册服务
// 注册服务
consul, err := NewConsul("127.0.0.1:8500")
if err != nil {
	log.Fatalf("NewConsul failed, err:%v\n", err)
}
ipObj, _ := GetOutboundIP()
ip := ipObj.String()
serviceId := fmt.Sprintf("%s-%s-%d", "hello", ip, port)
consul.RegisterService(serviceId, ip, port)

// 启动服务
go func() {
	err = s.Serve(lis)
	if err != nil {
		log.Printf("failed to serve: %v", err)
		return
	}
}()

quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGTERM, syscall.SIGINT)
<-quit
// 退出时注销服务
consul.Deregister(serviceId)

package main

import _ "github.com/mbobakov/grpc-consul-resolver"

// ...

conn, err := grpc.Dial(
		// consul服务
		"consul://127.0.0.1:8500/hello?healthy=true",
		// 指定round_robin策略
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)