Tower 与中间件

Tower 把一次异步请求抽象成 Service，把“包装一个服务”的配置抽象成 Layer。超时、并发限制、缓冲与重试就可以围绕同一套接口组合。它并不规定请求一定是 HTTP，也不自动解决业务幂等性或所有形式的背压。

本文以 Tower 0.5 API 为例。阅读中间件时先确定它控制的是接纳请求、等待队列、响应 Future，还是响应体的后续流动；这些时间范围通常不同。

Service 有接纳与调用两个阶段

核心接口的形状是：

use std::future::Future;
use std::task::{Context, Poll};

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>)
        -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, request: Request) -> Self::Future;
}

poll_ready 的接收者也是 &mut self。它可能需要预留许可或更新内部状态，并非只读地“看一眼当前负载”。Pending 时应安排状态变化后的唤醒，调用者等待后再轮询；Ready(Ok(())) 才表示本次可以调用，Ready(Err(e)) 则是服务错误。

调用者必须遵守 readiness 契约，不能把一次成功视为无限次调用许可。有的服务在未 ready 时调用会 panic。也不能在实例 A 上等待 ready 后 clone 出 B，再假定 B 继承 A 已预留的能力。Service 文档

call 的可变借用用于创建关联 Future，不等于把服务借用直到该 Future 完成。服务可以产生多个独立的在途 Future；是否接纳更多请求由 readiness 与实现决定。因此 Buffer 不是“解除 Rust 天生串行限制”的必需品。

用一个许可观察 ready 的状态变化

下面的完整程序使用本页后面的 Cargo 依赖。内部 Future 永远 Pending，因此状态变化完全由持有和丢弃许可决定，不需要等待计时器。

use std::{convert::Infallible, future::pending, task::{Context, Poll, Waker}};
use tower::{Service, ServiceBuilder, service_fn};
fn main() {
    let inner = service_fn(|(): ()| pending::<Result<(), Infallible>>());
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let mut cx = Context::from_waker(Waker::noop());
    assert!(matches!(service.poll_ready(&mut cx), Poll::Ready(Ok(()))));
    let first = service.call(());
    assert!(service.poll_ready(&mut cx).is_pending());
    drop(first);
    assert!(matches!(service.poll_ready(&mut cx), Poll::Ready(Ok(()))));
}

即使 first 从未被 poll，它也已持有从 readiness 转入的许可。丢弃后许可释放，下一次 poll_ready 恢复就绪。这说明接纳资源的起点可能早于内部业务首次执行；把大量响应 Future 创建后存起来，也可能占住服务容量。

Waker::noop 在这里仅用于手动检查立即可观察的状态，每次状态改变后程序主动再次 poll。真实等待代码应交给执行器和 ready().await，不能用无效唤醒器写忙循环。Service 就绪契约

Layer 是构造包装器的接口

Layer::layer 接收内部服务并返回新服务。ServiceBuilder 按声明顺序将先添加的层放在外面：A、B、S 构成 A(B(S))。请求先经过 A，再经过 B；完成结果沿相反方向回到调用者。ServiceBuilder

这个顺序会改变语义。假设 A 是重试、B 是超时：Retry(Timeout(S)) 可以让每次尝试各有超时；Timeout(Retry(S)) 则约束包含多次尝试的响应 Future。两者都还需要解释等待 readiness 的时间是否被纳入预算。

可运行例子：先 ready，再 call

在一个空 Cargo 二进制项目中使用：

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

[dependencies]
tokio = { version = "1", features = ["macros", "rt", "time"] }
tower = { version = "0.5", features = ["util", "limit", "timeout"] }

src/main.rs：

use std::convert::Infallible;
use std::time::Duration;
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|delay_ms: u64| async move {
        tokio::time::sleep(Duration::from_millis(delay_ms)).await;
        Ok::<_, Infallible>(delay_ms)
    });
    let mut service = ServiceBuilder::new()
        .timeout(Duration::from_millis(50))
        .concurrency_limit(1)
        .service(inner);

    let value = service.ready().await?.call(0).await?;
    println!("completed: {value}");

    let result = service.ready().await?.call(500).await;
    match result {
        Ok(value) => println!("completed: {value}"),
        Err(error) => println!("failed: {error}"),
    }
    Ok(())
}

运行 cargo run。通常第一项完成，第二项超时。机器严重阻塞时，调度会影响墙钟观察，不能用这段程序校准精确延迟。

这里 Timeout 在 ConcurrencyLimit 外面，但 ready().await 发生在 call 之前，50 ms 不包含那段等待。若需要端到端预算，可在调用方用一个绝对 deadline 包住整个 ready 加 call 的异步过程，并把剩余预算传给下游；重试也应消耗同一总预算。

将绝对期限放在 ready 和 call 外面

use std::{convert::Infallible, future::pending, time::Duration};
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|(): ()| pending::<Result<(), Infallible>>());
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let first = service.ready().await?.call(());
    let deadline = tokio::time::Instant::now() + Duration::from_millis(1);
    let second = tokio::time::timeout_at(deadline, async {
        service.ready().await?.call(()).await
    }).await;
    assert!(second.is_err()); // 等待接纳就已用完预算，尚未发出第二次 call
    drop(first);
    let ready = service.ready().await?;
    drop(ready.call(()));
    Ok(())
}

本例故意让第一个请求不完成，验证等待许可也会受外层期限约束；1 ms 只是很短的测试预算，不承诺线程按该精度得到调度。生产链路需要选定同一个请求期限，下游每次等待或重试都计算剩余时间，不能每经过一层就重新得到完整预算。

如果内层拿到响应对象后还会持续读 body，上面的期限只覆盖取得 Response 的阶段。需要全流期限时，应对流的下一项等待继续使用原期限；需要“空闲超时”时，则按最后一次有效进度更新另一个计时器。这两种超时解决不同问题。

超时预算从哪个位置开始？

把时间轴拆成接纳等待与响应 Future 两段。Timeout 层从 call 得到的响应 Future 开始约束，所以等待许可 40 ms、响应 30 ms 仍可能正常返回，总等待却已达 70 ms。若业务要求从入口起最多等待 50 ms，应让同一个绝对期限包住 ready 与 call；相同截止时刻的竞争需按实际运行时判断。

背压需要整个调用链配合

ConcurrencyLimit 使用许可控制在途响应 Future。达到上限时 readiness 等待，调用方如果转而无限创建等待任务，系统仍可能消耗大量内存。背压只有沿上游继续传播，或配合有界队列和拒绝策略，才形成整体容量控制。

Buffer 通过有界通道和后台 worker task 转发请求，提供可克隆的入口，不必创建专属 OS 线程。它隔开了两侧 readiness，也引入队列等待。上游获得的 ready 可能预留队列位置，应及时 call，避免持有不用的许可。Buffer

RateLimit 约束一段时间内可处理的请求数量，不等于并发限制；不能不核对实现就称为平滑补充 token 的通用令牌桶。LoadShed 则将暂时未就绪转成错误，适合明确拒绝而不是继续排队的边界。RateLimit

超时取消了什么

Timeout 结束返回的响应 Future 后，调用者通常将它丢弃，其内部 Future 随之释放。对于靠轮询推进的普通异步逻辑，这可以停止后续推进；但已经发送的网络请求、已提交的数据库事务、独立 spawn 的任务并不会因此自动撤销。

因此不能写成“超时完全不会停止内部执行”，也不能写成“超时会撤销全部副作用”。是否取消安全，应结合 Future 的 Drop、资源所有权与底层操作语义判断。超时错误也不证明远端业务没有成功。

重试与流式响应的两条边界

Retry 依据策略判断是否再次尝试，并需要能重建或克隆请求。可重试性来自业务效果和失败阶段：查询往往容易重试，写入可借助幂等键和持久去重；不能只依据 HTTP 方法名保证安全。还应限制次数、退避和总时间，避免故障时把负载进一步放大。

Service 的 Response 可能包含一个尚未消费的流。响应 Future 完成后，并不代表流已结束：普通 Timeout 或 ConcurrencyLimit 所覆盖的生命周期可能已经结束，而客户端仍在持续接收消息。限制完整流时，需要把许可、超时及清理策略延伸到流生命周期。

用可延后消费的 Response 看清许可释放位置

use std::convert::Infallible;
use tower::{Service, ServiceBuilder, ServiceExt, service_fn};
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), tower::BoxError> {
    let inner = service_fn(|(): ()| async {
        Ok::<_, Infallible>(vec![1, 2, 3].into_iter())
    });
    let mut service = ServiceBuilder::new().concurrency_limit(1).service(inner);
    let mut response = service.ready().await?.call(()).await?;
    // 第一个 Response 仍未消费，但已能接纳下一次调用。
    let second = service.ready().await?.call(()).await?;
    assert_eq!(response.next(), Some(1));
    assert_eq!(second.collect::<Vec<_>>(), [1, 2, 3]);
    Ok(())
}

这里用普通迭代器演示“返回对象以后还能继续消费”的拥有关系，并非模拟网络 Stream。Tower 的通用限制器无法从任意 Response 类型猜到何时才算真正结束。若受限资源属于流，应让流包装器持有许可，正常结束、错误和 Drop 都有明确释放路径；若只想限制产生响应头的工作，则当前范围可能正好合适。

按层次记录排队与处理时间

排队、超时和重试应共用逻辑请求标识，同时保留每次尝试的记录。否则一次请求重试三次可能同时拉高调用量和“成功率”，掩盖用户实际等待及下游额外负载。

响应返回后，流还占着许可吗？

Response 是一个值，它可以拥有后续才消费的流。将进度切到“Response 已返回”，观察普通响应 Future 的许可已释放，而 Body 仍存活。若目标是限制同时存在的完整流，就让流的拥有对象继续保管许可，并在结束、错误或 Drop 时释放。

接入 tonic 时先检查类型与错误映射

tonic 在传输及生成服务层使用 Tower，但不是“每个 gRPC 方法都能直接套任意 Layer”。中间件的 Request、Response、Error、Clone、Send 等约束必须匹配插入位置；HTTP 层的错误还需要正确映射为 gRPC Status。

HTTP 请求追踪通常使用 tower-http 的 TraceLayer，不能误认为所有 Trace 都来自 tower 本体。Interceptor 适合简短同步 metadata 处理，异步认证和响应体观测通常需要更完整的中间件，参见 tonic。