这是我的第399篇原创文章。

一、引言

以下是一个完整的双向gRPC流式传输的Python示例。这个示例包括一个简单的服务端和客户端，服务端接收客户端发送的流式数据，处理后返回流式响应。我们将通过一个简单的“回声服务”来演示这个过程，即服务端将客户端发送的每条消息原样返回。

二、实现过程

2.1 定义gRPC服务接口

首先，需要定义一个ProtoBuf文件来描述服务接口和消息格式。创建一个名为echo.proto的文件，内容如下：

syntax = "proto3";


package echo;


// 定义服务接口
service EchoService {
    rpc BidirectionalStream(stream EchoRequest) returns (stream EchoResponse);
}


// 请求消息格式
message EchoRequest {
    string message = 1;
}


// 响应消息格式
message EchoResponse {
    string message = 1;
}

然后，使用grpcio-tools生成Python代码：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. echo.proto

这将生成echo_pb2.py和echo_pb2_grpc.py文件，分别包含消息类和服务类的定义。

2.2 实现服务端

服务端代码负责接收客户端发送的流式数据，并将每条消息原样返回。以下是服务端的完整实现：

# server.py
from concurrent import futures
import grpc
import echo_pb2
import echo_pb2_grpc


class EchoService(echo_pb2_grpc.EchoServiceServicer):
    def BidirectionalStream(self, request_iterator, context):
        """
        处理双向流式请求
        """
        for request in request_iterator:
            print(f"Received message: {request.message}")
            # 将客户端发送的消息原样返回
            yield echo_pb2.EchoResponse(message=request.message)


def serve():
    # 创建gRPC服务器
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    echo_pb2_grpc.add_EchoServiceServicer_to_server(EchoService(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print("Echo Service started on port 50051.")
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

2.3 实现客户端

客户端代码负责发送流式数据，并接收服务端返回的流式响应。以下是客户端的完整实现：

#.py client
import grpc
import echo_pb2
import echo_pb2_grpc


def generate_messages():
    """
    生成流式请求数据
    """
    messages = ["Hello", "World", "This", "is", "a", "bidirectional", "stream"]
    for msg in messages:
        yield echo_pb2.EchoRequest(message=msg)


def run():
    # 连接到服务端
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = echo_pb2_grpc.EchoServiceStub(channel)
        # 调用双向流式接口
        responses = stub.BidirectionalStream(generate_messages())
        for response in responses:
            print(f"Received response: {response.message}")


if __name__ == "__main__":
    run()

2.4 运行和测试

启动服务端:

python server.py

服务端将在localhost:50051上启动并监听客户端连接。

启动客户端：

python client.py

客户端将发送一系列消息到服务端，并接收服务端返回的响应。

服务端输出：

Echo Service started on port 50051.
Received message: Hello
Received message: World
Received message: This
Received message: is
Received message: a
Received message: bidirectional
Received message: stream

客户端输出：

Received response: Hello
Received response: World
Received response: This
Received response: is
Received response: a
Received response: bidirectional
Received response: stream

说明：

1. 双向流式传输：客户端和服务端都可以同时发送和接收消息，非常适合实时需要交互的场景。
2. 生成消息：客户端通过generate_messages函数生成流式请求数据，每次yield返回一条消息。
3. 处理响应：服务端通过for request in request_iterator逐条处理客户端发送的消息，并通过yield返回响应。

在grpc的流式通信中，一个Request和一个Response并不一定是一一对应的，特别是在使用双向流式的RPC时，在双向流式RPC中，客户端可以连续发送多个Request消息，服务端可以发送多个Response消息。也就是说服务端的响应并不一定与客户端的请求完全同步，服务器可能在接收到多个请求后才开始发送响应或者接收到一个请求后发送多个响应。在流式RPC中是异步的，他们之间的关系取决于具体的实现逻辑。

作者简介： 读研期间发表6篇SCI数据算法相关论文，目前在某研究院从事数据算法相关研究工作，结合自身科研实践经历持续分享关于Python、数据分析、特征工程、机器学习、深度学习、人工智能系列基础知识与案例。关注gzh：数据杂坛，获取数据和源码学习更多内容。

原文链接：

「链接」