如何使用Python gRPC处理流消息

pip install grpcio grpcio-tools

syntax = "proto3";

package stream_example;

// 定义请求消息
message RequestMessage {
    string data = 1;
}

// 定义响应消息
message ResponseMessage {
    string result = 1;
}

// 定义服务，包含四种流模式
service StreamService {
    // 简单RPC：客户端发送一个请求，服务器返回一个响应
    rpc SimpleRPC (RequestMessage) returns (ResponseMessage);

    // 服务器流RPC：客户端发送一个请求，服务器返回多个响应
    rpc ServerStreamingRPC (RequestMessage) returns (stream ResponseMessage);

    // 客户端流RPC：客户端发送多个请求，服务器返回一个响应
    rpc ClientStreamingRPC (stream RequestMessage) returns (ResponseMessage);

    // 双向流RPC：客户端和服务器都可以发送多个请求和响应
    rpc BidirectionalStreamingRPC (stream RequestMessage) returns (stream ResponseMessage);
}

python -m grpc_tools.protoc \
    -I. \
    --python_out=. \
    --grpc_python_out=. \
    stream_example.proto

import grpc
from concurrent import futures
import time
import stream_example_pb2
import stream_example_pb2_grpc

class StreamServiceServicer(stream_example_pb2_grpc.StreamServiceServicer):
    def SimpleRPC(self, request, context):
        response = stream_example_pb2.ResponseMessage(result=f"Received: {request.data}")
        return response

    def ServerStreamingRPC(self, request, context):
        for i in range(5):
            yield stream_example_pb2.ResponseMessage(result=f"Stream message {i} for {request.data}")

    def ClientStreamingRPC(self, request_iterator, context):
        total = 0
        for req in request_iterator:
            total += len(req.data)
        response = stream_example_pb2.ResponseMessage(result=f"Total length: {total}")
        return response

    def BidirectionalStreamingRPC(self, request_iterator, context):
        for req in request_iterator:
            yield stream_example_pb2.ResponseMessage(result=f"Echo: {req.data}")

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    stream_example_pb2_grpc.add_StreamServiceServicer_to_server(StreamServiceServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server started, listening on 50051")
    try:
        while True:
            time.sleep(86400)
    except KeyboardInterrupt:
        server.stop(0)

if __name__ == '__main__':
    serve()

python server.py

import grpc
import stream_example_pb2
import stream_example_pb2_grpc

def run_simple_rpc():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = stream_example_pb2_grpc.StreamServiceStub(channel)
        response = stub.SimpleRPC(stream_example_pb2.RequestMessage(data="Hello"))
        print("SimpleRPC received:", response.result)

def run_server_streaming_rpc():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = stream_example_pb2_grpc.StreamServiceStub(channel)
        responses = stub.ServerStreamingRPC(stream_example_pb2.RequestMessage(data="Stream"))
        for response in responses:
            print("ServerStreamingRPC received:", response.result)

def run_client_streaming_rpc():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = stream_example_pb2_grpc.StreamServiceStub(channel)
        requests = (stream_example_pb2.RequestMessage(data=f"Message {i}") for i in range(5))
        response = stub.ClientStreamingRPC(iter(requests))
        print("ClientStreamingRPC received:", response.result)

def run_bidirectional_streaming_rpc():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = stream_example_pb2_grpc.StreamServiceStub(channel)
        def generate_requests():
            for i in range(5):
                yield stream_example_pb2.RequestMessage(data=f"Bidirectional {i}")
                time.sleep(1)
        
        responses = stub.BidirectionalStreamingRPC(generate_requests())
        for response in responses:
            print("BidirectionalStreamingRPC received:", response.result)

if __name__ == '__main__':
    run_simple_rpc()
    run_server_streaming_rpc()
    run_client_streaming_rpc()
    run_bidirectional_streaming_rpc()

python client.py