6.2 Tonic gRPC 框架：类型安全的 RPC 调用，微服务通信的最佳实践

引言：从数据传输到服务调用

在上一章，我们掌握了 Protobuf 和 prost，学会了如何高效、类型安全地序列化和反序列化我们的数据结构。这解决了“如何打包数据”的问题。但是，在微服务架构中，我们还面临另一个更重要的问题：“如何调用另一个服务的功能？”

传统的 RESTful API 通过 HTTP/1.1 和 JSON 来实现服务调用，但这存在一些问题：

• 性能：HTTP/1.1 的文本协议和队头阻塞（Head-of-Line Blocking）问题在高吞吐量场景下性能不佳。
• 类型安全：你需要手动维护客户端和服务器之间的 API 契约，很容易因为不一致而出错。
• 流式传输：实现双向流式传输非常困难。

gRPC 就是为了解决这些问题而生的。它是一个由 Google 开发的高性能、开源的通用 RPC（远程过程调用）框架。它使用 Protobuf 作为其接口定义语言（IDL）和底层消息交换格式，并构建在 HTTP/2 之上。

gRPC 的核心优势：

• 类型安全：通过 .proto 文件定义服务接口，客户端和服务器的代码都是自动生成的，从根本上杜绝了接口不匹配的问题。
• 高性能：基于 HTTP/2，支持多路复用、头部压缩，并使用 Protobuf 进行二进制序列化，性能远超 REST+JSON。
• 强大的流式处理：原生支持四种通信模式：一元调用、服务器流、客户端流和双向流。

在 Rust 生态中，tonic 是构建 gRPC 服务的首选框架。它由 tokio 团队出品，与 prost、tokio 和 tower 无缝集成，为构建现代、高性能的 Rust 微服务提供了完整的解决方案。

本章，我们将学习如何使用 tonic 和 prost 来定义和实现一个完整的 gRPC 服务。

gRPC 的四种通信模式

gRPC 定义了四种服务方法类型，覆盖了所有可能的通信场景。

1. 一元 RPC (Unary RPC)：
   最简单的模式，类似于普通的函数调用。客户端发送一个请求，服务器返回一个响应。

   rpc GetUser(GetUserRequest) returns (User);
   

2. 服务器流 RPC (Server-streaming RPC)：
   客户端发送一个请求，服务器返回一个数据流。客户端可以持续从这个流中读取消息，直到流结束。适用于服务器向客户端推送大量数据的场景，如下载、订阅。

   rpc SubscribeUpdates(SubscriptionRequest) returns (stream Update);
   

3. 客户端流 RPC (Client-streaming RPC)：
   客户端向服务器发送一个数据流，服务器在接收完所有数据后，返回一个响应。适用于客户端向服务器上传大量数据的场景。

   rpc UploadLog(stream LogEntry) returns (UploadSummary);
   

4. 双向流 RPC (Bidirectional-streaming RPC)：
   客户端和服务器都可以独立地、异步地向对方发送消息流。这是一种完全双工的通信模式，非常灵活，适用于聊天、实时协作等场景。

   rpc Chat(stream ChatMessage) returns (stream ChatMessage);
   

使用 .proto 定义 gRPC 服务

我们扩展上一章的 user.proto 文件，为其添加一个 UserService。

proto/user.proto:

syntax = "proto3";

package user;

// ... (User, UserState 等 message 定义保持不变) ...

message GetUserRequest {
  uint64 id = 1;
}

// 定义我们的 gRPC 服务
service UserService {
  // 一元 RPC
  rpc GetUser(GetUserRequest) returns (User);

  // 服务器流 RPC
  rpc ListUsers(ListUsersRequest) returns (stream User);
}

message ListUsersRequest {
  // 可以添加分页等参数
}

我们定义了一个名为 UserService 的服务，它包含两个 RPC 方法：GetUser（一元）和 ListUsers（服务器流）。

使用 tonic-build 生成服务代码

与 prost-build 类似，tonic 提供了一个 tonic-build crate，它可以在 build.rs 中与 prost-build 配合使用，来同时生成 Protobuf 的数据结构代码和 gRPC 的服务/客户端代码。

环境准备

修改 Cargo.toml，添加 tonic 和 tonic-build。

[package]
name = "grpc-example"
version = "0.1.0"
edition = "2021"

[dependencies]
prost = "0.11"
tonic = "0.8"
tokio = { version = "1", features = ["full"] }
anyhow = "1.0"
futures = "0.3"

[build-dependencies]
tonic-build = "0.8"

注意：prost 不再是 build-dependencies，而是 tonic-build 的一个传递性依赖。tonic 本身依赖 prost，所以我们把它加到 dependencies 中。

修改 build.rs

现在，我们让 tonic-build 来驱动代码生成。

build.rs:

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // tonic_build::compile_protos 会在内部调用 prost_build
    tonic_build::configure()
        .compile(
            &["proto/user.proto"], // 要编译的 .proto 文件
            &["proto"],           // .proto 文件的包含路径
        )?;
    Ok(())
}

tonic_build 比 prost-build 更智能，它的配置也更丰富。例如，我们可以像之前一样，为 prost 生成的类型添加 serde 支持：

// build.rs (带 serde 支持)
fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::configure()
        // 为所有 message 添加 serde 支持
        .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]")
        .compile(&["proto/user.proto"], &["proto"])?;
    Ok(())
}

生成了什么代码？

运行 cargo build 后，tonic-build 会生成一个 user.rs 文件，除了 prost 生成的 struct User 等，tonic 还额外生成了：

• 服务器端代码 (user_service_server)：
  • 一个 UserServiceServer 结构体，用于构建 gRPC 服务器。
  • 一个 UserService trait，这是我们需要为我们的服务逻辑实现的 trait。
• 客户端代码 (user_service_client)：
  • 一个 UserServiceClient 结构体，它就是我们可以用来调用 gRPC 服务的客户端 stub。

在代码中包含生成的部分

src/lib.rs (或 main.rs):

// 这会包含 prost 和 tonic 生成的所有代码
// `user` 是 .proto 文件中的 package 名
pub mod user {
    // 告诉 tonic 使用 prost 作为编解码器
    tonic::include_proto!("user"); 
}

tonic::include_proto! 是 tonic 提供的一个宏，它会处理好包含 OUT_DIR 中文件的所有细节。

实现 gRPC 服务器

实现一个 gRPC 服务的核心就是实现 tonic-build 为我们生成的那个 UserService trait。

src/server.rs:

use tonic::{transport::Server, Request, Response, Status};
use user::{
    user_service_server::{UserService, UserServiceServer},
    User, GetUserRequest, ListUsersRequest,
};

// 引入生成的代码
pub mod user {
    tonic::include_proto!("user");
}

// 1. 定义我们的服务逻辑结构体
#[derive(Debug, Default)]
pub struct MyUserService {}

// 2. 为我们的结构体实现 generated UserService trait
#[tonic::async_trait]
impl UserService for MyUserService {
    // 实现 GetUser (一元 RPC)
    async fn get_user(
        &self,
        request: Request<GetUserRequest>, // 请求被包装在 tonic::Request 中
    ) -> Result<Response<User>, Status> { // 响应需要包装在 Result<Response<...>, Status> 中
        println!("收到请求: {:?}", request);

        let user_id = request.into_inner().id;

        // 模拟从数据库查找用户
        if user_id == 1 {
            let user = User {
                id: 1,
                name: "Alice".to_string(),
                email: "alice@example.com".to_string(),
                // ...
            };
            // 使用 Response::new 包装我们的 message
            Ok(Response::new(user))
        } else {
            // 使用 Status::new 来返回一个 gRPC 错误
            Err(Status::not_found(format!("User with id {} not found", user_id)))
        }
    }

    // 为服务器流 RPC 定义响应流的类型
    type ListUsersStream = tokio::sync::mpsc::Receiver<Result<User, Status>>;

    // 实现 ListUsers (服务器流 RPC)
    async fn list_users(
        &self,
        request: Request<ListUsersRequest>,
    ) -> Result<Response<Self::ListUsersStream>, Status> {
        println!("收到 ListUsers 请求: {:?}", request);
        
        // 创建一个 channel 来将数据流式地发送给客户端
        let (tx, rx) = tokio::sync::mpsc::channel(4);

        // spawn 一个新任务来模拟生成数据流
        tokio::spawn(async move {
            for i in 0..5 {
                let user = User { id: i, name: format!("User {}", i), ..Default::default() };
                // 发送数据，如果客户端断开连接，就停止发送
                if tx.send(Ok(user)).await.is_err() {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
            println!("服务器流发送完毕");
        });

        // 返回 channel 的接收端 rx 作为响应流
        Ok(Response::new(rx))
    }
}

// 启动服务器的 main 函数
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:50051".parse()?;
    let user_service = MyUserService::default();

    println!("gRPC 服务器正在监听 {}", addr);

    Server::builder()
        // 添加我们实现的服务
        .add_service(UserServiceServer::new(user_service))
        .serve(addr)
        .await?;

    Ok(())
}

代码分析:

• #[tonic::async_trait]: tonic 自带了 async_trait 的功能，让我们可以直接在 trait 中使用 async fn。
• Request<T> 和 Response<T>: tonic 将所有传入的请求和传出的响应都包装在 Request 和 Response 结构体中。它们提供了对元数据（如 HTTP/2 头部）的访问。使用 request.into_inner() 来获取内部的 Protobuf message。
• Status: tonic 使用 Status 类型来表示 gRPC 错误。它对应 gRPC 的标准错误码（如 NotFound, InvalidArgument 等）。
• 服务器流: 实现服务器流 RPC 的关键是返回一个实现了 Stream<Item = Result<T, Status>> 的类型。tokio::sync::mpsc::Receiver 就是一个很好的选择。我们在一个新 spawn 的任务中向 channel 的 tx 端发送数据，并将 rx 端返回给客户端。

实现 gRPC 客户端

tonic-build 也为我们生成了开箱即用的客户端代码。编写一个客户端来调用我们的服务非常简单。

src/client.rs:

use tonic::Request;
use user::{
    user_service_client::UserServiceClient,
    GetUserRequest, ListUsersRequest,
};

// 引入生成的代码
pub mod user {
    tonic::include_proto!("user");
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. 创建一个到服务器的连接 channel
    let mut client = UserServiceClient::connect("http://[::1]:50051").await?;
    
    // --- 2. 调用一元 RPC ---
    println!("--- 调用 GetUser ---");
    let request = Request::new(GetUserRequest { id: 1 });
    match client.get_user(request).await {
        Ok(response) => println!("响应: {:?}", response.into_inner()),
        Err(e) => println!("错误: {}", e.message()),
    }
    
    // 调用一个会失败的
    let request_fail = Request::new(GetUserRequest { id: 99 });
    match client.get_user(request_fail).await {
        Ok(_) => {},
        Err(e) => println!("预期的错误: status={}, message={}", e.code(), e.message()),
    }

    // --- 3. 调用服务器流 RPC ---
    println!("\n--- 调用 ListUsers ---");
    let request = Request::new(ListUsersRequest {});
    // 调用会立即返回一个响应，其中包含一个数据流
    let mut stream = client.list_users(request).await?.into_inner();
    
    // 异步地遍历数据流
    while let Some(user) = stream.message().await? {
        println!("从流中收到用户: {:?}", user);
    }
    println!("服务器流已结束");
    
    Ok(())
}

代码分析:

• UserServiceClient::connect(...): 创建一个客户端。在底层，它会建立一个 HTTP/2 连接，并可以被多个克隆的客户端实例复用。
• client.get_user(...): tonic 为 .proto 服务中的每个 RPC 方法都在客户端上生成了一个对应的 async 方法。调用它就像调用一个普通的异步函数一样简单。
• 对于流式 RPC，client.list_users(...) 返回的响应中包含一个 Streaming<User> 对象。我们可以调用它的 message().await 在一个循环中来接收每一条消息。

现在，先启动服务器 (cargo run --bin server)，然后启动客户端 (cargo run --bin client)，你就能看到客户端成功地调用了服务器的 RPC 方法。

总结

tonic 与 prost 的结合，为 Rust 带来了与 Go、Java 等语言同等甚至更强大的 gRPC 开发体验。

1. IDL 驱动开发：从 .proto 文件开始，我们通过代码生成获得了完全类型安全的服务器 trait 和客户端存根 (stub)。这使得服务间的契约非常明确，且由编译器保证。
2. 高性能网络：tonic 构建在 hyper 和 tokio 之上，充分利用了 HTTP/2 和异步 I/O 的性能优势。
3. 强大的流处理能力：tonic 将 gRPC 的四种流模式与 Rust 的 Future 和 Stream trait 完美结合，使得处理复杂的数据流变得既自然又高效。
4. 与 tower 集成：tonic 的 Server 和 Client 都构建在 tower::Service 之上。这意味着我们可以像为 axum 应用添加中间件一样，为我们的 gRPC 服务添加 tower 的 Layer，例如日志、认证、限流等。我们将在下一章深入探讨这一点。

通过 tonic，我们可以用一种类型安全、高性能且符合人体工程学的方式来构建复杂的微服务系统。它将 RPC 调用抽象成了简单的异步函数调用，让开发者可以更专注于业务逻辑，而不是底层的网络和序列化细节。

思考题

1. gRPC 和 REST API 的主要区别是什么？在什么场景下你会优先选择 gRPC？
2. 在我们的 list_users 服务器流实现中，我们使用了 tokio::spawn 来发送数据。为什么需要一个新的任务？如果不 spawn 一个新任务，直接在 list_users 函数体内部循环发送数据，会发生什么？
3. tonic::Status 和 anyhow::Error/thiserror 在错误处理中扮演的角色有什么不同？在一个同时有 gRPC 服务和内部业务逻辑的应用中，你会如何组合使用它们？
4. gRPC 的元数据（metadata）相当于 HTTP/1.1 中的什么？在 tonic 中，你应该如何从 Request 中读取元数据，以及如何向 Response 中添加元数据？
5. 客户端流和双向流 RPC 在 tonic 中是如何实现的？请查阅 tonic 的文档或示例，简述一下实现这两种模式的大致思路。

实践练习

1. 实现客户端流 RPC：
   • 在 user.proto 中添加一个客户端流 RPC：rpc CreateUsers(stream User) returns (CreateUsersResponse);。CreateUsersResponse 可以包含一个成功创建的用户数量。
   • 在服务器端实现 create_users 方法。它需要接收一个 Request<Streaming<User>>，你可以遍历这个流来处理每个传入的 User。
   • 在客户端编写代码，创建一个 User 的流（例如，从一个 Vec），并调用 client.create_users 方法。
2. 添加 tower 中间件：
   • 为你的 UserService 服务器添加一个 tower_http::trace::TraceLayer。
   • 观察当你调用 gRPC 方法时，服务器端打印出的追踪日志是怎样的。
3. 错误处理实践：
   • 在 MyUserService::get_user 中，对于 id > 1000 的请求，返回一个 Status::invalid_argument("ID cannot be greater than 1000") 错误。
   • 在客户端捕获这个错误，并打印出它的 code() 和 message()。