第七章：完整代码示例 —— 一个城市调度中心的 gRPC 专线系统

syntax = "proto3";

package city.dispatch.v1;

service DispatchCenter {
  rpc GetRoute(RouteRequest) returns (RouteReply);
  rpc WatchRoad(RoadWatchRequest) returns (stream RoadEvent);
  rpc UploadSensors(stream SensorFrame) returns (UploadSummary);
  rpc ControlTunnel(stream ControlFrame) returns (stream ControlFrame);
}

message RouteRequest {
  string origin = 1;
  string destination = 2;
}

message RouteReply {
  string path = 1;
  int32 eta_minutes = 2;
}

message RoadWatchRequest {
  string road_id = 1;
}

message RoadEvent {
  string road_id = 1;
  string status = 2;
}

message SensorFrame {
  string device_id = 1;
  int32 cars = 2;
  double avg_speed = 3;
}

message UploadSummary {
  int32 frames = 1;
  double avg_speed = 2;
}

message ControlFrame {
  string lane_id = 1;
  string command = 2;
  string note = 3;
}

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

import time
from concurrent import futures

import grpc

import city_pb2
import city_pb2_grpc


class DispatchCenter(city_pb2_grpc.DispatchCenterServicer):
    def GetRoute(self, request, context):
        return city_pb2.RouteReply(
            path=f"{request.origin} -> 一号高架专线 -> {request.destination}",
            eta_minutes=3,
        )

    def WatchRoad(self, request, context):
        for status in ("畅通", "轻度拥堵", "施工绕行", "已恢复畅通"):
            yield city_pb2.RoadEvent(road_id=request.road_id, status=status)
            time.sleep(0.5)

    def UploadSensors(self, request_iterator, context):
        frames = 0
        total_speed = 0.0

        for frame in request_iterator:
            frames += 1
            total_speed += frame.avg_speed

        avg_speed = total_speed / frames if frames else 0.0
        return city_pb2.UploadSummary(frames=frames, avg_speed=avg_speed)

    def ControlTunnel(self, request_iterator, context):
        for frame in request_iterator:
            yield city_pb2.ControlFrame(
                lane_id=frame.lane_id,
                command="ACK",
                note=f"已收到指令：{frame.command}",
            )


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    city_pb2_grpc.add_DispatchCenterServicer_to_server(DispatchCenter(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print("gRPC server listening on :50051")

    try:
        while True:
            time.sleep(24 * 60 * 60)
    except KeyboardInterrupt:
        server.stop(5)


if __name__ == "__main__":
    serve()

import grpc

import city_pb2
import city_pb2_grpc


def sensor_frames():
    yield city_pb2.SensorFrame(device_id="cam-1", cars=12, avg_speed=48.5)
    yield city_pb2.SensorFrame(device_id="cam-2", cars=17, avg_speed=45.0)
    yield city_pb2.SensorFrame(device_id="cam-3", cars=9, avg_speed=52.3)


def tunnel_commands():
    for command in ("打开2号车道", "限制货车", "恢复常态"):
        yield city_pb2.ControlFrame(lane_id="tunnel-east", command=command)


with grpc.insecure_channel("localhost:50051") as channel:
    stub = city_pb2_grpc.DispatchCenterStub(channel)

    route = stub.GetRoute(
        city_pb2.RouteRequest(origin="税务局", destination="港口仓库"),
        timeout=1.5,
    )
    print("unary:", route.path, route.eta_minutes)

    print("server streaming:")
    for event in stub.WatchRoad(
        city_pb2.RoadWatchRequest(road_id="ring-1"),
        timeout=3,
    ):
        print(" -", event.road_id, event.status)

    summary = stub.UploadSensors(sensor_frames(), timeout=3)
    print("client streaming:", summary.frames, summary.avg_speed)

    print("bidirectional streaming:")
    for ack in stub.ControlTunnel(tunnel_commands(), timeout=3):
        print(" -", ack.lane_id, ack.command, ack.note)

city.proto
├── message
│   ├── RouteRequest / RouteReply
│   ├── RoadEvent
│   ├── SensorFrame / UploadSummary
│   └── ControlFrame
│
├── service
│   └── DispatchCenter
│
└── rpc
    ├── GetRoute
    ├── WatchRoad
    ├── UploadSensors
    └── ControlTunnel