gRPC Client Streaming到服务端Python代码实现失败报错求排查

• 1. proto文件定义与服务端、客户端实现完全不匹配
• 2. 方法调用逻辑混淆，把客户端流的请求传给了仅支持单参数的服务端流方法
• 3. 字段取值错误，直接拼接protobuf对象而不是取对象内的message属性

第一步：修正proto文件定义 #

你原来的proto中两个方法的出入参不符合业务需求，调整如下，确保返回类型匹配你定义的MessageResponse：

syntax = "proto3";

package streaming;

service Streaming{
  // 服务端流：客户端发单个消息，服务端返回流
  rpc ServerStreaming (Message) returns (stream MessageResponse) {}
  // 客户端流：客户端发流，服务端返回单个消息
  rpc ClientStreaming (stream Message) returns (MessageResponse) {}
}

message Message{
  string message = 1;
}

message MessageResponse{
  string message = 1;
  bool received = 2;
}

修改完成后重新生成Python对应的pb2和pb2_grpc文件，该步骤不能省略。

第二步：修正服务端代码 #

原来的服务端错误地把客户端流的逻辑写到了服务端流方法中，同时存在字段取值错误，修正后代码如下：

from concurrent import futures
import grpc
import streaming_pb2_grpc as streaming_pb2_grpc
import streaming_pb2 as streaming_pb2


def make_message(message):
    return streaming_pb2.Message(
        message=message
    )


class StreamingService(streaming_pb2_grpc.StreamingServicer):
    # 服务端流实现：接收单个请求，返回流
    def ServerStreaming(self, request, context):
        # 示例：返回5条流消息
        for i in range(5):
            yield streaming_pb2.MessageResponse(
                message=f"服务端流返回第{i+1}条消息，收到你的消息：{request.message}",
                received=True
            )

    # 客户端流实现：接收流请求，返回单个响应
    def ClientStreaming(self, request_iterator, context):
        message_full = ''
        for message in request_iterator:
            # 取message对象的message属性，不能直接拼接protobuf对象
            message_full += message.message + " "
        result = f'Hello I am up and running received "{message_full.strip()}" message from you'
        return streaming_pb2.MessageResponse(message=result, received=True)


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    streaming_pb2_grpc.add_StreamingServicer_to_server(StreamingService(), server)
    server.add_insecure_port('[::]:8091')
    server.start()
    server.wait_for_termination()


if __name__ == '__main__':
    serve()

第三步：修正客户端代码 #

原来的客户端调用服务端流方法时错误传入了流参数，调整为调用客户端流方法即可：

import grpc
import streaming_pb2_grpc as streaming_pb2_grpc
import streaming_pb2 as streaming_pb2


def make_message(message):
    return streaming_pb2.Message(
        message=message
    )


def generate_messages():
    messages = [
        make_message("First message"),
        make_message("Second message"),
        make_message("Third message"),
        make_message("Fourth message"),
        make_message("Fifth message"),
    ]
    for msg in messages:
        print("向服务端发送：%s" % msg.message)
        yield msg


def send_client_stream(stub):
    # 调用客户端流方法，传入消息生成器，接收单个返回
    response = stub.ClientStreaming(generate_messages())
    print("收到服务端响应：%s，接收状态：%s" % (response.message, response.received))

# 测试服务端流的方法可按需调用
def send_server_stream(stub):
    # 服务端流入参是单个消息
    response_stream = stub.ServerStreaming(make_message("测试服务端流"))
    for response in response_stream:
        print("收到服务端流消息：%s，接收状态：%s" % (response.message, response.received))


def run():
    with grpc.insecure_channel('localhost:8091') as channel:
        stub = streaming_pb2_grpc.StreamingStub(channel)
        # 测试客户端流
        send_client_stream(stub)
        # 如果要测试服务端流取消注释下方代码即可
        # send_server_stream(stub)


if __name__ == '__main__':
    run()

你遇到的Exception serializing request报错，本质是因为你调用的ServerStreaming方法在proto中定义的入参是单个Message对象，但你却传入了消息生成器（流结构），gRPC无法将生成器序列化为单个Message对象，因此抛出序列化错误。

内容的提问来源于stack exchange，提问作者Pavel Zagalsky