grpc 基于 etcd 的服务发现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

grpc-etcd
├── go.mod
├── grpc-client
│   └── main.go
├── grpc-server
│   ├── etcd.go
│   └── main.go
└── pb
    ├── hello_grpc.pb.go
    ├── hello.pb.go
    └── hello.proto

 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

// grpc-etcd/pb/hello/proto

syntax = "proto3"; // 版本声明，使用Protocol Buffers v3版本

option go_package = "grpc-etcd/pb";  // 指定生成的Go代码在你项目中的导入路径

package hello; // 包名，方便其他 proto 文件引入

// 定义服务，到时候是 Greeter.SayHello。注册到服务发现中心的是 ip:port
service Greeter1 {
    // SayHello 方法
    rpc SayHello1 (HelloRequest) returns (HelloResponse) {}
}


service Greeter2 {
    // SayHello 方法
    rpc SayHello2 (HelloRequest) returns (HelloResponse) {}
}

// 请求消息
message HelloRequest {
    string name = 1;
}

// 响应消息
message HelloResponse {
    string reply = 1;
}

 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

// grpc-etcd/grpc-server/mian.go

package main

import (
	"context"
	"google.golang.org/grpc"
	"grpc-etcd/pb"
	"log"
	"net"
)

// 服务1
type server1 struct {
	pb.UnimplementedGreeter1Server
}

func (server1) SayHello1(context.Context, *pb.HelloRequest) (*pb.HelloResponse, error) {
	resp := new(pb.HelloResponse)
	resp.Reply = "server1:hello"
	return resp, nil
}

// 服务2
type server2 struct {
	pb.UnimplementedGreeter2Server
}

func (server2) SayHello2(context.Context, *pb.HelloRequest) (*pb.HelloResponse, error) {
	resp := new(pb.HelloResponse)
	resp.Reply = "server2:hello"
	return resp, nil
}

const (
	ServerAddr1 = "127.0.0.1:8080"
	ServerAddr2 = "127.0.0.1:8081"
	ServerName1 = "ayang/server1"
	ServerName2 = "ayang/server2"
)

func main() {
	var err error
	// 1. 创建两个 tcp 连接
	conn1, err := net.Listen("tcp", ServerAddr1)
	conn2, err := net.Listen("tcp", ServerAddr2)

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

	// 2. 创建两个 grpc 服务器
	s1 := grpc.NewServer()
	s2 := grpc.NewServer()

	// 3. 注册到 grpc 服务器中
	pb.RegisterGreeter1Server(s1, &server1{})
	pb.RegisterGreeter2Server(s2, &server2{})

	// 4. 注册到 etcd 中
	go registerEndPointToEtcd(context.TODO(), ServerAddr1, ServerName1)
	go registerEndPointToEtcd(context.TODO(), ServerAddr2, ServerName2)

	go func() {
		err = s1.Serve(conn1)
		if err != nil {
			log.Fatal(err.Error())
			return
		}
	}()

	go func() {
		err = s2.Serve(conn2)
		if err != nil {
			log.Fatal(err.Error())
			return
		}
	}()

	<-make(chan struct{})

}

 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

// grpc-etcd/grpc-server/etcd.go

package main

import (
	"context"
	"fmt"
	eclient "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/naming/endpoints"
	"time"
)

const (
    // etcd 服务器的地址
	EtcdAddr = "http://localhost:2379"
)

func registerEndPointToEtcd(ctx context.Context, serverAddr, serverName string) {
	// 创建 etcd 客户端
	etcdClient, _ := eclient.NewFromURL(EtcdAddr)
	etcdManager, _ := endpoints.NewManager(etcdClient, serverName) 

	// 创建一个租约，每隔 10s 需要向 etcd 汇报一次心跳，证明当前节点仍然存活
	var ttl int64 = 10
	lease, _ := etcdClient.Grant(ctx, ttl)

	// 添加注册节点到 etcd 中，并且携带上租约 id  
    // 以 serverName/serverAddr 为 key，serverAddr 为 value
    // serverName/serverAddr 中的 serverAddr 可以自定义，只要能够区分同一个 grpc 服务器功能的不同机器即可
	_ = etcdManager.AddEndpoint(ctx, fmt.Sprintf("%s/%s", serverName, serverAddr), endpoints.Endpoint{Addr: serverAddr}, eclient.WithLease(lease.ID))

	// 每隔 5 s进行一次延续租约的动作
	for {
		select {
		case <-time.After(5 * time.Second):
			// 续约操作
			resp, _ := etcdClient.KeepAliveOnce(ctx, lease.ID)
			fmt.Printf("keep alive resp: %+v\n", resp)
		case <-ctx.Done():
			return
		}
	}
}

1
2
3
4
5

ayang@Ubuntu22:~$ etcdctl get --prefix ""
ayang/server1/127.0.0.1:8080
{"Op":0,"Addr":"127.0.0.1:8080","Metadata":null}
ayang/server2/127.0.0.1:8081
{"Op":0,"Addr":"127.0.0.1:8081","Metadata":null}

 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

// grpc-etcd/grpc-client/etcd.go

package main

import (
	"context"
	"fmt"
	eresolver "go.etcd.io/etcd/client/v3/naming/resolver"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"grpc-etcd/pb"
	"log"
	// etcd
	eclient "go.etcd.io/etcd/client/v3"
)

const (
	serverNamePreResolve = "etcd:///ayang/server1"
	EtcdAddr             = "http://localhost:2379"
)

func main() {
	var err error
	// 创建 etcd 客户端
	etcdClient, err := eclient.NewFromURL(EtcdAddr)
	if err != nil {
		log.Fatalln(err.Error())
		return
	}

	// 创建 etcd 实现的 grpc 服务注册发现模块 resolver
	etcdResolverBuilder, err := eresolver.NewBuilder(etcdClient)
	if err != nil {
		log.Fatalln(err.Error())
		return
	}

	// 创建 grpc 连接代理
	conn, err := grpc.Dial(
		// 服务名称
		serverNamePreResolve,
		// 注入 etcd resolver
		grpc.WithResolvers(etcdResolverBuilder),
		// 声明使用的负载均衡策略为 roundrobin，轮询。（测试 target 时去除该注释）
		// grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		log.Fatalln(err.Error())
		return
	}

	for i := 0; i < 4; i++ {
		greeter1 := pb.NewGreeter1Client(conn)
		resp, err := greeter1.SayHello1(context.Background(), &pb.HelloRequest{
			Name: "ayang",
		})

		if err != nil {
			log.Fatalln(err.Error())
			return
		}

		fmt.Printf("%d  %s\n", i, resp.Reply)
	}

	defer conn.Close()
}

1
2
3
4

0  server1:hello
1  server1:hello
2  server1:hello
3  server1:hello

1
2

0  server1:hello
2023/07/25 23:43:28 rpc error: code = Unimplemented desc = unknown service hello.Greeter1