RPC(Remote Procedure Call 远程过程调用)

一个节点请求另一个节点提供的服务
远程调用的参数传递的协议：json、xml、protobuf
远程调用会引入新问题：

Call ID映射

我们怎么告诉远程机器我们要调用add，而不是sub或者Foo呢？在本地调用中，函数体是直接通过函数指针来指定的，我们调用add，编译器就自动帮我们调用它相应的函数指针。但是在远程调用中，函数指针是不行的，因为两个进程的地址空间是完全不一样的。所以，在RPC中，所有的函数都必须有自己的一个ID。这个ID在所有进程中都是唯一确定的。客户端在做远程过程调用时，必须附上这个ID。==然后我们还需要在客户端和服务端分别维护一个 {函数 <–> Call ID} 的对应表。两者的表不一定需要完全相同，但相同的函数对应的Call ID必须相同。当客户端需要进行远程调用时，它就查一下这个表，找出相应的Call ID，然后把它传给服务端，服务端也通过查表，来确定客户端需要调用的函数，然后执行相应函数的代码。==

序列化和反序列化

客户端怎么把参数值传给远程的函数呢？在本地调用中，我们只需要把参数压到栈里，然后让函数自己去栈里读就行。但是在远程过程调用时，客户端跟服务端是不同的进程，不能通过内存来传递参数。==甚至有时候客户端和服务端使用的都不是同一种语言（比如服务端用C++，客户端用Java或者Python）。这时候就需要客户端把参数先转成一个字节流，传给服务端后，再把字节流转成自己能读取的格式。这个过程叫序列化和反序列化。同理，从服务端返回的值也需要序列化反序列化的过程。==

网络传输

远程调用往往用在网络上，客户端和服务端是通过网络连接的。所有的数据都需要通过网络传输，因此就需要有一个网络传输层。网络传输层需要把Call ID和序列化后的参数字节流传给服务端，然后再把序列化后的调用结果传回客户端。只要能完成这两者的，都可以作为传输层使用。因此，它所使用的协议其实是不限的，能完成传输就行。尽管大部分RPC框架都使用TCP协议，但其实UDP也可以，而gRPC干脆就用了HTTP2。Java的Netty也属于这层的东西。

解决了上面三个机制，就能实现RPC了，具体过程如下：

client端解决的问题：

1
2
3
4
5
6
1. 将这个调用映射为Call ID。这里假设用最简单的字符串当Call ID的方法
2. 将Call ID，a和b序列化(a,b为参数)。可以直接将它们的值以二进制形式打包
3. 把2中得到的数据包发送给ServerAddr，这需要使用网络传输层
4. 等待服务器返回结果
4. 如果服务器调用成功，那么就将结果反序列化，并赋给total(函数返回值)

server端解决的问题

1
2
3
4
5
6
7
1. 在本地维护一个Call ID到函数指针的映射call_id_map，可以用dict完成
2. 等待请求，包括多线程的并发处理能力
3. 得到一个请求后，将其数据包反序列化，得到Call ID
4. 通过在call_id_map中查找，得到相应的函数指针
5. 将a和rb反序列化后，在本地调用add函数，得到结果
6. 将结果序列化后通过网络返回给Client

RPC技术在架构设计上有四部分组成，分别是：客户端、客户端存根、服务端、服务端存根。

  • 客户端(Client)：
    服务调用发起方，也称为服务消费者。

  • 客户端存根(Client Stub)：
    该程序运行在客户端所在的计算机机器上，主要用来存储要调用的服务器的地址，另外，该程序还负责将客户端请求远端服务器程序的数据信息打包成数据包，通过网络发送给服务端Stub程序；其次，还要接收服务端Stub程序发送的调用结果数据包，并解析返回给客户端。

  • 服务端(Server)：
    远端的计算机机器上运行的程序，其中有客户端要调用的方法。

  • 服务端存根(Server Stub)：
    接收客户Stub程序通过网络发送的请求消息数据包，并调用服务端中真正的程序功能方法，完成功能调用；其次，将服务端执行调用的结果进行数据处理打包发送给客户端Stub程序。

    6037326e09f549e107620449

rpc案例

使用go自己的协议：Gob

Client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func main() {
	//1. Create a new connection
	//Gob协议
	client, err := rpc.Dial("tcp", "localhost:1234")
	if err != nil {
		panic("failed to connect")
	}
	var reply string
	err = client.Call("HelloService.Hello", "rider", &reply)
	if err != nil {
		panic("failed to call")
	}
	fmt.Println(reply)
}

Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
type HelloService struct{}

func (s *HelloService) Hello(request string, reply *string) error {
	*reply = "Hello, " + request
	return nil
}

func main() {
	//1. Create a new server
	listener, _ := net.Listen("tcp", ":1234")
	//2. Register the handler
	_ = rpc.RegisterName("HelloService", &HelloService{})
	//3. Start the server
	conn, _ := listener.Accept()
	rpc.ServeConn(conn)
}

使用json协议

Client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
func main() {
	//1. Create a new connection
	//使用net获得连接
	conn, err := net.Dial("tcp", "localhost:1234")
	if err != nil {
		panic("failed to connect")
	}
	var reply string
	//获取json协议客户端
	client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))
	err = client.Call("HelloService.Hello", "Rider", &reply)
	if err != nil {
		panic("failed to call")
	}
	fmt.Println(reply)
}

Server
for循环持续监听，通过goroutine处理并发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
type HelloService struct{}

func (s *HelloService) Hello(request string, reply *string) error {
	*reply = "Hello, " + request
	return nil
}

func main() {
	//1. Create a new server
	listener, _ := net.Listen("tcp", ":1234")
	//2. Register the handler
	_ = rpc.RegisterName("HelloService", &HelloService{})
	//3. Start the server
	for {
		conn, _ := listener.Accept()
		//start goroutine to handle concurrency issues
		go rpc.ServeCodec(jsonrpc.NewServerCodec(conn))
	}
}

跨语言调用

Python代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import json
import socket

request = {
    "id":0,
    "params":["Rider_Python"],
    "method":"HelloService.Hello"
}

client = socket.create_connection(("localhost", 1234))
client.sendall(json.dumps(request).encode())

rsp = client.recv(1024) #二进制返回值
rsp = json.loads(rsp.decode())

print(rsp)
{'id': 0, 'result': 'Hello, Rider_Python', 'error': None}

GRPC

注意-使用protoc生成对应语言的代码

看官网文档下载对应语言的包到项目中，在使用protoc生成语句

文档地址：https://grpc.io/docs/languages/go/quickstart/

同时项目中要导入protobuf和grpc包。才能使用生成的相关代码

Protobuf

protoc生成目标源文件到当前目录下：

option go_package = "./;currentFolder";

GRPC Quick Start

proto文件

syntax = "proto3";

option go_package = "./;proto";

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply);
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

生成go client和go server代码

protoc --go_out=. --go-grpc_out=require_unimplemented_servers=false:. helloworld.proto

Go Client请求代码

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
package main

import (
	"GoTest/grpc/grpc_quickstart/proto"
	"context"
	"fmt"
	"google.golang.org/grpc"
)

func main() {
	conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure())
	if err != nil {
		panic("Failed to connect: " + err.Error())
	}
	defer func(conn *grpc.ClientConn) {
		err := conn.Close()
		if err != nil {
			panic("Failed to close connection: " + err.Error())
		}
	}(conn)

	cli := proto.NewGreeterClient(conn)
	rsp, err := cli.SayHello(context.Background(), &proto.HelloRequest{
		Name: "Hello server! Rider calling",
	})
	if err != nil {
		panic("Failed to say hello" + err.Error())
	}
	fmt.Println("Call Rsp: " + rsp.Message)
}

Go Server代码

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
package main

import (
	"GoTest/grpc/grpc_quickstart/proto"
	"context"
	"google.golang.org/grpc"
	"net"
)

type Server struct{}

func (s *Server) SayHello(ctx context.Context, req *proto.HelloRequest) (*proto.HelloReply, error) {
	return &proto.HelloReply{
		Message: "Hello proto grpc server [" + req.Name + "]",
	}, nil
}

func main() {
	g := grpc.NewServer()
	proto.RegisterGreeterServer(g, &Server{})
	lis, err := net.Listen("tcp", "0.0.0.0:8080")
	if err != nil {
		panic("Failed to listen: " + err.Error())
	}
	err = g.Serve(lis)
	if err != nil {
		panic("Failed to start grpc serve: " + err.Error())
	}
}

运行Server以后，服务一直在线，底层通过协程处理，可以一直请求。请求结果：

1
Call Rsp: Hello proto grpc server [Hello server! Rider calling]

GRPC 流模式

服务端流模式：服务端持续向客户端发送消息（像气象信息，服务端可以实时更新发送）。

客户端流模式：客户端持续向服务端发送消息（像农业物联网设备，持续发送相关传感器参数）。

双向流模式：服务端和客户端持续通信，两者的收发都在写成中，收发互不影响。

测试

proto文件

syntax = "proto3";

option go_package = "./;proto";

service Greeter{
  //服务端流模式
  rpc ServerStream(StreamReqData) returns (stream StreamRspData);

  //客户端流模式
  rpc ClientStream(stream StreamReqData) returns (StreamRspData);

  //双向流模式
  rpc DuplexStream(stream StreamReqData) returns (stream StreamRspData);
}

message StreamReqData{
  string data = 1;
}

message StreamRspData{
  string data = 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
84
85
86
87
88
89
90
91
92
93
94
95
	package main

import (
	"GoTest/grpc_stream/proto"
	"fmt"
	"google.golang.org/grpc"
	"net"
	"sync"
	"time"
)

const PORT = ":50052"

type server struct{}

// 服务端流模式
func (s *server) ServerStream(req *proto.StreamReqData, serstr proto.Greeter_ServerStreamServer) error {
	tar := 0
	for {
		_ = serstr.Send(&proto.StreamRspData{
			Data: fmt.Sprintf("%v", time.Now().String()),
		})
		time.Sleep(time.Second)
		tar++
		if tar > 10 {
			break
		}
	}
	return nil
}

// 客户端流模式
func (s *server) ClientStream(clistr proto.Greeter_ClientStreamServer) error {
	for {
		recv, err := clistr.Recv()
		if err != nil {
			fmt.Println(err)
			break
		}
		fmt.Println("Client stream: " + recv.Data)
	}
	return nil
}

// 双向流模式
func (s *server) DuplexStream(dupstr proto.Greeter_DuplexStreamServer) error {
	wg := sync.WaitGroup{}
	wg.Add(2)
	//send message
	go func() {
		defer wg.Done()
		times := 0
		for {
			times++
			err := dupstr.Send(&proto.StreamRspData{
				Data: fmt.Sprintf("Server Current Time: %v", time.Now().String()),
			})
			if err != nil {
				panic(err)
			}
			if times > 10 {
				break
			}
			time.Sleep(time.Second)
		}
	}()

	//receive message
	go func() {
		defer wg.Done()
		for {
			rec, err := dupstr.Recv()
			if err != nil {
				fmt.Println("Receive interrupt: ", err.Error())
				break
			}
			fmt.Println(rec.Data)
		}
	}()
	wg.Wait()
	return nil
}

func main() {
	lis, err := net.Listen("tcp", PORT)
	if err != nil {
		panic(err)
	}
	s := grpc.NewServer()
	proto.RegisterGreeterServer(s, &server{})
	err = s.Serve(lis)
	if err != nil {
		panic(err)
	}
}

客户端，三种模式的调用对应在三个方法中

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
105
106
107
108
109
110
111
112
113
114
115
116
package main

import (
	"GoTest/grpc_stream/proto"
	"context"
	"fmt"
	"google.golang.org/grpc"
	"sync"
	"time"
)

var wg = sync.WaitGroup{}

// 服务端流模式
func ServerStream(c proto.GreeterClient) {
	defer wg.Done()
	rsp, err := c.ServerStream(context.Background(), &proto.StreamReqData{
		Data: "imooc",
	})
	if err != nil {
		panic(err)
	}
	for {
		data, err := rsp.Recv()
		if err != nil {
			fmt.Println(err)
			break
		}
		fmt.Println("Server Time:", data)
	}
}

// 客户端流模式
func ClientStream(c proto.GreeterClient) {
	defer wg.Done()
	stream, err := c.ClientStream(context.Background())
	if err != nil {
		panic("Failed to invoke method: " + err.Error())
	}
	tar := 0
	for {
		tar++
		err = stream.Send(&proto.StreamReqData{
			Data: fmt.Sprintf("%v", time.Now().String()),
		})
		if err != nil {
			return
		}
		if tar > 10 {
			break
		}
		time.Sleep(time.Second)
	}
}

// 双工流模式
func DuplexStream(c proto.GreeterClient) {
	wg := sync.WaitGroup{}
	stream, err := c.DuplexStream(context.Background())
	if err != nil {
		panic(err.Error())
	}
	wg.Add(2)
	//send message
	go func() {
		defer wg.Done()
		times := 0
		for {
			times++
			err := stream.Send(&proto.StreamReqData{
				Data: fmt.Sprintf("Client Current Time: %v", time.Now().String()),
			})
			if err != nil {
				panic(err)
			}
			if times > 10 {
				break
			}
			time.Sleep(time.Second)
		}
	}()
	//receive message
	go func() {
		defer wg.Done()
		for {
			rec, err := stream.Recv()
			if err != nil {
				fmt.Println("Receive interrupt: ", err.Error())
				break
			}
			fmt.Println(rec.Data)
		}
	}()
	wg.Wait()
}

func main() {
	conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())
	if err != nil {
		panic(err)
	}
	defer func(conn *grpc.ClientConn) {
		err := conn.Close()
		if err != nil {

		}
	}(conn)
	c := proto.NewGreeterClient(conn)
	//服务端流和客户端流
	wg.Add(2)
	go ServerStream(c)
	go ClientStream(c)
	wg.Wait()
	//全双工
	DuplexStream(c)
}
进一步的使用其他语言来调用这个golang的grpc服务

这儿使用python来访问

安装grpc相关包

1
pip install grpcio-tools

使用grpcio-tools生成proto文件对应的代码

1
2
python -m grpc_tools.protoc -I F:\WorkSpace\Py_WorkSpace\Grpc-test --python_out=. --grpc_python_out=. stream.proto

==注意：protoc生成的python代码，pycharm可能会报错找不到相关类，因为python中，proto生成的代码相关message类是动态创建的，所以在执行时不会产生错误。==

python客户端代码：

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
import grpc
from stream_pb2 import StreamReqData
import stream_pb2_grpc
def run():
    # 打开一个 gRPC channel
    channel = grpc.insecure_channel('localhost:50052')  # 请根据实际情况修改服务器地址和端口

    # 创建一个 stub (客户端)
    stub = stream_pb2_grpc.GreeterStub(channel)

    # 创建一个 StreamReqData 请求
    req = StreamReqData()
    req.data = "hello protobuf_python"
    print(req)

    # 调用 ServerStream 方法
    responses = stub.ServerStream(req)

    try:
        # 遍历响应流
        for response in responses:
            print("Received message:", response.data)
    except grpc._channel._Rendezvous as err:
        print(err)

if __name__ == '__main__':
    run()
# 得到和golang client相同的结果
data: "hello protobuf_python"

Received message: 2024-01-19 22:31:48.5708539 +0800 CST m=+303.380034601
Received message: 2024-01-19 22:31:49.5839327 +0800 CST m=+304.393113401
Received message: 2024-01-19 22:31:50.5960374 +0800 CST m=+305.405218101
Received message: 2024-01-19 22:31:51.6042546 +0800 CST m=+306.413435301
Received message: 2024-01-19 22:31:52.6147512 +0800 CST m=+307.423931901
Received message: 2024-01-19 22:31:53.6261654 +0800 CST m=+308.435346101
Received message: 2024-01-19 22:31:54.6268166 +0800 CST m=+309.435997301
Received message: 2024-01-19 22:31:55.6313363 +0800 CST m=+310.440517001
Received message: 2024-01-19 22:31:56.6462103 +0800 CST m=+311.455391001
Received message: 2024-01-19 22:31:57.6549169 +0800 CST m=+312.464097601
Received message: 2024-01-19 22:31:58.6669002 +0800 CST m=+313.476080901

ProtoBuf详解

官方文档：https://protobuf.dev/programming-guides/proto3/

类型对应

Proto3 Type Go Tyepe
double float64
float float32
int32 使用变长编码，对于负值的效率很低，如果你的域有可能有负值，请使用sint64替代 int32
uint32 使用变长编码 uint32
uint64 使用变长编码 uint64
sint32 使用变长编码，这些编码在负值时比int32高效的多 int32
sint64 使用变长编码，有符号的整型值。编码时比通常的int64高效。 int64
fixed32 总是4个字节，如果数值总是比总是比228大的话，这个类型会比uint32高效。 uint32
fixed64 总是8个字节，如果数值总是比总是比256大的话，这个类型会比uint64高效。 uint64
sfixed32 总是4个字节 int32
sfixed64 总是8个字节 int64
bool bool
string 一个字符串必须是UTF-8编码或者7-bit ASCII编码的文本。 string
bytes 可能包含任意顺序的字节数据。 []byte

proto文件引用其他proto文件

1
import "google/protobuf/empty.proto"

protobuf的数组定义：repeated关键字

1
2
3
message Test{
	repeated string message = 1;
}

message嵌套

1
2
3
4
5
6
7
8
message HelloReply {
  string message = 1;
  repeated Result data = 2;
  message Result{
    string name = 1;
    string url = 2;
  }
}

枚举

1
2
3
4
5
6
7
8
message HelloRequest {
  string name = 1;
  Gender g= 2;
  enum Gender{
    MALE = 0;
    FEMALE = 1;
  }
}

内置时间类型

proto：

import "google/protobuf/timestamp.proto";

message HelloRequest {
  string name = 1;
  Gender g= 2;
  google.protobuf.Timestamp addTime = 3;
  map<string, string> mp = 4;
  enum Gender{
    MALE = 0;
    FEMALE = 1;
  }
}

Grpc进阶

metadata（可以权限验证）

metadata使得client和server能够为对方提供关于本次调用的一些信息，就像一次http请求的RequestHeader和ResponseHeader一样。http中header的生命周周期是一次http请求，那么metadata的生命周期就是一次RPC调用。

MD 类型实际上是map，key是string，value是string类型的slice。

1
type MD map[string][]string

创建的时候可以像创建普通的map类型一样使用new关键字进行创建：

1
2
3
4
5
6
7
8
//第一种方式
md := metadata.New(map[string]string{"key1": "val1", "key2": "val2"})
//第二种方式 key不区分大小写，会被统一转成小写。
md := metadata.Pairs(
    "key1", "val1",
    "key1", "val1-2", // "key1" will have map value []string{"val1", "val1-2"}
    "key2", "val2",
)

Client端发送

1
2
3
4
5
6
md := metadata.New(map[string]string{
	"name":    "bobby",
	"pasword": "imooc",
})//创建md对象
ctx := metadata.NewOutgoingContext(context.Background(), md)//绑定到contex
r, err := c.SayHello(ctx, &proto.HelloRequest{Name: "bobby"})

Server端接收

1
2
3
4
5
6
7
md, ok := metadata.FromIncomingContext(ctx)//从contex接收
if !ok {
	fmt.Println("get metadata error")
}
for key, val := range md {
	fmt.Println(key, ": ", val)
}

拦截器

服务端

1
2
3
4
5
6
7
8
//拦截器
	interceptor := func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
		fmt.Println("Get a new request")
		return handler(ctx, req)
	}

	opt := grpc.UnaryInterceptor(interceptor)
	g := grpc.NewServer(opt)

客户端

1
2
3
4
5
6
7
8
9
//stream
	interceptor := func(ctx context.Context, method string, req any, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		start := time.Now()
		err := invoker(ctx, method, req, reply, cc, opts...)
		fmt.Printf("耗时: %s\n", time.Since(start))
		return err
	}
	opt := grpc.WithUnaryInterceptor(interceptor)
	conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithInsecure(), opt)

通过metadata和拦截器实现token验证

客户端

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
/*
package: client
Desc:
Author: DuanQiHang
Date: 2024/1/22 17:43:00
*/

package main

import (
	"GoTest/metadata/proto"
	"context"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"time"
)

func main() {
	//stream
	interceptor := func(ctx context.Context, method string, req any, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		start := time.Now()
		md := metadata.New(map[string]string{
			"appid":  "101010",
			"appkey": "I am Rider",
		})
		ctx = metadata.NewOutgoingContext(context.Background(), md)
		err := invoker(ctx, method, req, reply, cc, opts...)
		fmt.Printf("耗时: %s\n", time.Since(start))
		return err
	}
	opt := grpc.WithUnaryInterceptor(interceptor)
	conn, err := grpc.Dial("127.0.0.1:50051", grpc.WithInsecure(), opt)
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	c := proto.NewGreeterClient(conn)

	//md := metadata.Pairs("timestamp", time.Now().Format(timestampFormat))
	/*md := metadata.New(map[string]string{
		"name":    "bobby",
		"pasword": "imooc",
	})
	ctx := metadata.NewOutgoingContext(context.Background(), md)*/
	r, err := c.SayHello(context.Background(), &proto.HelloRequest{Name: "bobby"})
	if err != nil {
		panic(err)
	}
	fmt.Println(r.Message)
}

服务端

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
/*
package: server
Desc:
Author: DuanQiHang
Date: 2024/1/22 17:47:00
*/
package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"net"

	"google.golang.org/grpc"

	"GoTest/metadata/proto"
)

type Server struct{}

func (s *Server) SayHello(ctx context.Context, request *proto.HelloRequest) (*proto.HelloReply,
	error) {

	/*md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		fmt.Println("get metadata error")
	}
	for key, val := range md {
		fmt.Println(key, ": ", val)
	}*/
	return &proto.HelloReply{
		Message: "hello " + request.Name,
	}, nil
}

func main() {
	//拦截器
	interceptor := func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
		fmt.Println("Get a new request")
		md, ok := metadata.FromIncomingContext(ctx)
		if !ok {
			return resp, status.Error(codes.Unauthenticated, "bad token")
		}
		for key, val := range md {
			fmt.Println(key, ": ", val)
		}
		var (
			appid  string
			appkey string
		)
		if val, ok := md["appid"]; ok {
			appid = val[0]
			fmt.Println("Appid: ", appid)
		}
		if val, ok := md["appkey"]; ok {
			appkey = val[0]
			fmt.Println("Appkey: ", appkey)
		}
		if appid != "101010" && appkey != "I am Rider" {
			fmt.Println("bad request 403")
		}
		fmt.Println("request done")
		return handler(ctx, req)
	}

	opt := grpc.UnaryInterceptor(interceptor)
	g := grpc.NewServer(opt)
	proto.RegisterGreeterServer(g, &Server{})
	lis, err := net.Listen("tcp", "0.0.0.0:50051")
	if err != nil {
		panic("failed to listen:" + err.Error())
	}
	err = g.Serve(lis)
	if err != nil {
		panic("failed to start grpc:" + err.Error())
	}
}

==更优雅的实现方式==

client端

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
/*
package: main
Desc:
Author: DuanQiHang
Date: 2024/1/23 21:55:00
*/
package main

import (
	"GoTest/metadata/proto"
	"context"
	"fmt"
	"google.golang.org/grpc"
	"time"
)

//实现接口 PerRPCCredentials ，包含如下两个方法
type customCredential struct{}

// 返回与请求相关的认证元数据
func (c *customCredential) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
		"appid":  "101010",
		"appkey": "I am Rider",
	}, nil
}
// 指示是否需要传输层安全性（TLS/SSL）。
func (c *customCredential) RequireTransportSecurity() bool {
	return false
}
// 自定义拦截1器逻辑
func customInterceptor(ctx context.Context, method string, req any, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	start := time.Now()
	// 调用实际的 gRPC 调用
	err := invoker(ctx, method, req, reply, cc, opts...)
	// 输出耗时
	fmt.Printf("耗时: %s\n", time.Since(start))
	return err
}
func main() {
	var opts []grpc.DialOption
	opts = append(opts, grpc.WithInsecure())
	opts = append(opts, grpc.WithUnaryInterceptor(customInterceptor))
	opts = append(opts, grpc.WithPerRPCCredentials(&customCredential{}))
	conn, err := grpc.Dial("127.0.0.1:50051", opts...)
	if err != nil {
		panic(err)
	}
	defer conn.Close()
	c := proto.NewGreeterClient(conn)
	r, err := c.SayHello(context.Background(), &proto.HelloRequest{Name: "bobby"})
	if err != nil {
		panic(err)
	}
	fmt.Println(r.Message)
}

验证器

开源项目：protoc-gen-validata [https://github.com/bufbuild/protoc-gen-validate?tab=readme-ov-file]

后续升级版：protovalidate [https://github.com/bufbuild/protovalidate?tab=readme-ov-file]

proto文件中message字段的校验规则写法如项目文档所示。

错误码处理

服务端

1
2
3
func (s *Server) SayHello(ctx context.Context, req *proto.HelloRequest) (*proto.HelloReply, error) {
	return nil, status.Errorf(codes.NotFound, "Not Found args: %s", req.Name)
}

客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
_, err = cli.SayHello(context.Background(), &proto.HelloRequest{
		Name:    "Hello server! Rider calling",
		G:       proto.HelloRequest_MALE,
		AddTime: timestamppb.New(time.Now()),
	})
if err != nil {
		//错误处理
		st, ok := status.FromError(err)
		if !ok {
			// Error was not a status error
			panic("Error was not a status error")
		}
		fmt.Println(st.Message())
		fmt.Println(st.Code())
	}

超时机制

看到这儿我建议复习一下contex 在 **编程思想与并发.md **笔记中

场景：网络抖动，网络拥塞等，和数据库交互的服务可能非常慢这会导致服务之间的并发度降低

客户端

1
2
3
4
5
6
7
//设置超时上下文
ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
defer cancel()

rsp, err := cli.SayHello(ctx, &proto.HelloRequest{
	Name:    "Hello server! Rider calling",
})