6.2 gRPC 合约、流式 RPC 与失败边界

gRPC 是远程过程调用框架，常用 Protocol Buffers 定义合约，并把一次调用尝试映射到一条 HTTP/2 流。网络细节被库封装起来，失败边界却不会消失：调用可能超时、只完成一部分，或者服务端已经成功而客户端没收到结果。

1. .proto 同时定义 数据与服务

syntax = "proto3";

package atlas.weather.v1;

service Weather {
  rpc Get(GetRequest) returns (Reading);
  rpc Watch(WatchRequest) returns (stream Reading);
  rpc Upload(stream Reading) returns (UploadSummary);
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message GetRequest {
  string city = 1;
}

message WatchRequest {
  string city = 1;
  uint32 count = 2;
}

message Reading {
  string city = 1;
  double celsius = 2;
  int64 observed_unix_seconds = 3;
}

message UploadSummary {
  uint32 accepted = 1;
}

message ChatMessage {
  string room = 1;
  string sender = 2;
  string text = 3;
}

四种 方法形态 是：

shape	request	response	常见用途
Unary	one	one	query / command
Server streaming	one	stream	watch、大 result set
Client streaming	stream	one	batch upload / aggregation
Bidirectional streaming	stream	stream	独立双向 event flow

字段编号属于二进制合约。删掉字段后用 reserved 保留编号和名字，避免旧数据被误读。新增字段则要检查旧客户端忽略它、新服务端收到默认值时的行为。“还能解码”只能证明一部分兼容性。proto3 普通标量未设置时使用默认值；若必须区分未传与显式零值，可选用 optional 等具备存在性语义的设计。

2. gRPC 如何使用 HTTP/2

下面讨论原生 gRPC over HTTP/2：一次调用尝试对应一条流，重试会开新流。多条流复用一条连接，既受各自流窗口限制，也共享连接级流控窗口。请求使用 POST，路径类似：

/atlas.weather.v1.Weather/Get

每条 gRPC 消息以 1 字节压缩标志和 4 字节大端长度开头，长度统计随后传输的消息字节，压缩时即压缩后的长度。HTTP/2 DATA 帧可以切分或合并这些字节，边界不等于 Protobuf 消息边界。最终状态通过 trailers 传递，也允许只发带状态的 trailers-only 响应：

grpc-status: 0
grpc-message: ...

只看 HTTP 状态 200 不足以判断 RPC 成功。客户端库结合 HTTP/2 传输状态、gRPC 状态和应用响应得出结果。代理需要支持 trailers、HTTP/2 和 gRPC 内容类型，否则可能出现 HTTP 请求通了，RPC 仍失败的情况。

例如未压缩的两个字节 08 01，加上 gRPC 前缀后是 00 00 00 00 02 08 01。这个前缀是 5 字节，不是 HTTP/2 的 9 字节帧头。HTTP 200 加 grpc-status: 3 表示 RPC 因 INVALID_ARGUMENT 失败；成功需要最终 gRPC 状态为 0。

多路复用消除了 HTTP/1.1 同一连接内响应必须依次返回的限制，但多条流仍共享 TCP 连接。丢包可能阻塞该连接后续字节的交付，因此 HTTP/2 并未消除所有层次的队头阻塞。

3. 把截止时间纳入 API 合约

客户端应给 RPC 设置截止时间。没有截止时间的调用可能一直等待，占住线程、连接容量和上游资源。

服务端需要：

• 查看剩余时间；
• 在客户端取消或超时后停止无意义的工作；
• 调用下游时传递或收紧截止时间，为清理工作保留预算；
• 不要把超时当成服务端一定没有执行操作的证明。

取消只是信号，不会回滚事务。有重要副作用的 RPC 需要幂等键、操作状态查询或补偿机制。

Python 同步处理器需要主动检查取消状态，或注册 context.add_callback 通知自己的任务停止；库不会替你终止任意计算。向下游传递预算时可读取 context.time_remaining()，扣除清理余量，预算已耗尽就不要再启动调用。具体语言是否自动传播 deadline 要查对应实现。

4. 状态码要保留语义

常见状态码边界：

• INVALID_ARGUMENT：请求本身无效，不会因系统状态变化而变对；
• FAILED_PRECONDITION：请求可能有效，但当前系统状态不允许；
• NOT_FOUND：指定实体不存在；
• ALREADY_EXISTS：待创建的目标已存在；
• PERMISSION_DENIED：调用者身份可识别但无权；
• UNAUTHENTICATED：缺少或无法验证有效身份凭据；
• RESOURCE_EXHAUSTED：配额或资源耗尽；
• UNAVAILABLE：暂时不可用，可能适合受控重试；
• DEADLINE_EXCEEDED：调用的截止时间已到；
• INTERNAL：内部不变量被破坏等严重错误，不是隐藏所有异常的兜底状态。

不要把所有异常都映射到 UNKNOWN 或 INTERNAL，也不要将 数据库异常原文 直接放进 grpc-message。对客户端公开稳定的状态码与结构化错误详情，把敏感诊断留在服务端日志或追踪中。

5. 重试必须与幂等性一起设计

重试可能来自客户端库、服务网格、代理或应用，多层叠加会放大负载。启用重试前先回答：

1. 方法重复执行是否安全？
2. 哪些状态码表示暂时故障？
3. 每次尝试的超时与整个调用的截止时间是多少？
4. 最多尝试几次，退避与抖动怎样设置？
5. 服务端如何用幂等键去重？

客户端没收到响应时，不能推断服务端尚未提交。创建订单等 RPC 应用由调用者生成的稳定请求 ID 建立去重合约，不能看到 UNAVAILABLE 就无条件重发。

gRPC 可以对确定未进入服务端应用的失败进行透明重试，但默认没有通用的状态码重试策略。收到响应头后，库把调用视为已提交，不再按该机制重试；这与数据库事务提交不是一回事。流式调用已经返回部分消息后，需要应用定义续传，不能期待重试机制自动拼接两次流。

6. 流式调用仍需限制缓冲

HTTP/2 和 gRPC 库提供流控，但应用仍需限制内存。流式处理器要考虑：

• 单条消息大小上限；
• 接收与发送队列上限；
• 慢读者如何向生产者施加背压；
• 取消时停止生产者；
• 部分进度能否恢复；
• 最终状态到达前，已收到的消息能否提交。

双向流式调用允许两个方向独立推进。库保持单次调用中各方向的消息顺序，但不会自动提供业务处理顺序、恰好一次交付或持久重放。

7. 可运行的 Unary + Server Streaming 最小项目

以下示例按 Python 3.12 与下列固定依赖验证。在一个空的实验目录中创建隔离环境：

python3 -m venv .venv
source .venv/bin/activate
python3 -m pip install grpcio==1.83.1 grpcio-tools==1.83.1 protobuf==7.36.1

将前面 .proto 缩减为实验所需的 weather.proto：

syntax = "proto3";
package atlas.weather.v1;

service Weather {
  rpc Get(GetRequest) returns (Reading);
  rpc Watch(WatchRequest) returns (stream Reading);
}

message GetRequest { string city = 1; }
message WatchRequest { string city = 1; uint32 count = 2; }
message Reading { string city = 1; double celsius = 2; }

生成 Python 代码：

python3 -m grpc_tools.protoc \
  -I. \
  --python_out=. \
  --grpc_python_out=. \
  weather.proto

server.py：

from concurrent import futures
import threading

import grpc
import weather_pb2
import weather_pb2_grpc

LIMITS = [("grpc.max_receive_message_length", 65536),
          ("grpc.max_send_message_length", 65536)]

class Weather(weather_pb2_grpc.WeatherServicer):
    def validate_city(self, request, context):
        if not request.city.strip() or len(request.city) > 100:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "city needs 1-100 characters")

    def Get(self, request, context):
        self.validate_city(request, context)
        return weather_pb2.Reading(city=request.city, celsius=21.5)

    def Watch(self, request, context):
        self.validate_city(request, context)
        count = request.count or 3
        if count > 10:
            context.abort(grpc.StatusCode.INVALID_ARGUMENT, "count must be 0-10")
        stopped = threading.Event()
        if not context.add_callback(stopped.set):
            return
        for index in range(count):
            if stopped.is_set() or not context.is_active():
                return
            yield weather_pb2.Reading(city=request.city, celsius=21.5 + index * 0.1)
            if index + 1 < count and stopped.wait(0.1):
                return

def serve():
    with futures.ThreadPoolExecutor(max_workers=8) as workers:
        server = grpc.server(workers, maximum_concurrent_rpcs=16, options=LIMITS)
        weather_pb2_grpc.add_WeatherServicer_to_server(Weather(), server)
        if server.add_insecure_port("127.0.0.1:50051") == 0:
            raise RuntimeError("failed to bind 127.0.0.1:50051")
        server.start()
        print("listening on 127.0.0.1:50051", flush=True)
        try:
            server.wait_for_termination()
        except KeyboardInterrupt:
            pass
        finally:
            server.stop(grace=1).wait()

if __name__ == "__main__":
    serve()

client.py：

客户端在流结束或提前退出时都调用 cancel() 释放未完成调用；正常完成后调用它无害。每个 RPC 分别有 2 秒预算，不是整个客户端共用 2 秒。

import grpc
import weather_pb2
import weather_pb2_grpc

options = [("grpc.max_receive_message_length", 65536),
           ("grpc.max_send_message_length", 65536)]
with grpc.insecure_channel("127.0.0.1:50051", options=options) as channel:
    stub = weather_pb2_grpc.WeatherStub(channel)
    try:
        current = stub.Get(weather_pb2.GetRequest(city="Nanjing"), timeout=2)
        print(f"Get: {current.city} {current.celsius:.1f}")
        stream = stub.Watch(weather_pb2.WatchRequest(city="Nanjing", count=3), timeout=2)
        try:
            for reading in stream:
                print(f"Watch: {reading.city} {reading.celsius:.1f}")
        finally:
            stream.cancel()
    except grpc.RpcError as error:
        print(f"RPC failed: {error.code().name}")
        raise SystemExit(1)
# Get: Nanjing 21.5
# Watch: Nanjing 21.5
# Watch: Nanjing 21.6
# Watch: Nanjing 21.7

服务端只监听 loopback，客户端也连接同一地址，用明文仅做本机实验。在两个已激活同一环境的终端分别运行 python3 server.py 和 python3 client.py。输出是一条当前读数和三条流式读数，均为固定模拟数据。服务端最多同时接纳 16 个 RPC、由 8 个工作线程执行；消息收发上限各为 64 KiB，城市名限 100 个字符，count=0 表示默认 3 条，超过 10 则报 INVALID_ARGUMENT。这限制的是该示例的应用负载，不能当成进程内存的硬上限。

取消回调会唤醒两次读数之间的等待，循环还会检查活动状态；取消发生在检查与 yield 之间时仍有竞态，不能承诺“取消瞬间绝不再产生一条消息”。生产环境还需要 TLS、身份认证、权限检查和可观察性。TLS 保护传输并验证配置的对端身份，不会自动授予方法调用权限。

8. 负载均衡与浏览器边界

gRPC channel 往往长时间复用连接。若负载均衡器只在建立 TCP 连接时选后端，后续 RPC 就不会按请求重新分配。可以采用理解 gRPC 的七层代理或客户端负载均衡，结合 xDS 配置分发，并配套健康检查、连接寿命和异常后端检测。

浏览器不直接暴露通用 HTTP/2 帧接口。gRPC-Web 通常通过兼容客户端和代理与原生 gRPC 后端交互；其流式能力、内容类型和传输行为不同，不能直接假设浏览器客户端支持原生双向流。

9. 验收问题

1. 为什么 HTTP 状态 200 不足以证明 gRPC 成功？
2. 区分截止时间耗尽、取消和事务回滚。
3. 设计一个可安全重试的创建操作，说明幂等键存在哪里。
4. 为服务端流式 RPC 定义消息上限、队列上限和慢读者策略。
5. 为什么 HTTP/2 的多路复用不能消除 TCP 丢包对整条连接的影响？

参考

• gRPC over HTTP/2 protocol

• gRPC Core concepts

• gRPC Deadlines

• gRPC Cancellation

• gRPC Retry

• Protocol Buffers language guide

• gRPC Python quick start

• gRPC status codes