在同一个 Python 服务器中结合 gRPC 标准和异步方法

Question

我有一个 gRPC 服务器，用 Python 3.6 编写。此服务器遵循 gRPC examples 中描述的模式，如下所示：

from concurrent import futures
import logging

import grpc

import helloworld_pb2
import helloworld_pb2_grpc


class Greeter(helloworld_pb2_grpc.GreeterServicer):

    def SayHello(self, request, context):
        return helloworld_pb2.HelloReply(message='Hello, %s!' % request.name)

    # ...

    def ReadRemoteData(self, request, context):
        return fetch_some_data_io_bound()


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()


if __name__ == '__main__':
    logging.basicConfig()
    serve()

在我的服务器应用程序中，有很多方法，但大体上是相同的。

目前有许多方法是我的服务器的瓶颈，我想通过使用async 运行它们来优化它们（这些方法在等待 IO 上花费了很多时间，因此它们是一个很好的候选者）。

最新版本的grpc 现在支持async，如this example 所示，通过使用server = grpc.aio.server() 创建异步服务器。

我的问题如下：

服务端有很多gRPC方法，其中很多都非常复杂。我想避免重写任何当前不是瓶颈的方法，并让它们保持原样。我只想重写那些将从异步实现中受益的方法，这只是总数的一小部分。出于向后兼容的原因，我无法更改 .proto 定义以将服务拆分为异步/非异步。

问题是，是否可以在同一个 gRPC 服务中以某种方式组合异步和非异步方法？

Answer 1

是的，这是可能的。应用程序需要创建带有额外参数migration_thread_pool 的异步服务器。这确实是迁移的必备功能。

请参阅https://grpc.github.io/grpc/python/grpc_asyncio.html#grpc.aio.server 上的文档

import asyncio
from concurrent import futures
import logging

import grpc

import helloworld_pb2
import helloworld_pb2_grpc


class Greeter(helloworld_pb2_grpc.GreeterServicer):

    def SayHello(self, request, context):
        return helloworld_pb2.HelloReply(message='Hello, %s!' % request.name)

    # ...

    def ReadRemoteData(self, request, context):
        return fetch_some_data_io_bound()


async def serve():
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))
    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
    server.add_insecure_port('[::]:50051')
    await server.start()
    await server.wait_for_termination()


if __name__ == '__main__':
    logging.basicConfig()
    asyncio.run(serve())