golang grpc之etcd服务注册发现

什么是etcd？什么是grpc？为什么要使用etcd。本文将简单对etcd与grpc介绍与代码实现。

etcd

etcd是一个高可用的键值分布式存储系统，主要用于共享配置和服务发现。etcd使用Go语言编写，并通过Raft一致性算法处理日志复制以保证强一致性。Raft通过选举的方式来实现一致性，在Raft中，任何一个节点都可能成为Leader。k8s也使用了etcd。

• Raft算法：

1. Leader领导者: 处理所有客户端交互，日志复制等，一般一次只有一个Leader.
2. Follower信徒: 类似选民，完全被动
3. Candidate候选人: 可以被选为一个新的领导人。

• docker-compose安装etcd v3
  官方elcolio/etcd镜像是v2版本，所以这里使用的是bitnami/etcd镜像，
  docker-compose.yaml

version: '3' 

services:
  etcd1:
    image: bitnami/etcd
    container_name: etcd1
    ports:
      - 2379:2379
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd1
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd1:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd1:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
  etcd2:
    image: bitnami/etcd
    container_name: etcd2
    ports:
      - 22379:2379
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd2
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd2:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd2:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new
  etcd3:
    image: bitnami/etcd
    container_name: etcd3
    ports:
      - 32379:2379
    environment:
      - ALLOW_NONE_AUTHENTICATION=yes
      - ETCD_NAME=etcd3
      - ETCD_INITIAL_ADVERTISE_PEER_URLS=http://etcd3:2380
      - ETCD_LISTEN_PEER_URLS=http://0.0.0.0:2380
      - ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
      - ETCD_ADVERTISE_CLIENT_URLS=http://etcd3:2379
      - ETCD_INITIAL_CLUSTER_TOKEN=etcd-cluster
      - ETCD_INITIAL_CLUSTER=etcd1=http://etcd1:2380,etcd2=http://etcd2:2380,etcd3=http://etcd3:2380
      - ETCD_INITIAL_CLUSTER_STATE=new

由于etcd没有自带管理界面，可以使用e3w。

• go的etcd v3安装包

go get -u -v go.etcd.io/etcd/clientv3

• etcd服务注册

服务注册：主要思路是创建一个lease租约，put一个前缀的Key(方便服务发现时根据前缀取key对应的值)；然后通过keepAlive续约，并监听keepAlive通道保持在线，如果不监听，etcd会删除这个租约上的key。

服务注册代码：registry.go

package etcd

import (
    "context"
    "encoding/json"
    "errors"
    "go.etcd.io/etcd/clientv3"
    "log"
    "time"
)

// 服务信息
type ServiceInfo struct {
    Name string
    IP   string
}

type Service struct {
    ServiceInfo ServiceInfo
    stop        chan error
    leaseId     clientv3.LeaseID
    client      *clientv3.Client
}

// NewService 创建一个注册服务
func NewService(info ServiceInfo, endpoints []string) (service *Service, err error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: time.Second * 200,
    })

    if err != nil {
        log.Fatal(err)
        return nil, err
    }

    service = &Service{
        ServiceInfo: info,
        client:      client,
    }
    return
}

// Start 注册服务启动
func (service *Service) Start() (err error) {

    ch, err := service.keepAlive()
    if err != nil {
        log.Fatal(err)
        return
    }

    for {
        select {
        case err := <-service.stop:
            return err
        case <-service.client.Ctx().Done():
            return errors.New("service closed")
        case resp, ok := <-ch:
            // 监听租约
            if !ok {
                log.Println("keep alive channel closed")
                return service.revoke()
            }
            log.Printf("Recv reply from service: %s, ttl:%d", service.getKey(), resp.TTL)
        }
    }

    return
}

func (service *Service) Stop() {
    service.stop <- nil
}

func (service *Service) keepAlive() (<-chan *clientv3.LeaseKeepAliveResponse, error) {
    info := &service.ServiceInfo
    key := info.Name + "/" + info.IP
    val, _ := json.Marshal(info)

    // 创建一个租约
    resp, err := service.client.Grant(context.TODO(), 5)
    if err != nil {
        log.Fatal(err)
        return nil, err
    }

    _, err = service.client.Put(context.TODO(), key, string(val), clientv3.WithLease(resp.ID))
    if err != nil {
        log.Fatal(err)
        return nil, err
    }
    service.leaseId = resp.ID
    return service.client.KeepAlive(context.TODO(), resp.ID)
}

func (service *Service) revoke() error {
    _, err := service.client.Revoke(context.TODO(), service.leaseId)
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("servide:%s stop\n", service.getKey())
    return err
}

func (service *Service) getKey() string {
    return service.ServiceInfo.Name + "/" + service.ServiceInfo.IP
}

• etcd服务发现(grpc)

服务发现：通过前缀取出key对应的values；然后启动一个监听服务监听key的变化

服务发现代码(grpc)：grpc_resolver.go

package etcd

import (
    "context"
    "encoding/json"
    "fmt"
    "github.com/coreos/etcd/clientv3"
    "github.com/coreos/etcd/mvcc/mvccpb"
    "google.golang.org/grpc/resolver"
    "log"
)

const schema = "etcd"

// resolver is the implementaion of grpc.resolve.Builder
// Resolver 实现grpc的grpc.resolve.Builder接口的Build与Scheme方法
type Resolver struct {
    endpoints []string
    service   string
    cli       *clientv3.Client
    cc        resolver.ClientConn
}

// NewResolver return resolver builder
// endpoints example: http://127.0.0.1:2379 http://127.0.0.1:12379 http://127.0.0.1:22379"
// service is service name
func NewResolver(endpoints []string, service string) resolver.Builder {
    return &Resolver{endpoints: endpoints, service: service}
}

// Scheme return etcd schema
func (r *Resolver) Scheme() string {
    // 最好用这种，因为grpc resolver.Register(r)在注册时，会取scheme，如果一个系统有多个grpc发现，就会覆盖之前注册的
    return schema + "_" + r.service
}

// ResolveNow
func (r *Resolver) ResolveNow(rn resolver.ResolveNowOption) {
}

// Close
func (r *Resolver) Close() {
}

// Build to resolver.Resolver
// 实现grpc.resolve.Builder接口的方法
func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) {
    var err error

    r.cli, err = clientv3.New(clientv3.Config{
        Endpoints: r.endpoints,
    })
    if err != nil {
        return nil, fmt.Errorf("grpclb: create clientv3 client failed: %v", err)
    }

    r.cc = cc

    // go r.watch(fmt.Sprintf("/%s/%s/", schema, r.service))
    go r.watch(fmt.Sprintf(r.service))

    return r, nil
}

func (r *Resolver) watch(prefix string) {
    addrDict := make(map[string]resolver.Address)

    update := func() {
        addrList := make([]resolver.Address, 0, len(addrDict))
        for _, v := range addrDict {
            addrList = append(addrList, v)
        }
        r.cc.NewAddress(addrList)
    }

    resp, err := r.cli.Get(context.Background(), prefix, clientv3.WithPrefix())
    if err == nil {
        for i, kv := range resp.Kvs {
            info := &ServiceInfo{}
            err := json.Unmarshal([]byte(kv.Value), info)
            if err != nil {

            }
            addrDict[string(resp.Kvs[i].Value)] = resolver.Address{Addr: info.IP}
        }
    }

    update()

    rch := r.cli.Watch(context.Background(), prefix, clientv3.WithPrefix(), clientv3.WithPrevKV())
    for n := range rch {
        for _, ev := range n.Events {
            switch ev.Type {
            case mvccpb.PUT:
                info := &ServiceInfo{}
                err := json.Unmarshal([]byte(ev.Kv.Value), info)
                if err != nil {
                    log.Println(err)
                } else {
                    addrDict[string(ev.Kv.Key)] = resolver.Address{Addr: info.IP}
                }
            case mvccpb.DELETE:
                delete(addrDict, string(ev.PrevKv.Key))
            }
        }
        update()
    }
}

grpc

• RPC是什么
  在分布式计算，远程过程调用（英语：Remote Procedure Call，缩写为 RPC）是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一个地址空间（通常为一个开放网络的一台计算机）的子程序，而程序员就像调用本地程序一样，无需额外地为这个交互作用编程（无需关注细节）。RPC是一种服务器-客户端（Client/Server）模式，经典实现是一个通过发送请求-接受回应进行信息交互的系统。

• gRPC是什么
  gRPC是一种现代化开源的高性能RPC框架，能够运行于任意环境之中。最初由谷歌进行开发。它使用HTTP/2作为传输协议。

在gRPC里，客户端可以像调用本地方法一样直接调用其他机器上的服务端应用程序的方法，是你更容易创建分布式应用程序和服务。与许多RPC系统一样，gRPC是基于定义一个服务，指定一个可以远程调用的带有参数和返回类型的的方法。在服务端程序中实现这个接口并且运行gRPC服务处理客户端调用。在客户端，有一个stub提供和服务端相同的方法。

image.png

• 为什么要用gRPC
  使用gRPC， 我们可以一次性的在一个.proto文件中定义服务并使用任何支持它的语言去实现客户端和服务端，反过来，它们可以应用在各种场景中，从Google的服务器到你自己的平板电脑—— gRPC帮你解决了不同语言及环境间通信的复杂性。使用protocol buffers还能获得其他好处，包括高效的序列号，简单的IDL以及容易进行接口更新。总之一句话，使用gRPC能让我们更容易编写跨语言的分布式代码。
• 安装grpc

go get -u google.golang.org/grpc

• 安装Protocol Buffers v3
  安装用于生成gRPC服务代码的协议编译器，最简单的方法是从下面的链接：https://github.com/google/protobuf/releases下载适合你平台的预编译好的二进制文件（protoc-<version>-<platform>.zip）。

下载完之后，执行下面的步骤：

1. 解压下载好的文件
2. 把protoc二进制文件的路径加到环境变量中

接下来执行下面的命令安装protoc的Go插件：

go get -u github.com/golang/protobuf/protoc-gen-go

编译插件protoc-gen-go将会安装到$GOBIN，默认是$GOPATH/bin，它必须在你的$PATH中以便协议编译器protoc能够找到它。

• grpc简单示例
  目录结构

.
├── mail
│   ├── Makefile
│   ├── client_test.go
│   └── server.go
├── proto
│   ├── mail
│      ├── mail.pb.go
│      └── mail.proto

编写proto代码

syntax = "proto3"; // 版本声明，使用Protocol Buffers v3版本
package g.srv.mail; // 包名
// 创建一个邮件服务
service MailService {
    rpc SendMail (MailRequest) returns (MailResponse) {
    }
}
// 请求消息
message MailRequest {
    string Mail = 1; 
    string Text = 2;
}
// 响应消息
message MailResponse {
    bool Ok = 1;
}

Makefile文件

build:
    protoc --proto_path=../proto --go_out=plugins=grpc:../proto ../proto/mail/mail.proto

执行make build就会生成mail.pb.go文件

编写grpc服务端代码

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "google.golang.org/grpc/reflection"
    "log"
    "net"
    pb "chen/app/srvs/proto/mail"
    "chen/pkg/etcd"
)

type service struct {
}

func (s *service) SendMail(ctx context.Context, req *pb.MailRequest) (res *pb.MailResponse, err error) {
    fmt.Printf("邮箱:%s;发送内容:%s", req.Mail, req.Text)
    return &pb.MailResponse{
        Ok: true,
    }, nil
}

func main() {

    // 监听本地的8972端口
    lis, err := net.Listen("tcp", ":8972")
    if err != nil {
        fmt.Printf("failed to listen: %v", err)
        return
    }
    s := grpc.NewServer()                       // 创建gRPC服务器

    pb.RegisterMailServiceServer(s, &service{}) // 在gRPC服务端注册服务

    reflection.Register(s) //在给定的gRPC服务器上注册服务器反射服务

    // Serve方法在lis上接受传入连接，为每个连接创建一个ServerTransport和server的goroutine。
    // 该goroutine读取gRPC请求，然后调用已注册的处理程序来响应它们。
    
    //etcd服务注册
    reg, err := etcd.NewService(etcd.ServiceInfo{
        Name: "g.srv.mail",
        IP:   "127.0.0.1:8972", //grpc服务节点ip
    }, []string{"127.0.0.1:2379", "127.0.0.1:22379", "127.0.0.1:32379"}) // etcd的节点ip
    if err != nil {
        log.Fatal(err)
    }
    go reg.Start()

    if err := s.Serve(lis); err != nil {
        fmt.Println(err)
    }
}

编写grpc客户端代码

package main

import (
    "context"
    "fmt"
    "google.golang.org/grpc"
    "time"

    //"google.golang.org/grpc/balancer/roundrobin"
    "google.golang.org/grpc/resolver"
    "log"
    "testing"
    //"time"
    pb "chen/app/srvs/proto/mail"

    "chen/pkg/etcd"
)

func TestService_SendMail(t *testing.T) {
    r := etcd.NewResolver([]string{
        "127.0.0.1:2379",
        "127.0.0.1:22379",
        "127.0.0.1:32379",
    }, "g.srv.mail")
    resolver.Register(r)

    ctx, _ := context.WithTimeout(context.Background(), 3*time.Second)
    // https://github.com/grpc/grpc/blob/master/doc/naming.md
    // The gRPC client library will use the specified scheme to pick the right resolver plugin and pass it the fully qualified name string.

    addr := fmt.Sprintf("%s:///%s", r.Scheme(), "g.srv.mail" /*g.srv.mail经测试，这个可以随便写，底层只是取scheme对应的Build对象*/)

    conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(),

        // grpc.WithBalancerName(roundrobin.Name),
        //指定初始化round_robin => balancer (后续可以自行定制balancer和 register、resolver 同样的方式)
        grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),

        grpc.WithBlock())

    // 这种方式也行
    //conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithBalancerName("round_robin"))

    //conn, err := grpc.Dial(":8972", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("failed to dial: %v", err)
    }

    /*conn, err := grpc.Dial(
        fmt.Sprintf("%s://%s/%s", "consul", GetConsulHost(), s.Name),
        //不能block => blockkingPicker打开，在调用轮询时picker_wrapper => picker时若block则不进行robin操作直接返回失败
        //grpc.WithBlock(),
        grpc.WithInsecure(),
        //指定初始化round_robin => balancer (后续可以自行定制balancer和 register、resolver 同样的方式)
        grpc.WithBalancerName(roundrobin.Name),
        //grpc.WithDefaultServiceConfig(`{"loadBalancingConfig": [{"round_robin":{}}]}`),
    )
    //原文链接：https://blog.csdn.net/qq_35916684/article/details/104055246*/

    if err != nil {
        panic(err)
    }

    c := pb.NewMailServiceClient(conn)

    resp, err := c.SendMail(context.TODO(), &pb.MailRequest{
        Mail: "qq@mail.com",
        Text: "test,test",
    })
    log.Print(resp)
}

运行代码就能看到效果了。

golang微服务框架

go-micro：可以看这个教程: Golang 微服务教程

有疑问加站长微信联系（非本文作者）