单服务中间件

这一部分的中间件都只包一个内层服务（Reconnect 包的是一个能造服务的 MakeService，造出来的也只有一个）。读它们只需要问两个问题：poll_ready 里做了什么（就绪主线），call 返回的 future 里做了什么（请求主线）。有的只改一段，有的两段都改，有的把一段整个换掉。

Timeout：poll_ready 原样转发，call 时才建一个 Sleep，和内层 future 赛跑。等待就绪的时间从来不算在超时里；排队时间算不算，取决于它在 Buffer 的外面还是里面。
ConcurrencyLimit：poll_ready 先从信号量取一个许可，存在服务自己身上；call 把许可移进响应 future，响应完成或 future 被丢弃时归还。
RateLimit：固定窗口计数，call 扣次数，用完后 poll_ready 等到窗口结束。不能 Clone，每个实例单独计数。
LoadShed：内层没就绪也告诉调用方「就绪」，到 call 时直接返回 Overloaded。它要放在 ConcurrencyLimit 外面才起作用。
Buffer：把服务交给一个后台 worker，调用方之间只共享一条有界通道。poll_ready 预留通道位置，响应 future 先等 worker 回传内层的 future，再在调用方自己的任务里 poll 它。
Filter：call 时先检查请求，不通过就返回错误，不调用内层。AsyncFilter 的检查是异步的，在响应 future 里做完检查再调用内层。
SpawnReady：内层没就绪时，把「等它就绪」放进一个新的 tokio 任务，poll_ready 之后只看这个任务的 JoinHandle。
Reconnect：poll_ready 里驱动 Idle → Connecting → Connected，连接失败不让服务报废，而是把错误留给下一次 call。

flowchart TB
  subgraph ready_only["只改 poll_ready"]
    rl[RateLimit]
    sr[SpawnReady]
  end
  subgraph future_only["只改响应 future"]
    to[Timeout]
    fi[Filter]
  end
  subgraph both["两段都改"]
    cl[ConcurrencyLimit]
    ls[LoadShed]
    bu[Buffer]
    rc[Reconnect]
  end

RateLimit 的 call 也有逻辑（扣次数），但响应 future 就是内层的；SpawnReady 的响应 future 只把错误转成 BoxError；Filter 的 poll_ready 原样转发，拒绝请求发生在 call 里。

两段各做什么

下表逐个对照源码。「没 poll_ready 就 call」一列是实际跑出来的结果：内层是一个永远就绪的 service_fn，每个中间件新建出来就直接 call 一次（程序见下文）。

中间件	`poll_ready`	响应 future	`Clone`	没 `poll_ready` 就 `call`
`Timeout`	原样转发内层，只把错误转成 `BoxError`	`ResponseFuture`：先 poll 内层，没完成再 poll `Sleep`，到期返回 `Elapsed`	derive，内层要 `Clone`	正常返回
`ConcurrencyLimit`	手里没许可就先取一个（取不到 `Pending`），再问内层	`ResponseFuture`：只转发 poll，字段里带着许可，drop 时归还	手写，共享信号量，新实例不带许可	panic
`RateLimit`	窗口次数用完时等 `Sleep` 到期，重开窗口；否则问内层	没有自己的类型，就是内层的 `S::Future`	没有	窗口还有次数时正常返回；用完后 panic
`LoadShed`	记下内层是否就绪，自己总是返回 `Ready(Ok)`（内层报错除外）	`ResponseFuture`：`Called` 转发内层，`Overloaded` 立即返回错误	手写，新实例 `is_ready = false`	返回 `Overloaded` 错误
`Buffer`	worker 还活着就用 `PollSender::poll_reserve` 在通道里预留位置	`ResponseFuture`：先等 `oneshot` 送回内层 future，再在调用方任务里 poll 它	手写，不要求内层 `Clone`；新实例不带预留位置	panic（来自 tokio-util）
`Filter`	原样转发内层	`ResponseFuture`：检查不通过是一个立即完成的 `Err`，通过就是内层 future	derive，内层和谓词都要 `Clone`	正常返回
`AsyncFilter`	原样转发内层	`AsyncResponseFuture`：`Check` 等异步检查，通过后在 future 里调用内层，进入 `WaitResponse`	derive；实现 `Service` 本身就要求内层 `Clone`	未测
`SpawnReady`	先直接问内层；`Pending` 就把内层移进 `tokio::spawn` 的任务里等就绪，之后只 poll `JoinHandle`	`ResponseFuture`：内层 future 加 `map_err`	没有	服务还在手里时正常返回
`Reconnect`	驱动 `Idle → Connecting → Connected`，再问内层；连接失败返回 `Ready(Ok)`，错误存起来	`ResponseFuture`：内层 future，或存下的连接错误	没有	panic

几点从表里能直接读出来的：

响应 future 带着资源的只有 ConcurrencyLimit。它的许可要活到响应结束，所以只能放进 future；其他中间件的状态都留在服务身上。
「新实例不带就绪状态」是通则。ConcurrencyLimit、LoadShed、Buffer 的 Clone 都是手写的，手写的原因都一样：poll_ready 攒下的东西（许可、is_ready、通道位置）不能跟着复制。所以对一个已经 ready 的服务 clone() 再去 call 副本，和没 poll_ready 就 call 是同一种错。规则本身见 Service trait。
panic 只是允许，不是保证。Service::call 的文档说没就绪就 call 时实现「允许 panic」。上面的实验里 panic 的是三个，RateLimit 要看当时的状态，其余的照常返回或返回错误。不能拿「没 panic」当作「用对了」。

四条 panic 信息的原文：

中间件	信息	出处
`ConcurrencyLimit`	`max requests in-flight; poll_ready must be called first`	`call` 里的 `expect`
`RateLimit`	`service not ready; poll_ready must be called first`	`call` 在 `Limited` 状态下的 `panic!`
`Reconnect`	`service not ready; poll_ready must be called first`	`call` 不在 `Connected` 状态时的 `panic!`
`Buffer`	`send_item` called without first calling `poll_reserve`	tokio-util 0.7.20 的 `PollSender::send_item`，`Buffer::call` 调用它

SpawnReady 的 call 里还有一个 unreachable!("poll_ready must be called")，只在内层已经被移进后台任务、还没取回来时触发；新建的 SpawnReady 服务还在手里，所以上面的实验里没有 panic。

验证程序（M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同）：

tower = { version = "=0.5.3", features = ["full"] }
tokio = { version = "=1.54.1", features = ["full"] }
futures = "0.3"

use std::{
  convert::Infallible,
  fmt::Debug,
  future::Future,
  panic::{catch_unwind, AssertUnwindSafe},
  time::Duration,
};
use tower::{
  buffer::Buffer, filter::Filter, limit::{ConcurrencyLimit, RateLimit}, limit::rate::Rate,
  load_shed::LoadShed, reconnect::Reconnect, service_fn, spawn_ready::SpawnReady,
  timeout::Timeout, BoxError, Service,
};

fn try_call<S, E>(name: &str, mut svc: S)
where
  S: Service<u32, Error = E>,
  S::Future: Future<Output = Result<u32, E>>,
  E: Debug + ToString,
{
  let r = catch_unwind(AssertUnwindSafe(|| {
    let fut = svc.call(7);
    futures::executor::block_on(fut).map_err(|e| e.to_string())
  }));
  match r {
    Ok(v) => println!("{name:<16} {v:?}"),
    Err(p) => {
      let msg = p
        .downcast_ref::<String>()
        .cloned()
        .or_else(|| p.downcast_ref::<&str>().map(|s| s.to_string()));
      println!("{name:<16} panic: {}", msg.unwrap_or_default());
    }
  }
}

#[tokio::main]
async fn main() {
  std::panic::set_hook(Box::new(|_| {})); // keep the default hook quiet
  let inner = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  tokio::task::spawn_blocking(move || {
    try_call("Timeout", Timeout::new(inner, Duration::from_secs(1)));
    try_call("ConcurrencyLimit", ConcurrencyLimit::new(inner, 1));
    try_call("RateLimit", RateLimit::new(inner, Rate::new(1, Duration::from_secs(1))));
    try_call("LoadShed", LoadShed::new(inner));
    try_call("Filter", Filter::new(inner, |r: u32| Ok::<_, BoxError>(r)));
    try_call("SpawnReady", SpawnReady::new(inner));
    let mk = service_fn(move |_: ()| std::future::ready(Ok::<_, Infallible>(inner)));
    try_call("Reconnect", Reconnect::new(mk, ()));
  })
  .await
  .unwrap();
  // Buffer::new spawns its worker, so build it inside the runtime
  let buf = Buffer::new(inner, 1);
  tokio::task::spawn_blocking(move || try_call("Buffer", buf)).await.unwrap();
}

Timeout          Ok(7)
ConcurrencyLimit panic: max requests in-flight; poll_ready must be called first
RateLimit        Ok(7)
LoadShed         Err("service overloaded")
Filter           Ok(7)
SpawnReady       Ok(7)
Reconnect        panic: service not ready; poll_ready must be called first
Buffer           panic: `send_item` called without first calling `poll_reserve`

RateLimit 那一行是窗口里的第一次调用，所以没有 panic；同一个实例次数用完后再 call 就会 panic，见 RateLimit。

单服务中间件

两段各做什么

💬 关联讨论

登录