tonic 与 gRPC

gRPC 用服务定义描述 RPC 方法和消息类型，tonic 提供 Rust 的异步实现，常见组合是 Tokio、Tower 与 Prost。生成代码减少手写序列化和接口样板，但网络两端不会因为一方修改 .proto 就自动重新编译。

协议的编译期类型检查、线上编码兼容和业务语义是三个层次。HTTP API 同样可以使用 OpenAPI 等描述与代码生成；选择 gRPC 的理由应来自调用模型、流式能力及生态要求，而不是把另一类接口一概说成没有合约。

先理解四种 RPC

双向流不要求每收到一条请求才发一条响应，也不保证两个方向存在一一对应关系。每个方向内的消息有顺序，应用仍需定义消息 ID、完成条件与失败后的恢复方式。

最小完整工程：明确生成工具版本

以下使用 tonic 0.14 系列。该系列的 Prost 集成位于 tonic-prost 与 tonic-prost-build；不要直接混用旧教程的 tonic_build::compile_protos。需要本机安装 protoc，并保存 Cargo.lock 以记录实际解析版本。tonic、tonic-prost-build

Cargo.toml：

[package]
name = "grpc-demo"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tonic = "0.14"
tonic-prost = "0.14"
prost = "0.14"

[build-dependencies]
tonic-prost-build = "0.14"

proto/greeter.proto：

syntax = "proto3";
package greeter;
service Greeter {
  rpc SayHello(HelloRequest) returns (HelloReply);
}
message HelloRequest { string name = 1; }
message HelloReply { string message = 1; }

build.rs：

fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("cargo:rerun-if-changed=proto/greeter.proto");
    tonic_prost_build::configure()
        .compile_protos(&["proto/greeter.proto"], &["proto"])?;
    Ok(())
}

src/lib.rs：

pub mod greeter {
    tonic::include_proto!("greeter");
}

src/bin/server.rs：

use grpc_demo::greeter::{HelloReply, HelloRequest};
use grpc_demo::greeter::greeter_server::{Greeter, GreeterServer};
use tonic::{Request, Response, Status};

#[derive(Default)]
struct HelloService;

#[tonic::async_trait]
impl Greeter for HelloService {
    async fn say_hello(&self, request: Request<HelloRequest>)
        -> Result<Response<HelloReply>, Status>
    {
        let name = request.into_inner().name;
        if name.trim().is_empty() {
            return Err(Status::invalid_argument("name is required"));
        }
        Ok(Response::new(HelloReply { message: format!("Hello, {name}!") }))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic::transport::Server::builder()
        .add_service(GreeterServer::new(HelloService))
        .serve("127.0.0.1:50051".parse()?)
        .await?;
    Ok(())
}

src/bin/client.rs：

use grpc_demo::greeter::{HelloRequest, greeter_client::GreeterClient};
use std::time::Duration;
use tonic::Request;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = GreeterClient::connect("http://127.0.0.1:50051").await?;
    let mut request = Request::new(HelloRequest { name: "Rust".into() });
    request.set_timeout(Duration::from_secs(2));
    let reply = client.say_hello(request).await?.into_inner();
    println!("{}", reply.message);
    Ok(())
}

分别在两个终端运行 cargo run --bin server 和 cargo run --bin client，客户端应打印 Hello, Rust!。此处本地明文监听用于观察请求链路；真实部署根据环境配置 TLS、身份验证及资源限制。

#[tonic::async_trait] 用于匹配该系列生成的 trait 形式。Rust 从 1.75 已支持 trait 中的原生 async fn，保留宏不能解释为语言“尚不支持”；原生 async trait、动态分派与生成库采用的约定需要分别看待。

修改 .proto 为什么不一定破坏旧客户端

Protobuf 二进制编码依赖字段号。增加兼容字段时，旧客户端仍可能正常读取自己认识的字段；本地 Rust 代码是否需要修改，取决于是否重新生成及如何构造类型。线上兼容不能由单个仓库编译通过证明。

删除字段后应保留其编号，通常也保留名字以防将来误用；不要复用旧编号表达不同含义。字段类型变化要按 wire 兼容规则逐项判断。即使编码仍可解码，单位从秒改成毫秒也可能直接破坏业务语义。Proto3 更新规则

需要区分“没有提供”与“提供了默认值”时，应明确使用字段存在性设计，例如适用的 optional 字段。空字符串并不会因为类型是 string 就自动变成验证错误，所以上面 handler 显式检查 name。

用两份消息类型验证字段存在性与旧中转节点

下面是独立的 src/bin/compat.rs，使用工程已有的 Prost 依赖，直接声明消息以便在同一个程序中同时观察新旧结构。字段号 1 保留 name，版本二增加 optional 的 locale 字段号 2。

use prost::Message;
#[derive(Clone, PartialEq, Message)]
struct V1 {
    #[prost(string, tag = "1")]
    name: String,
}
#[derive(Clone, PartialEq, Message)]
struct V2 {
    #[prost(string, tag = "1")]
    name: String,
    #[prost(string, optional, tag = "2")]
    locale: Option<String>,
}
fn main() {
    let absent = V2 { name: String::new(), locale: None };
    let empty = V2 { name: String::new(), locale: Some(String::new()) };
    assert!(absent.encode_to_vec().is_empty());
    assert_eq!(empty.encode_to_vec(), [0x12, 0]);

    let new = V2 { name: "Rust".into(), locale: Some("zh-CN".into()) };
    let bytes = new.encode_to_vec();
    let old = V1::decode(bytes.as_slice()).unwrap();
    assert_eq!(old.name, "Rust");
    let relayed = old.encode_to_vec();
    let decoded = V2::decode(relayed.as_slice()).unwrap();
    assert_eq!(decoded.name, "Rust");
    assert_eq!(decoded.locale, None);
}

optional 将缺失和显式空值映射到 None 与 Some("")。本例中 0x12 是字段号 2、wire type 2 的键，后面的 0 表示零字节长度；因此两种状态即使业务字符串同样为空，线上表示也不同。Prost 字段表示

旧 V1 可以读取新数据中认识的 name，但这个 Prost 消息结构不会在重新编码时保留未知 locale。于是“旧客户端能读取新增字段的报文”与“经过旧节点解码再编码后，新字段仍能到达终点”是两种兼容要求。如果旧节点只是透传原始消息字节，则没有发生这里的有损重建；具体链路要按实际转换过程验证。

发布前按读写方向检查

逐步发布通常先让读取方接受新形式，再启用新写入；回滚还需检查已经写下的新数据是否能被旧版本合理处理。保留字段号避免编码误读，却不会自动解决单位、权限或状态机语义改变。

旧节点读得懂，也能保留新字段吗？

比较两个只有细微差别的报文：None 根本不发送字段 2，Some("") 则发送字段键和零长度。原始转发能保留这一差别；本文的旧 Prost V1 只认识 name，重建消息后就丢掉 locale。兼容性因此需要检查整条读写路径，而不只是旧端能否解码。

Status、deadline 与重试

Status 表示 RPC 结果，不能只看 HTTP 响应状态。常用分类如下，完整代码定义见 gRPC Status Codes：

客户端应有截止时间，总预算要覆盖排队、连接、处理与可能的重试。Request::set_timeout 写入 grpc-timeout metadata，和仅在本地设定传输超时不是同一个操作；上面的设置发生在 connect 之后，因此没有给建连本身设置两秒上限。

服务取消后，应用还需停止自己创建的后台工作。向下游继续发 RPC 时应传播剩余预算，而不是每层重新给完整超时；不同语言和库对传播的支持不同，需明确验证。gRPC Deadlines

流式处理的生命周期更长

服务端返回 Response 只表示建立了响应流，不表示所有业务消息已经完成。随后流的元素通常是 Result<Message, Status>；消费端须区分正常结束与错误，不能用 while let Some(Ok(...)) 静默吞掉错误。

向有界 mpsc 通道 send().await 可把消费者变慢反馈给生产者。发送失败表示接收方已消失，应结束相应工作而不是 unwrap；无界队列则可能把网络背压转成内存增长。后台任务还需持有并响应适当的取消信号。

消息大小限制不等于整个流的总大小限制。长流还需要时长、总字节、并发数或业务窗口控制；客户端只拉取一部分后丢弃，服务端也应释放关联资源。

在进程内验证生产者和消费者的终止协议

为前面的 tokio 依赖 features 加上 "sync"，再保存下面的 src/bin/stream_state.rs。它使用与流式服务常见生产端相同的有界通道，直接观察队列中的 Result；这一步不需要监听端口或调用远端服务。

use tokio::sync::mpsc;
use tonic::{Code, Status};
use grpc_demo::greeter::HelloReply;

#[tokio::main]
async fn main() {
    let (sender, mut receiver) = mpsc::channel::<Result<HelloReply, Status>>(1);
    let producer = tokio::spawn(async move {
        if sender.send(Ok(HelloReply { message: "part-1".into() })).await.is_err() {
            return;
        }
        let _ = sender.send(Err(Status::unavailable("source interrupted"))).await;
        // 终止错误之后不再生产业务消息，退出时释放 sender。
    });
    let mut received = Vec::new();
    let mut failure = None;
    while let Some(item) = receiver.recv().await {
        match item {
            Ok(reply) => received.push(reply.message),
            Err(status) => {
                failure = Some(status.code());
                break;
            }
        }
    }
    drop(receiver);
    producer.await.unwrap();
    assert_eq!(received, ["part-1"]);
    assert_eq!(failure, Some(Code::Unavailable));

    let (sender, receiver) = mpsc::channel::<Result<HelloReply, Status>>(1);
    drop(receiver);
    assert!(sender.send(Ok(HelloReply { message: "unused".into() })).await.is_err());
}

有界容量 1 只约束排队的消息项数，不限制一条 message 里的字符串字节数，也不限制生产者在 send 前已经构造的对象。生产者遇到接收方离开会停止发送；如果还持有数据库游标或启动了其他任务，应同时释放或取消那些资源。

实际 tonic 客户端可按 stream.message().await 的 Result<Option, Status> 分别处理消息、正常结束和错误，不能把 Err 当成 None。若业务要求一次完整快照，上例中的 part-1 只能算暂存数据，最终失败时不应当成完整结果提交；若业务允许增量消费，则需要消息序号、恢复游标和幂等处理来承接重连。

本地验证与传输验证覆盖不同问题

可以直接调用生成的 Greeter trait 方法验证空 name 与正常响应，再用 Prost 测试字段编码；这些检查快速定位业务和协议模型问题。传输层还需单独检查 metadata、TLS、消息限制、deadline 和流结束状态，因为直接调用 handler 不会经过 tonic 的 HTTP/2 编解码及中间件。

同样，客户端 set_timeout 产生的 metadata 需要服务端和中转链路正确理解。它不为创建 Channel 之前的 DNS/连接等待补上预算，也不等于后台任务会自动停止。围绕一次用户操作建立总期限，再分别验证建连、RPC 等待与应用资源清理。

Interceptor 适合什么

Interceptor 的接口处理 Request<()> 并同步返回请求或 Status，适合读取 metadata、注入轻量上下文和即时拒绝。它不能直接 await 远程鉴权，也不提供任意响应体变换能力。Interceptor

需要异步外部查询、请求与响应完整观测或更细的流控制时，使用适合插入层次的 Tower 服务，或在业务服务中显式完成。无论放在哪里，都应明确错误如何变成 Status，以及流结束时怎样清理资源，见 Tower 与中间件。