连接没有被复用

主线位置：客户端主线的一头一尾。请求开始时 Client 先到池里按 (scheme, authority) 找一条空闲连接；响应 body 读完、连接任务再次说「要」以后，连接才回到池里。复用失败，要么是回池这一步没发生，要么是回了池却在下次取出前失效，要么是取的时候根本没去同一个池子、同一个键下找。

flowchart TB
  req["client.request"] --> key["算池的键"]
  key --> pop["空闲列表里找"]
  pop -->|找到且可用| reuse["复用"]
  pop -->|没有| fresh["新建连接"]
  reuse --> busy["请求在途"]
  fresh --> busy
  busy -->|"body 读完，再次 want"| back["Pooled drop：put"]
  back -->|"is_open 且未满"| idle["空闲列表"]
  back -->|"已关 / 已满 / 池已没"| gone["连接关闭"]
  idle -->|"过期 / 对端关了"| gone
  idle --> pop

这个现象没有错误信息，请求都成功了，只是慢一点（每次多一次 TCP 握手，用 TLS 的话再多一次 TLS 握手），服务端的连接数、客户端的临时端口用得多一些。池子的机制在连接池一章讲过，这一页按「看到的现象」把能让复用失败的情况排一遍。

怎么看出来

服务端能数 accept 的次数，但服务端不一定是你的。只看客户端也可以：HttpConnector 建连接时把一个 HttpInfo（本地地址、对端地址）放进 Connected 的 extra 里，legacy Client 拿到响应后把它塞进响应的 extensions（client.rs 第 343–346 行）。同一条连接，本地端口不变；端口变了，就是新连接。

let port = res.extensions().get::<HttpInfo>().unwrap().local_addr().port();

复现：九种情况

每种情况用一个新的计数服务端，客户端发两个请求，中间隔 200 ms（「同时」那一行除外），最后打印服务端 accept 了几次、两次请求的本地端口是否相同：

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::client::legacy::connect::{HttpConnector, HttpInfo};
use hyper_util::client::legacy::Client;
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::net::TcpListener;

type C = Client<HttpConnector, Empty<Bytes>>;

// /close answers with `connection: close`; /big has a 1 MiB body.
async fn handler(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  let mut res = match req.uri().path() {
    "/big" => Response::new(Full::new(Bytes::from(vec![b'x'; 1 << 20]))),
    _ => Response::new(Full::new(Bytes::from("ok"))),
  };
  if req.uri().path() == "/close" {
    res.headers_mut().insert("connection", "close".parse().unwrap());
  }
  Ok(res)
}

// `idle`: let hyper's header_read_timeout close idle keep-alive connections
async fn server(idle: Option<Duration>) -> (SocketAddr, Arc<AtomicUsize>) {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let accepts = Arc::new(AtomicUsize::new(0));
  let counter = accepts.clone();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      counter.fetch_add(1, SeqCst);
      let mut b = http1::Builder::new();
      if let Some(d) = idle {
        b.timer(TokioTimer::new()).header_read_timeout(d);
      }
      tokio::spawn(b.serve_connection(TokioIo::new(stream), service_fn(handler)));
    }
  });
  (addr, accepts)
}

fn client() -> C {
  Client::builder(TokioExecutor::new()).build_http()
}

// returns the client-side port of the connection that carried the request
async fn get(c: &C, url: String, read_body: bool) -> u16 {
  let res = c.get(url.parse().unwrap()).await.unwrap();
  let port = res.extensions().get::<HttpInfo>().unwrap().local_addr().port();
  if read_body {
    res.into_body().collect().await.unwrap();
  }
  port
}

async fn case(label: &str, idle: Option<Duration>, f: impl AsyncFnOnce(SocketAddr) -> (u16, u16)) {
  let (addr, accepts) = server(idle).await;
  let (p1, p2) = f(addr).await;
  println!("{label:<40} accepts: {}, same client port: {}", accepts.load(SeqCst), p1 == p2);
}

const GAP: Duration = Duration::from_millis(200);

#[tokio::main]
async fn main() {
  case("baseline: read body, same Client", None, async |a| {
    let c = client();
    let p1 = get(&c, format!("http://{a}/"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("a new Client for every request", None, async |a| {
    let p1 = get(&client(), format!("http://{a}/"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&client(), format!("http://{a}/"), true).await)
  })
  .await;
  case("1 MiB body dropped unread", None, async |a| {
    let c = client();
    let p1 = get(&c, format!("http://{a}/big"), false).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("both requests at the same time", None, async |a| {
    let c = client();
    tokio::join!(get(&c, format!("http://{a}/"), true), get(&c, format!("http://{a}/"), true))
  })
  .await;
  case("server answers connection: close", None, async |a| {
    let c = client();
    let p1 = get(&c, format!("http://{a}/close"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("server closes idle conns after 100 ms", Some(Duration::from_millis(100)), async |a| {
    let c = client();
    let p1 = get(&c, format!("http://{a}/"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("client pool_idle_timeout(100 ms)", None, async |a| {
    let c: C = Client::builder(TokioExecutor::new()).pool_idle_timeout(Duration::from_millis(100)).build_http();
    let p1 = get(&c, format!("http://{a}/"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("client pool_max_idle_per_host(0)", None, async |a| {
    let c: C = Client::builder(TokioExecutor::new()).pool_max_idle_per_host(0).build_http();
    let p1 = get(&c, format!("http://{a}/"), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://{a}/"), true).await)
  })
  .await;
  case("127.0.0.1 then localhost", None, async |a| {
    let c = client();
    let p1 = get(&c, format!("http://127.0.0.1:{}/", a.port()), true).await;
    tokio::time::sleep(GAP).await;
    (p1, get(&c, format!("http://localhost:{}/", a.port()), true).await)
  })
  .await;
}

M2 Max、rustc 1.99.0、--release，连跑 3 次输出相同：

baseline: read body, same Client         accepts: 1, same client port: true
a new Client for every request           accepts: 2, same client port: false
1 MiB body dropped unread                accepts: 2, same client port: false
both requests at the same time           accepts: 2, same client port: false
server answers connection: close         accepts: 2, same client port: false
server closes idle conns after 100 ms    accepts: 2, same client port: false
client pool_idle_timeout(100 ms)         accepts: 2, same client port: false
client pool_max_idle_per_host(0)         accepts: 2, same client port: false
127.0.0.1 then localhost                 accepts: 2, same client port: false

只有第一行复用了。下面按「连接卡在哪一步」分成三组。

根本没去同一个池子、同一个键下找

每次新建 Client。 池子是 Client 的一个字段，Builder::build 每次都 pool::Pool::new(...) 建一个新的。第一个 Client 在第一次请求后就被 drop 了，连接回池时 Pooled 的 Drop 发现 self.pool.upgrade() 失败（池子已经没了），连接随之关闭。这是实际代码里最常见的一种：在每个请求处理函数里 Client::builder(...).build_http()，或者 reqwest::Client::new()。reqwest 的 Client 里装的就是 legacy Client，同样一个 Client 一个池。

localhost 和 127.0.0.1。 池的键是 PoolKey，(Scheme, Authority)，按 URI 里写的字面值比较。两次请求解析到同一个 IP 也没用。同理，http 和 https 是两个池，经过不同代理配置的请求、带不同端口的同一主机也各算各的。

📝NOTE

没有蠢问题

问：池子为什么不按 IP 地址分组？反正最后连的是同一个 IP。

答：因为查池子在解析 DNS 之前。Client::request 一开始就用 URI 算出键，拿键去 checkout，同时才去调连接器（连接器里才有 DNS）。按 IP 分组就得先解析再查池，复用一条连接也要付一次 DNS 的代价，而省掉握手和解析本来就是复用的目的。另外，同一个 IP 上的不同主机名在 HTTPS 下证书不同，HTTP 下 Host 也可能被反向代理用来分流，按主机名分组才不会把请求送错地方。

没有回到池里

响应 body 没读完就 drop。 HTTP/1 连接要等 body 读完、连接任务再次从请求通道取请求（发出 want 信号）才回池；drop 一个 1 MiB 还没读的 body，连接只会再读一步，读不完就关掉。小 body 可能在读响应头时已经整个进了缓冲，drop 了也能回池。四种对待响应的方式各会怎样，连接池的「什么时候回到池里」一节有完整实验。

两个请求同时发。 这不是 bug。HTTP/1 一条连接同一时间只跑一个请求，第一条连接还在用，就不在空闲列表里，第二个请求只能新建。之后两条连接都回池，后面的请求就会复用它们。真正要担心的是并发量大时连接数跟着涨：pool_max_idle_per_host 只限空闲连接，不限同时在用的连接数。需要在一条连接上并发，要用 HTTP/2（见一个慢请求卡住同一连接上后面的请求）。

服务端回 connection: close。 客户端的 Conn 读到这个头，把 keep-alive 关掉（Conn 的读写状态机），响应结束后连接任务就退出。回池时 Drop 先查 is_open，已经不 ready 的连接不放回去。服务端发 connection: close，常见的是它正在停机（优雅停机实验里忙着的 HTTP/1 连接就收到了这个头），或者它本来就不做 keep-alive。

📝NOTE

冷知识：请求里写了 Connection: close，1.11.1 之前连接照样回池

hyper 的 #4110（首次发布于 v1.11.1）修的是客户端这一侧。提交说明：「hyper's client derives connection reuse from the response alone, so a request carrying Connection: close whose backend response omits it (keeps the socket alive) leaves the connection pooled and reusable.」也就是说，之前客户端只看响应：请求里说了「用完就关」，只要服务端的响应没带 close、也没关 socket，连接就被当成可复用的放回池里，下一个请求会用上一条本该关掉的连接。修法是编码请求头时检查每一行 Connection，带 close 就直接关掉这条连接的 keep-alive：「Disable keep-alive at request-encode time when the outgoing request carries a Connection: close token so the connection is evicted regardless of the response.」本站锁的 hyper 1.10.1 还没有这个修复。

回了池，下次取出前失效

服务端先关了空闲连接。 实验里服务端用 header_read_timeout(100 ms)：hyper 服务端的这个超时同时管 keep-alive 空闲时间（见服务端 Builder 与 Connection），空闲 100 ms 就关。客户端的连接任务读到 EOF 后结束，is_ready() 变假，下一次 checkout 从空闲列表弹出它时发现 is_open 为假，丢掉，新建（连接池「checkout：先看池子」一节有时间线）。客户端池子的默认 idle_timeout 是 90 秒，而服务端的空闲超时常常更短（hyper 服务端只要设了 timer，header_read_timeout 默认就是 30 秒），所以这种情况在线上很常见。危险的不是多建一条连接，而是时机正好卡在服务端关闭的那一刻：请求写到一条对端刚关掉的连接上，就成了 connection closed before message completed。

客户端池子自己判它过期。 pool_idle_timeout(100 ms)：取出时 now - idle_at > timeout 就丢弃，不需要 timer（连接池「idle_timeout：两处检查」）。把它设得比服务端的空闲超时短一点，是避开上一条竞态的常规做法，代价就是这一行看到的：空闲稍久的连接不再复用。

pool_max_idle_per_host 满了或为 0。 设成 0，Pool::new 根本不建池；设成 N，第 N+1 条要回池的空闲连接直接关掉。实验见连接池的「max_idle_per_host」一节。

修法

现象	原因	怎么办
每个请求一条新连接，没有并发	每次新建 `Client`	整个进程共用一个 `Client`（`clone` 很便宜，见下面的动脑筋）
同一个服务的请求分成好几组连接	键不同：主机名写法、scheme、端口	统一 base URL
读了响应头就扔掉响应，或者拿着不读	body 没读完	不想要内容也要把 body 读完（比如 `collect`），读完再 drop
响应都带 `connection: close`	服务端不想复用	查服务端为什么关；这种情况客户端无能为力
空闲一会儿后的第一个请求总是新连接	服务端或池子的空闲超时	正常现象；`pool_idle_timeout` 应短于服务端的空闲超时
并发时连接数跟着涨	HTTP/1 一条连接一个在途请求	正常现象；要复用就上 HTTP/2

❗IMPORTANT

动脑筋：每个请求用一个 client.clone()，复用吗？

把「每次新建 Client」换成「每次 clone 同一个 Client」，用完就 drop 掉这个克隆：

use std::convert::Infallible;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::Response;
use hyper_util::client::legacy::Client;
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let accepts = Arc::new(AtomicUsize::new(0));
  let counter = accepts.clone();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      counter.fetch_add(1, SeqCst);
      let svc = service_fn(|_| async { Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("ok")))) });
      tokio::spawn(http1::Builder::new().serve_connection(TokioIo::new(stream), svc));
    }
  });

  let client: Client<_, Empty<Bytes>> = Client::builder(TokioExecutor::new()).build_http();
  for _ in 0..3 {
    let c = client.clone();
    let res = c.get(format!("http://{addr}/").parse().unwrap()).await.unwrap();
    res.into_body().collect().await.unwrap();
    drop(c);
    tokio::time::sleep(Duration::from_millis(200)).await;
  }
  println!("3 requests, each through its own clone: accepts = {}", accepts.load(SeqCst));
}

依赖同上。打印的 accepts 是几？答案在本页末尾。

动脑筋答案