--- title: Tower 与中间件 url: https://doc.liz6.com/rust/12-async-ecosystem/01-tower-and-middleware locale: zh area: rust tags: - rust - 异步生态 date: 2026-06-30 modified: 2026-07-11 description: Tower 用 Service trait(处理请求并返回 Future)和 Layer(包装 Service 产生新 Service)两层抽象构建异步中间件栈——中间件可以叠加(超时→重试→限流→业务逻辑),backpressure 通过等待 readiness 传递到上游。这是 Rust 异步生态中最优雅的洋葱架… --- # Tower 与中间件 > Tower 用 Service trait(处理请求并返回 Future)和 Layer(包装 Service 产生新 Service)两层抽象构建异步中间件栈——中间件可以叠加(超时→重试→限流→业务逻辑),backpressure 通过等待 readiness 传递到上游。这是 Rust 异步生态中最优雅的洋葱架构实现。 ## Tower 解决的问题构建网络服务时有一些横切关注点（cross-cutting concerns）：超时、限流、并发控制、重试、日志/metrics。如果把这些逻辑硬编码在每个服务中，会导致代码重复和不可组合——你的 gRPC handler 既要处理业务逻辑又要处理超时，你的 HTTP handler 需要同样的限流逻辑。 Tower 的答案：把服务抽象为 `Service` trait，用 `Layer` 做装饰器模式组合这些关注点。每个中间件只负责一件事，Layer stack 将它们组合成完整的服务。 ## Service trait: 异步的 request → response ```rust pub trait Service { type Response; type Error; type Future: Future>; fn poll_ready(&self, cx: &mut Context<'_>) -> Poll>; fn call(&mut self, req: Request) -> Self::Future; } ``` 这个 trait 的两个方法有明确分工： **`poll_ready`**: 检查服务是否可以接受新请求。rate limiter 检查剩余额度、concurrency limiter 检查空闲 slot。如果服务还不能接收，返回 `Poll::Pending`——调用者必须等下次被唤醒再试。这是 Tower 内建的 **backpressure 机制**——下游服务的压力通过 `poll_ready` 向上游传导，不需要任何额外的消息通道。 **`call`**: 执行请求并返回 Future。`call` 的 `&mut self` 签名意味着同一 Service 实例的调用天然是序列化的——如果需要在同一实例上并行，用 `Buffer` middleware 将其变为 Clone。这两个阶段的分离是 Tower 与大多数中间件框架（如 Express/actix middleware）的核心区别——它们通常只有一个 `fn handle(&self, req) -> Future`，没有 backpressure 的语义。 ## Layer: 装饰器模式的函数化表示 ```rust pub trait Layer { type Service; fn layer(&self, inner: S) -> Self::Service; } ``` Layer 不是 Service——它是 Service 的**工厂**。给定 inner service，Layer 返回包裹了 inner 的 wrapper service。核心优势：**Layer 可以被组合**： ```rust let svc = ServiceBuilder::new() .layer(TimeoutLayer::new(Duration::from_secs(5))) .layer(ConcurrencyLimitLayer::new(100)) .layer(RateLimitLayer::new(10, Duration::from_secs(1))) .service(my_service); ``` `ServiceBuilder::new().layer(A).layer(B).service(S)` 实际上构建了： ``` A::layer(B::layer(S)) ``` 即 `A << B << S`——每个 layer 包裹 inner。真正请求来临时，顺序是外到内：request → A → B → S → B → A → response。 ## 理解 backpressure 没有 backpressure，并发请求直接打到服务上，服务可能被超载。Tower 的 `poll_ready` 使得中间件**在接收请求之前**就可以拒绝：这意味着 `ConcurrencyLimitLayer` 达到并发上限时，新的 `poll_ready` 返回 Pending——不是 panic，不是 reject，而是**优雅地推迟**。caller 的 Future 在 slot 空闲后被唤醒，继续处理。整个过程在 Future 的 poll 模型中自然表达，不需要额外的信号量或 channel。 ## 常用中间件 - **`Timeout`**: `call` 返回的 Future 在指定时间后还没 Ready → 返回 timeout error。注意：这不会停止内部执行（tokio 中无法强制取消底层 task，除非用 `tokio::select!` 或 `JoinHandle::abort`） - **`ConcurrencyLimit`**: 内部计数器限制同时处理的请求数——poll_ready 返回 Pending 当计数器达到上限 - **`RateLimit`**: token bucket 算法限制请求速率——poll_ready 返回 Pending 当 token 不足 - **`Buffer`**: 解决 `call(&mut self)` 不能共享的问题——内部用 channel 将请求转发给 worker thread，对外暴露 Clone - **`Retry`**: 自动重试失败的请求——可配置重试条件和指数退避。但注意幂等性：retry 只对幂等操作（GET, PUT）安全 - **`Trace`**: 自动记录每个请求的耗时——不需要在业务代码中手动插入记时代码 ## tonic 是 tower 的原生消费者 ```rust tonic::transport::Server::builder() .layer(TimeoutLayer::new(Duration::from_secs(5))) .add_service(GreeterServer::new(MyGreeter)) .serve(addr).await?; ``` 每个 gRPC 方法自动成为一个 `Service`——tonic 的 codegen 为你的 handler 生成了 `Service` impl。这意味着所有 tower 的中间件可以不加修改地应用到任何 gRPC 服务上。 ## 参考 - **tower**: docs.rs/tower (README 中有设计哲学的解释) - **tonic**: github.com/hyperium/tonic *Keywords: tower, Service, Layer, middleware, tonic, poll_ready, backpressure*