python asyncio grpc

1. 准备环境

python3.11 -m venv venv
source venv/*/activate
pip install grpcio-tools #包含了grpcio和protobuf
pip install types-protobuf grpc-stubs # 可选安装，用于mypy静态检查

2. 编写msg.proto

syntax = "proto3";

// 这是注释，同时也是类文档
service MsgService {
 rpc handler (MsgRequest) returns (MsgResponse){}
}

// 这也是注释
message MsgRequest {
  // 1,2,3...是字段编号，正整数就行，可以不连续
  string name = 1;  // 姓名
  optional uint32 age = 2;  // 年龄
  optional float high = 3;  // 身高
  optional bytes avatar = 4;  // 头像
}


message MsgResponse {  // 注释也可以在行尾
  uint64 id = 1; // ID
  Role role = 2;  // 角色
  optional uint64 last_login = 10;  // 上一次登陆的时间戳
}

// 角色（嵌套字段）
message Role {
  string name = 1;
  int32 level = 2;
}

3. 把proto编译成python文件

python -m grpc_tools.protoc -I . --python_out=. --grpc_python_out=. msg.proto
ls msg_pb2*.py

4. 服务端程序msg_server.py

参考了官方示例：https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_server_with_graceful_shutdown.py

#!/usr/bin/env python3
import asyncio

import grpc

import msg_pb2
import msg_pb2_grpc

try:
    from rich import print
except ImportError:
    ...


class MsgServicer(msg_pb2_grpc.MsgServiceServicer):
    def handler(self, request: "msg_pb2.MsgRequest", context) -> "msg_pb2.MsgResponse":
        print("Received name: %s" % request.name)
        # 响应的处理逻辑写在这里
        # ...
        role = {'name': request.name, 'level': 0}
        return msg_pb2.MsgResponse(role=role, id=1)


def serve() -> None:
    _cleanup_coroutines = []

    async def run() -> None:
        server = grpc.aio.server()
        msg_pb2_grpc.add_MsgServiceServicer_to_server(MsgServicer(), server)
        listen_addr = "[::]:50051"
        server.add_insecure_port(listen_addr)
        print(f"Starting server on {listen_addr}")
        await server.start()

        async def server_graceful_shutdown():
            print("Starting graceful shutdown...")
            # Shuts down the server with 5 seconds of grace period. During the
            # grace period, the server won't accept new connections and allow
            # existing RPCs to continue within the grace period.
            await server.stop(5)
            print(f"{server} was graceful shutdown~")

        _cleanup_coroutines.append(server_graceful_shutdown())
        await server.wait_for_termination()

    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(run())
    finally:
        loop.run_until_complete(*_cleanup_coroutines)
        loop.close()


if __name__ == "__main__":
    serve()

5. 启动服务

python msg_server.py
# Starting server on [::]:50051

6. 客户端代码msg_client.py

import asyncio
import os

import grpc

import msg_pb2
import msg_pb2_grpc

try:
    from rich import print
except ImportError:
    ...


def main():
    async def run() -> None:
        host = os.getenv("RPC_HOST", "localhost")
        async with grpc.aio.insecure_channel(f"{host}:50051") as channel:
            stub = msg_pb2_grpc.MsgServiceStub(channel)

            response = await stub.handler(msg_pb2.MsgRequest(name="you"))
            print("Client received: ")
            print(response)

    asyncio.run(run())


if __name__ == "__main__":
    main()

7. 运行客户端

python msg_client.py

结果如下：

Client received:
id: 1
role {
  name: "you"
}