grpc学习记录#

1
// 1使用哪种protobuf版本
2
syntax = "proto3";
3
// 2给包起一个名字
4
package test;
5
// 3服务器中服务的名字
6
service Bilibili {
7
    // 4里面写各种函数
8
    rpc HelloDewei(HelloDeweiReq) returns (HelloDeweiReply) {}
9
    // 7定义一个新的函数
10
    // 11将请求参数和相应参数改为流式传输
11
    rpc HelloTest(stream HelloTestRequest) returns (stream HelloTestResponse) {}
12
    // 12服务器以流的方式返回响应
13
    rpc TestClientRecvStream(TestClientRecvStreamRequest) returns (stream TestClientRecvStreamResponse) {}
14
}
15

16
// 5定义请求参数
17
message HelloDeweiReq {
18
    string name = 1;
19
    int32 age = 2;
20
}
21
// 6定义响应参数
22
message HelloDeweiReply {
23
    string result = 1;
24
}
25
// 8定义新的请求参数
26
message HelloTestRequest {
27
    string name = 1;
28
    int64 age = 2;
29
    repeated string data = 3;
30
    // 9定义map键值对的值的数据结构
31
    message HelloTestRequestNumberValue {
32
        string name = 1;
33
        int32 age = 2;
34
        bool is_active = 3;
35
    }
36
    map<string, HelloTestRequestNumberValue> number = 4; // string, int32, bool
37
}
38

39
// 10定义新的响应参数
40
message HelloTestResponse {}
41

42
// 13定义服务器流式向客户端返回响应的请求参数
43
message TestClientRecvStreamRequest {
44
    string data = 1;
45
}
46
// 14定义服务器流式向客户端返回响应的响应参数
47
message TestClientRecvStreamResponse {
48
    string result = 1;
49
}

1
# -*- coding: utf-8 -*-
2
# 0导包
3
import grpc
4
import hello_bilibili_pb2 as pb2
5
import hello_bilibili_pb2_grpc as pb2_grpc
6
# 8导入线程池所需要的包
7
from concurrent import futures
8
# 13导入阻塞需要的time模块
9
import time
10

11
# 1书写服务类，参数为生成的hello_bilibili_pb2_grpc文件里的BilibiliServicer
12
class Bilibili(pb2_grpc.BilibiliServicer):
13
    # 2实现服务类中的方法，request为请求参数，context为上下文
14
    def HelloDewei(self, request, context):
15
        # 3下面是请求参数中的字段
16
        name = request.name
17
        age = request.age
18
        # 4拼接返回值
19
        result = f'my name is {name}, my age is {age}'
20
        # 5返回这个返回值
21
        return pb2.HelloDeweiReply(result=result)
22
    # 15定义服务器流式向客户端传输的函数
23
    def TestClientRecvStream(self, request, context):
24
        index = 0
25
        # 16首先看客户端是否是活跃状态
26
        while context.is_active():
27
            data = request.data
28
            # 18服务器主动关闭向客户端输出
29
            if data == 'close':
30
                print('data is close, request will cancel')
31
                # 19关闭向客户端输出
32
                context.cancel()
33
            index += 1
34
            time.sleep(1)
35
            result = 'send %d %s' % (index, data)
36
            print(result)
37
            # 17流式向客户端返回
38
            yield pb2.TestClientRecvStreamResponse(
39
                result=result
40
            )
41

42
# 6编写服务启动函数
43
def run():
44
    # 7创建一个grpc服务器
45
    # grpc是支持线程的，所以这里需要创建一个线程池，这需要导包
46
    grpc_server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
47
    # 9将服务类注册到服务器，参数是Bilibili服务类和刚创建的grpc_server服务器
48
    pb2_grpc.add_BilibiliServicer_to_server(Bilibili(), grpc_server)
49
    # 10绑定ip和端口号
50
    grpc_server.add_insecure_port('0.0.0.0:50001')
51
    # 11启动服务器，注意启动服务器后要特殊处理，不然会启动一下就退出了
52
    grpc_server.start()
53
    # 12阻塞主线程
54
    try:
55
        while True:
56
            print('grpc_server is running')
57
            time.sleep(60 * 60 * 24)
58
    except KeyboardInterrupt:
59
        # 14如果按下ctrl+c，则正常停止服务器
60
        print('KeyboardInterrupt, grpc_server is stop')
61
        grpc_server.stop(0)
62

63
if __name__ == '__main__':
64
    run()

1
# -*- coding:utf-8 -*-
2
# 0导包
3
import grpc
4
import hello_bilibili_pb2 as pb2
5
import hello_bilibili_pb2_grpc as pb2_grpc
6

7
# 1写一个主函数
8
def run():
9
    # 2创建一个grpc的频道channel，绑定服务端的ip和端口号
10
    channel = grpc.insecure_channel('127.0.0.1:50001')
11
    # 3创建一个grpc的client绑定刚创建的频道channel
12
    client = pb2_grpc.BilibiliStub(channel=channel)
13
    # 4调用client中的方法，并传入参数
14
    # response = client.HelloDewei(pb2.HelloDeweiReq(
15
    #     name='dewei',
16
    #     age=18
17
    # ))
18
    # print(response.result)
19
    # 5向客户端发送请求
20
    response = client.TestClientRecvStream(pb2.TestClientRecvStreamRequest(
21
        data='dewei'
22
    ))
23
    # 6服务器流式向客户端返回
24
    for item in response:
25
        print(item.result)
26

27
if __name__ == '__main__':
28
    run()