📚 课程目录第2章 gRPC基础回顾

2. gRPC 基础回顾：Protocol Buffers 定义与四种 RPC 模式

在深入事务边界与错误处理之前，必须牢固掌握 gRPC 的核心通信原语。本章从 Protocol Buffers（protobuf） 接口定义语言出发，系统回顾四种 RPC 模式的语法、适用场景与运行时行为。这些模式构成了后续所有分布式事务编排与异常传播的基础信道。

2.1 Protocol Buffers 服务定义

gRPC 使用 protobuf 作为接口定义语言（IDL）和默认序列化协议。一个典型的 .proto 文件包含消息类型和服务接口两部分。

  核心原则：服务定义即契约。客户端与服务器严格遵循 .proto 文件中声明的消息结构与 RPC 签名进行通信，任何字段编号、类型或方法签名的变更都视为破坏性变更。

syntax = "proto3";

package ecommerce.order;

// 订单服务：支持四种 RPC 模式
service OrderService {
  // 一元 RPC：创建订单
  rpc CreateOrder(CreateOrderRequest) returns (CreateOrderResponse);

  // 服务端流式 RPC：实时推送订单状态更新
  rpc SubscribeOrderStatus(SubscribeRequest) returns (stream OrderStatusEvent);

  // 客户端流式 RPC：批量提交订单项
  rpc BatchSubmitItems(stream OrderItem) returns (BatchSummary);

  // 双向流式 RPC：订单协商与确认
  rpc NegotiateOrder(stream NegotiateMessage) returns (stream NegotiateMessage);
}

// 消息定义
message CreateOrderRequest {
  string user_id = 1;
  repeated OrderItem items = 2;
  double total_amount = 3;
}

message CreateOrderResponse {
  string order_id = 1;
  int32 http_status_code = 2;
  string error_message = 3;
}

message OrderItem {
  string sku = 1;
  int32 quantity = 2;
  double unit_price = 3;
}

message OrderStatusEvent {
  string order_id = 1;
  string new_status = 2;
  int64 timestamp_unix_ms = 3;
}

message SubscribeRequest {
  string order_id = 1;
}

message BatchSummary {
  int32 total_items = 1;
  int32 accepted_count = 2;
  repeated string failed_skus = 3;
}

message NegotiateMessage {
  string role = 1;       // "client" 或 "server"
  string action = 2;     // "propose", "counter", "accept", "reject"
  string payload = 3;
}

protobuf 字段规则与 gRPC 映射
规则	说明	gRPC 序列化行为
`singular`	默认规则，字段出现 0 或 1 次	未设置时使用零值，不占用线缆字节
`repeated`	有序列表，可包含 0 到 N 个元素	流式场景下常用于批量数据块
`map<K,V>`	无序键值对，键类型必须为标量	适合传递元数据或配置快照
`oneof`	一组字段中最多同时设置一个	用于互斥的请求变体，减少消息膨胀

2.2 一元 RPC（Unary RPC）

客户端发送单个请求，服务器返回单个响应。这是最基础的 RPC 模式，语义与传统的 HTTP 请求-响应类似，但基于 HTTP/2 和 protobuf 二进制帧，延迟更低、吞吐更高。

事务边界提示：一元 RPC 天然适合“请求-确认”型事务操作（如创建订单、支付扣款）。客户端在收到响应后即可判断操作是否成功，错误处理直接映射为 gRPC 状态码。

// Python 客户端示例
import grpc
import order_pb2
import order_pb2_grpc

channel = grpc.insecure_channel('localhost:50051')
stub = order_pb2_grpc.OrderServiceStub(channel)

request = order_pb2.CreateOrderRequest(
    user_id="user_001",
    items=[order_pb2.OrderItem(sku="SKU-1001", quantity=2, unit_price=49.99)],
    total_amount=99.98
)

try:
    response = stub.CreateOrder(request, timeout=5.0)
    print(f"Order created: {response.order_id}")
except grpc.RpcError as e:
    print(f"gRPC error: {e.code()} - {e.details()}")

2.3 服务端流式 RPC（Server Streaming RPC）

客户端发送一个请求，服务器返回一个消息流。客户端通过迭代读取流中的消息，直到流关闭。适用于事件推送、日志尾随、大数据集分页等场景。

// Go 服务端实现片段
func (s *OrderServer) SubscribeOrderStatus(req *pb.SubscribeRequest, stream pb.OrderService_SubscribeOrderStatusServer) error {
    // 模拟状态变更推送
    statuses := []string{"PENDING", "CONFIRMED", "SHIPPING", "DELIVERED"}
    for _, st := range statuses {
        event := &pb.OrderStatusEvent{
            OrderId:       req.OrderId,
            NewStatus:     st,
            TimestampUnixMs: time.Now().UnixMilli(),
        }
        if err := stream.Send(event); err != nil {
            return err  // 客户端断开或网络异常
        }
        time.Sleep(500 * time.Millisecond)
    }
    return nil  // 正常结束流
}

错误处理要点：服务端流中，错误可以在任意 Send 调用后发生。服务器应使用 stream.Context().Err() 检查客户端是否已取消；客户端应使用 for response in stub.SubscribeOrderStatus(...): 循环捕获 grpc.RpcError。

2.4 客户端流式 RPC（Client Streaming RPC）

客户端发送一个消息流，服务器在接收完所有消息后返回单个响应。常用于批量上传、文件分块提交、传感器数据聚合等场景。

# Python 客户端批量提交
def generate_items():
    items_data = [
        ("SKU-2001", 5, 19.99),
        ("SKU-2002", 3, 29.99),
        ("SKU-2003", 0, 9.99),  # 数量为 0，服务端可拒绝
    ]
    for sku, qty, price in items_data:
        yield order_pb2.OrderItem(sku=sku, quantity=qty, unit_price=price)

stub = order_pb2_grpc.OrderServiceStub(channel)
summary = stub.BatchSubmitItems(generate_items())
print(f"Accepted: {summary.accepted_count}/{summary.total_items}")

  事务边界挑战：客户端流式 RPC 中，服务器直到收到 stream.Close() 信号才开始处理完整数据集。若流中途中断，服务器可能已处理部分数据但未提交。工程上常采用“先暂存后批量提交”的两阶段模式来保证原子性。

2.5 双向流式 RPC（Bidirectional Streaming RPC）

客户端和服务器各自独立发送和接收消息流。两者之间的读写是全双工的，没有固定的请求-响应顺序。适用于实时聊天、游戏状态同步、分布式协商协议等。

// Go 双向流协商示例
func (s *OrderServer) NegotiateOrder(stream pb.OrderService_NegotiateOrderServer) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil  // 客户端关闭写端
        }
        if err != nil {
            return err
        }

        // 根据客户端消息生成服务器响应
        response := &pb.NegotiateMessage{
            Role:    "server",
            Action:  "counter",
            Payload: fmt.Sprintf("Server counter for: %s", msg.Payload),
        }
        if err := stream.Send(response); err != nil {
            return err
        }
    }
}

协同工程价值：双向流是构建分布式事务协调器（如 Saga 执行器、两阶段提交协调者）的理想信道。协调者可以同时向多个参与者发送指令，并实时接收每个参与者的反馈，无需反复建立连接。

2.6 四种 RPC 模式对比

四种 RPC 模式特性矩阵
模式	请求方向	响应方向	典型延迟	事务适用性
一元 RPC	1 条消息	1 条消息	低（一次往返）	简单提交/确认
服务端流式	1 条消息	N 条消息	中（持续推送）	状态订阅/事件溯源
客户端流式	N 条消息	1 条消息	中（批量后响应）	批量提交/聚合
双向流式	N 条消息	N 条消息	可变（全双工）	协调/协商/实时同步

2.7 本章小结

Protocol Buffers 是 gRPC 的契约基础，消息字段编号、类型和 repeated/map/oneof 规则直接影响序列化效率与兼容性。
一元 RPC 提供最直观的请求-响应模型，适合事务提交与确认。
服务端流式 RPC 适用于事件推送与状态订阅，服务器可在任意时刻发送消息。
客户端流式 RPC 适用于批量数据上传，服务器在流关闭后统一处理。
双向流式 RPC 实现全双工通信，是分布式事务协调与实时协商的核心信道。
选择 RPC 模式时，应基于请求/响应数量比和实时性需求进行决策，避免将流式模式用于简单的一元交互，反之亦然。