GO-GRPC实践(一) 完成第一个GRPC接口并使用etcd作为服务注册和发现

go项目实践总结 , 评论

demo代码地址

https://github.com/Me1onRind/go-demo

环境搭建

go

go 1.13 以上

需安装的二进制文件

可执行文件名 安装方式 作用
protoc https://github.com/protocolbuffers/protobuf/releases 下载安装 将.proto文件编译为具体编程语言的文件
protoc-gen-go go get github.com/golang/protobuf/protoc-gen-go@v1.3.3 将.proto文件编译为go文件时需要的插件

使用etcd作为服务注册中心

https://github.com/etcd-io/etcd

非生产环境在本地单机部署或者使用docker运行即可

docker-compose.yml

https://github.com/Me1onRind/my_docker/blob/master/etcd/docker-compose.yml

编写go服务端代码

项目目录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
./
├── cmd
│   └── grpc
│       ├── client_test.go # 测试文件
│       └── main.go # main文件
├── go.mod
├── go.sum
├── internal
│   ├── controller # grpc接口实现
│   │   └── foo_controller.go
│   ├── core
│   │   └── register # 服务注册实现
│   │       └── etcd.go
├── protobuf # proto原型文件和编译后的文件
│   ├── build.sh
│   ├── foo.proto
│   └── pb
│       └── foo.pb.go
└── README.md

代码

greet.proto

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
syntax = "proto3";

package pb;

service Foo {  
    rpc Greet(GreetReq) returns (GreetResp);
}

message GreetReq {
    string my_name = 1;
    string msg = 2;
}

message GreetResp {
    string msg = 1;
}
编译proto文件生成go代码
1
[root@debian go-demo]# protoc --proto_path=./:  --go_out=plugins=grpc:./pb/. ./*.proto

为了避免每次都要输一串命令(还有其他用处), 将编译命令写在shell脚本里

build.sh
1
2
3
4
#!/bin/bash
set -ue
cd `dirname $0`
protoc --proto_path=./:  --go_out=plugins=grpc:./pb/. ./*.proto

之后更新proto文件后只需执行

1
[root@debian go-demo]# sh protobuf/build.sh

foo_controller.go

实现定义的Foo接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
package controller

import (
    "context"
    "fmt"

    "github.com/Me1onRind/go-demo/protobuf/pb"
)

type FooController struct {
}

func NewFooController() *FooController {
    f := &FooController{}
    return f
}

func (f *FooController) Greet(ctx context.Context, in *pb.GreetReq) (*pb.GreetResp, error) {
    reply := fmt.Sprintf("Hello %s, I got your msg:%s", in.GetMyName(), in.GetMsg())
    out := &pb.GreetResp{}
    out.Msg = reply
    return out, nil
}

etcd.go

服务注册功能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
package register

import (
    "context"
    "fmt"
    "log"
    "time"

    uuid "github.com/satori/go.uuid"
    clientv3 "go.etcd.io/etcd/client/v3"
    "go.etcd.io/etcd/client/v3/naming/endpoints"
)

var client *clientv3.Client

const (
    prefix = "service"
)

func init() {
    var err error 
    client, err = clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    }) 
    if err != nil {
        panic(err)
    }
}

func Register(ctx context.Context, serviceName, addr string) error {
    log.Println("Try register to etcd ...")
    // 创建一个租约
    lease := clientv3.NewLease(client)
    cancelCtx, cancel := context.WithTimeout(ctx, time.Second*3)
    defer cancel()
    leaseResp, err := lease.Grant(cancelCtx, 3)
    if err != nil {
        return err
    }

    leaseChannel, err := lease.KeepAlive(ctx, leaseResp.ID) // 长链接, 不用设置超时时间
    if err != nil {
        return err
    }

    em, err := endpoints.NewManager(client, prefix)
    if err != nil {
        return err
    }

    cancelCtx, cancel = context.WithTimeout(ctx, time.Second*3)
    defer cancel()
    if err := em.AddEndpoint(cancelCtx, fmt.Sprintf("%s/%s/%s", prefix, serviceName, uuid.NewV4().String()), endpoints.Endpoint{
        Addr: addr,
    }, clientv3.WithLease(leaseResp.ID)); err != nil {
        return err
    }
    log.Println("Register etcd success")

    del := func() {
        log.Println("Register close")

        cancelCtx, cancel = context.WithTimeout(ctx, time.Second*3)
        defer cancel()
        em.DeleteEndpoint(cancelCtx, serviceName)

        lease.Close()
    }
    // 保持注册状态(连接断开重连)
    keepRegister(ctx, leaseChannel, del, serviceName, addr)

    return nil
}

func keepRegister(ctx context.Context, leaseChannel <-chan *clientv3.LeaseKeepAliveResponse, cleanFunc func(), serviceName, addr string) {
    go func() {
        failedCount := 0
        for {
            select {
            case resp := <-leaseChannel:
                if resp != nil {
                    //log.Println("keep alive success.")
                } else {
                    log.Println("keep alive failed.")
                    failedCount++
                    for failedCount > 3 {
                        cleanFunc()
                        if err := Register(ctx, serviceName, addr); err != nil {
                            time.Sleep(time.Second)
                            continue
                        }
                        return
                    }
                    continue
                }
            case <-ctx.Done():
                cleanFunc()
                client.Close()
                return
            }
        }
    }()
}

main.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package main
    
import (
    "context"
    "fmt"
    "log"
    "net"

    "github.com/Me1onRind/go-demo/internal/controller"
    "github.com/Me1onRind/go-demo/internal/core/register"
    "github.com/Me1onRind/go-demo/protobuf/pb"

    "google.golang.org/grpc"
)

func registerService(s *grpc.Server) {
    pb.RegisterFooServer(s, controller.NewFooController())
}   
    
func main() {
    addr := "127.0.0.1:8080"
    ctx := context.Background()

    lis, err := net.Listen("tcp", addr)
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    s := grpc.NewServer()
    registerService(s)

    if err := register.Register(ctx, "go-demo", addr); err != nil { // 服务注册名: go-demo
        log.Fatalf("register %s failed:%v", "go-demo", err)
    }

    fmt.Printf("start grpc server:%s", addr)
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

通过服务发现调用Foo.Greet方法

client_test.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
package main
                               
import (
    "context"
    "testing"                  
    "time"
    
    "github.com/Me1onRind/go-demo/protobuf/pb"
    "go.etcd.io/etcd/client/v3"
    "go.etcd.io/etcd/client/v3/naming/resolver"
    "google.golang.org/grpc"   
) 

    
func Test_Greet(t *testing.T) {
  
    cli, err := clientv3.NewFromURL("http://localhost:2379")
    if err != nil {            
        t.Fatal(err)           
    }
    builder, err := resolver.NewBuilder(cli) 
    if err != nil {
        t.Fatal(err)
    }
    conn, err := grpc.Dial("etcd:///service/go-demo",
        grpc.WithResolvers(builder),    
        grpc.WithBalancerName("round_robin"),
        grpc.WithInsecure(), grpc.WithTimeout(time.Second))
    if err != nil {
        t.Fatal(err)
    }

    fooClient := pb.NewFooClient(conn)
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
    defer cancel()
    resp, err := fooClient.Greet(ctx, &pb.GreetReq{
        MyName: "Bar",
        Msg:    "Hello, World",
    })
    if err != nil {
        t.Fatal(err)
    }

    t.Log(resp.Msg)
}

验证

启动server

1
2
[root@debian go-demo]# go run cmd/grpc/main.go 
start grpc server:127.0.0.1:8080

可以使用etcd命令行客户端/UI客户端看到, 服务已经注册上去

客户端调用

1
2
3
4
5
=== RUN   Test_Greet
    client_test.go:43: Hello Bar, I got your msg:Hello, World
-- PASS: Test_Greet (0.00s)
PASS
ok      github.com/Me1onRind/go-demo/cmd/grpc   0.010s