如何异步处理双向 grpc 流

Question

我有一个游戏或任何带有服务器和多个客户端的远程用户界面，它们应该通过网络进行通信。 客户端和服务器都应该能够异步发送更新。

这似乎是一个非常自然的服务定义，让 grpc 管理会话。

syntax = "proto3";

package mygame;

service Game {
    rpc participate(stream ClientRequest) returns (ServerResponse);
}

message ClientRequest {
    // Fields for the initial request and further updates
}

message ServerResponse {
    // Game updates
}

实现客户端很简单（尽管下面的代码显然是不完整和简化的）。

syntax = "proto3";

package mygame;

service Game {
    rpc participate(stream ClientRequest) returns (ServerResponse);
}

message ClientRequest {
    // Fields for the initial request and further updates
}

message ServerResponse {
    // Game updates
}

看起来困难的是在不阻塞的情况下实现服务器。

class Client:
    def __init__(self):
        self.channel = grpc.insecure_channel("localhost:50051")
        self.stub = game_pb2_grpc.GameStub(channel)
        self.output_queue = queue.Queue()

    def output_iter(self):
        while True:
            client_output_msg = self.output_queue.get()
            self.output_queue.task_done()
            yield client_output_msg

    def do_work(self):
        for response in self.stub.participate(self.output_iter()):
            print(response)  # handle update


with grpc.insecure_channel("localhost:50051") as channel:
    client = Client()
    client.do_work()

正如代码中所注释的，如果客户端不不断发送请求，则不会收到更新。也许异步方法会更好，这也可能解决此设计中的其他问题。

PS：这个问题已经通过 go here中的 grpc 解决了，但是我不知道如何将其转换为 python grpc 实现。

如果有任何帮助，我将非常高兴！

Answer 1

I was finally able to get it working using the python asynio api. The basic idea is to decouple read and write into two coroutines using asyncio.create_task. For anybody interested, here is a solution.

class Game(game_pb2_grpc.GameServicer):
    async def read_client_requests(self, request_iter):
        async for client_update in request_iter:
            print("Recieved message from client:", client_update, end="")

    async def write_server_responses(self, context):
        for i in range(15):
            await context.write(game_pb2.ServerResponse(dummy_value=str(i)))
            await asyncio.sleep(0.5)

    async def participate(self, request_iter, context):
        read_task = asyncio.create_task(self.read_client_requests(request_iter))
        write_task = asyncio.create_task(self.write_server_responses(context))

        await read_task
        await write_task


async def serve():
    server = grpc.aio.server()
    game_pb2_grpc.add_GameServicer_to_server(Game(), server)
    server.add_insecure_port("[::]:50051")
    await server.start()
    await server.wait_for_termination()


if __name__ == "__main__":
    asyncio.run(serve())

Note that instead of the write coroutine, a yield would also be sufficient.