Protocol Buffers gRPC 双向流

protobuf syntax = "proto3";

package chat;

// 聊天服务 service ChatService { // 双向流 RPC：客户端和服务器互相发送消息 rpc Chat(stream ChatMessage) returns (stream ChatMessage); }

// 聊天消息 message ChatMessage { string user = 1; string text = 2; int64 timestamp = 3; }

- `service ChatService` 定义了一个服务，其中包含 `Chat` 方法。
- `stream` 关键字出现在请求和响应前，表示该方法是双向流 RPC。
- `ChatMessage` 包含用户名称、文本内容和时间戳三个字段。

保存文件为 `chat.proto`，之后将通过 `protoc` 编译器生成对应语言的代码。在本教程中，我们将使用 Python 作为实现语言。

使用以下命令生成 Python 代码（需先安装 `grpcio-tools`）：

```bash
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. chat.proto

这会生成 chat_pb2.py（消息类）和 chat_pb2_grpc.py（客户端与服务器基类）。

服务器端实现

服务器需要继承生成文件中的 ChatServiceServicer 类，并实现 Chat 方法。该方法的参数是一个双向流上下文对象，我们可以通过它读取客户端发来的消息，同时向客户端发送消息。

import grpc
from concurrent import futures
import chat_pb2
import chat_pb2_grpc
import time

class ChatServicer(chat_pb2_grpc.ChatServiceServicer):
    def Chat(self, request_iterator, context):
        # request_iterator 是客户端发送的流
        for message in request_iterator:
            print(f"[{message.user}]: {message.text}")
            # 构造回复消息，回显并附加时间戳
            reply = chat_pb2.ChatMessage(
                user="Server",
                text=f"Echo: {message.text}",
                timestamp=int(time.time())
            )
            yield reply

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    chat_pb2_grpc.add_ChatServiceServicer_to_server(ChatServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

代码说明：

• Chat 方法的 request_iterator 是一个迭代器，客户端每发送一条消息，迭代器就会产出该消息。
• 方法内部使用了 yield，这使得 Chat 成为一个生成器函数，用于向客户端发送流式响应。
• 服务器每次收到客户端消息后，会打印到控制台，并生成一条回显消息发送给客户端。
• 在实际场景中，你也可以同时从其他数据源并发地向客户端推送消息（例如使用多线程），不受客户端发送流的限制。

客户端实现

客户端需要创建一个存根（stub），并调用 Chat 方法。该方法返回一个双向流对象，我们可以通过它同时发送和接收消息。通常需要用到多线程或异步处理，以避免发送和接收互相阻塞。

import grpc
import chat_pb2
import chat_pb2_grpc
import threading
import time

def read_responses(response_stream):
    """接收服务器返回的流式消息"""
    for response in response_stream:
        print(f"Received from server: {response.text}")

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = chat_pb2_grpc.ChatServiceStub(channel)
    # 调用双向流 RPC，获取一个流对象
    response_stream = stub.Chat(iter([...]))  # 实际使用时，iter 会在发送线程中提供消息

    # 更常见的做法：使用生成器作为请求流，并同时启动接收线程
    def request_generator():
        messages = [
            chat_pb2.ChatMessage(user="Alice", text="Hello"),
            chat_pb2.ChatMessage(user="Alice", text="How are you?"),
            chat_pb2.ChatMessage(user="Alice", text="Goodbye"),
        ]
        for msg in messages:
            print(f"Sending: {msg.text}")
            yield msg
            time.sleep(1)  # 模拟间隔

    # 启动接收线程
    response_stream = stub.Chat(request_generator())
    t = threading.Thread(target=read_responses, args=(response_stream,))
    t.start()
    t.join()

if __name__ == '__main__':
    run()

重要说明：

• stub.Chat() 的参数必须是 ChatMessage 的一个可迭代对象（如生成器），用于代表客户端要发送的消息流。
• 调用后返回的 response_stream 是一个迭代器，用于接收服务器发送的消息。
• 由于双向流是并发的，通常需要将读操作放入单独的线程，否则发送和接收会串行执行，无法体现“同时性”。
• 本例中，客户端依次发送三条消息，接收线程会打印服务器返回的回显。

运行与测试

1. 先启动服务器：

   python server.py
   

2. 再运行客户端：

   python client.py
   

服务器端控制台输出：

[Alice]: Hello
[Alice]: How are you?
[Alice]: Goodbye

客户端控制台输出：

Sending: Hello
Received from server: Echo: Hello
Sending: How are you?
Received from server: Echo: How are you?
Sending: Goodbye
Received from server: Echo: Goodbye