Rust in Everything
基于 hyper-util 0.1.21

HttpConnector

主线位置：客户端主线上「池里没有可用连接」的那一支。连接池里的 connect_to 把 pool key 拼回一个只有 scheme 和 authority 的 Uri，交给连接器；HttpConnector 做 DNS 和 TCP，交回一个 TokioIo<TcpStream>，之后才是 HTTP 握手（连接级客户端）。

flowchart TB
  uri["Uri"] --> hp{"get_host_port：scheme 检查"}
  hp -- 不合格 --> err1["ConnectError"]
  hp --> ip{"host 是 IP 字面量？"}
  ip -- 是 --> addrs["SocketAddrs"]
  ip -- 否 --> dns["resolver：默认 getaddrinfo"]
  dns --> addrs
  addrs --> split["按第一个地址的协议族分两组"]
  split --> pref["首选组：逐个连"]
  split --> fb["备用组：延迟到期或首选失败"]
  pref --> tcp["TcpStream"]
  fb --> tcp
  tcp --> io["set_nodelay，包成 TokioIo"]

源码在 client/legacy/connect/http.rs、dns.rs 和 mod.rs。它直接用 tokio 的 TcpSocket 和 tokio::time，所以只在 hyper-util 的 tokio feature 下存在，不走 hyper 的 rt 抽象。

连接器就是一个 Service<Uri>

legacy Client 对连接器的要求写在 sealed trait Connect 上，它对所有满足条件的 tower Service<Uri> 有一个通用实现：

    impl<S, T> Connect for S
    where
        S: tower_service::Service<Uri, Response = T> + Send + 'static,
        S::Error: Into<Box<dyn StdError + Send + Sync>>,
        S::Future: Unpin + Send,
        T: Read + Write + Connection + Unpin + Send + 'static,
    {
        type _Svc = S;

        fn connect(self, _: Internal, dst: Uri) -> crate::service::Oneshot<S, Uri> {
            crate::service::Oneshot::new(self, dst)
        }
    }

也就是说：任何 Service<Uri>，返回的 IO 实现了 hyper 的 Read + Write（见 IO trait）和 Connection，future 是 Unpin + Send，就能当连接器。每次建连接，Client 克隆一份连接器，用 Oneshot 先 poll_ready 再 call，和 tower 的 ServiceExt::oneshot 是同一个意思。连接池一页的赛跑实验就是用 tower::service_fn 包了一个慢 300ms 的连接器（Box::pin 是为了满足 Unpin）。

Connection::connected 返回 Connected，告诉 Client 这条连接的情况：ALPN 是否协商出 h2（决定用 HTTP/1 还是 HTTP/2 握手，以及连接池的 HTTP/2 锁）、是否经过代理（决定请求行写绝对形式还是 origin 形式）、一个附加值 extra、以及 poison 标记。TcpStream 的实现把 HttpInfo（本地和远端地址）放进 extra，Client 把它插进每个响应的 extensions，所以 res.extensions().get::<HttpInfo>() 能拿到对端地址。

HttpConnector 自己的 Service 实现：Response = TokioIo<TcpStream>，poll_ready 转发给 resolver，call 克隆自身，把 call_async 装箱：

    async fn call_async(&mut self, dst: Uri) -> Result<TokioIo<TcpStream>, ConnectError> {
        let config = &self.config;

        let (host, port) = get_host_port(config, &dst)?;
        let host = crate::client::strip_ipv6_brackets(host);

        // If the host is already an IP addr (v4 or v6),
        // skip resolving the dns and start connecting right away.
        let addrs = if let Some(addrs) = dns::SocketAddrs::try_parse(host, port) {
            addrs
        } else {
            let addrs = resolve(&mut self.resolver, dns::Name::new(host.into()))
                .await
                .map_err(ConnectError::dns)?;
            let addrs = addrs
                .map(|mut addr| {
                    set_port(&mut addr, port, dst.port().is_some());

                    addr
                })
                .collect();
            dns::SocketAddrs::new(addrs)
        };

        let c = ConnectingTcp::new(addrs, config);

        let sock = c.connect().await?;

        if let Err(e) = sock.set_nodelay(config.nodelay) {
            warn!("tcp set_nodelay error: {}", e);
        }

        Ok(TokioIo::new(sock))
    }

call_async 就是本页的目录：检查 Uri、解析地址、建 TCP、设 nodelay。配置都在一个 Arc<Config> 里，克隆连接器只是加一次引用计数；set_* 方法用 Arc::make_mut，改的时候如果有别的克隆，会先复制一份。

第一步：检查 Uri

get_host_port：

    if config.enforce_http {
        if dst.scheme() != Some(&Scheme::HTTP) {
            return Err(ConnectError {
                msg: INVALID_NOT_HTTP,
                addr: None,
                cause: None,
            });
        }
    } else if dst.scheme().is_none() {
        return Err(ConnectError {
            msg: INVALID_MISSING_SCHEME,
            addr: None,
            cause: None,
        });
    }

enforce_http 默认开：scheme 不是 http 就报错，不解析、不连接。关掉以后只要求有 scheme。随后取 host（没有就报 invalid URL, host is missing），端口没写时 https 用 443、其余用 80。三条报错原文是 三个静态字符串。做 SOCKS 代理的内层连接器时也要关掉它：SocksV5 交给内层的是 socks5:// 代理地址，默认的 HttpConnector 会拒绝，外面只看得到 SOCKS error: failed to create underlying connection，见 这个坑 和 代理连接器。

拿一个 https 地址分别直接调连接器、通过 Client 发请求：

hyper = "=1.12.0"
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "0.1"
tokio = { version = "=1.54.1", features = ["full"] }
tower = { version = "=0.5.3", features = ["util"] }

use std::error::Error;
use http_body_util::Empty;
use hyper::{body::Bytes, Uri};
use hyper_util::{
  client::legacy::{connect::HttpConnector, Client},
  rt::TokioExecutor,
};
use tower::ServiceExt;

#[tokio::main]
async fn main() {
  // 1. The connector on its own.
  let uri: Uri = "https://localhost:1/".parse().unwrap();
  let err = HttpConnector::new().oneshot(uri.clone()).await.unwrap_err();
  println!("connector Display: {err}");
  println!("connector Debug:   {err:?}");

  // 2. Through the legacy Client.
  let client: Client<HttpConnector, Empty<Bytes>> =
    Client::builder(TokioExecutor::new()).build_http();
  let err = client.get(uri).await.unwrap_err();
  println!("client Display:    {err}");
  println!("client is_connect: {}", err.is_connect());
  let mut src = err.source();
  while let Some(e) = src {
    println!("  caused by: {e}");
    src = e.source();
  }
  println!("client Debug:      {err:?}");
}

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

connector Display: invalid URL, scheme is not http
connector Debug:   ConnectError("invalid URL, scheme is not http")
client Display:    client error (Connect)
client is_connect: true
  caused by: invalid URL, scheme is not http
client Debug:      hyper_util::client::legacy::Error(Connect, ConnectError("invalid URL, scheme is not http"))

Client 的错误只显示 client error (Connect)，具体原因在 source() 里。日志里只打印 {} 的话只会看到前一半，这是 legacy Client 错误的通病，见 hyper::Error。

📝NOTE
没有蠢问题

问： 既然 HttpConnector 默认拒绝 https，那 reqwest 这些库是怎么连 HTTPS 的？把 enforce_http 关掉就行了吗？

答： 关掉它，HttpConnector 会老老实实连上 443 端口，交回一条明文 TCP；Client 接着在上面写明文 HTTP，对端是 TLS 服务器，只会得到握手失败或者乱码。enforce_http 默认开，就是为了让「忘了配 TLS」在第一步就报错。TLS 连接器是包在外面的另一层 Service<Uri>：hyper-tls 0.6.0 的 HttpsConnector::new 和 hyper-rustls 0.27.10 的 builder 都先构造一个 HttpConnector、调 enforce_http(false)，在它返回的 TCP 上为 https 地址做 TLS 握手（hyper-tls 对 http 地址原样透传）。reqwest 0.13.5 的连接器也调了 enforce_http(false)。这样 HttpConnector 只管 TCP，TLS 库可以换，Client 不用知道有没有 TLS（ALPN 结果通过 Connected::negotiated_h2 告诉它）。

第二步：DNS

host 已经是 IP 字面量时，SocketAddrs::try_parse 直接得到一个地址，不走 resolver。否则调 resolver。resolver 是另一个 Service：Service<Name>，返回一个 SocketAddr 迭代器（Resolve 是它的「trait 别名」）。HttpConnector::new() 用默认的 GaiResolver，自定义 resolver 用 new_with_resolver。

GaiResolver::call：

    fn call(&mut self, name: Name) -> Self::Future {
        let blocking = tokio::task::spawn_blocking(move || {
            (&*name.host, 0)
                .to_socket_addrs()
                .map(|i| SocketAddrs { iter: i })
        });

        GaiFuture { inner: blocking }
    }

to_socket_addrs 底下是系统的 getaddrinfo，它是阻塞调用，所以放进 tokio 的阻塞线程池。端口传 0，回来以后由 set_port 换成 URI 的端口（URI 写了端口时总是覆盖；没写时只覆盖 resolver 给的 0，自定义 resolver 返回的非 0 端口会保留）。GaiFuture 在 drop 时 abort 这个 JoinHandle，但已经在执行的阻塞闭包是取消不了的（tokio 卷那一页有实验），一次卡住的 getaddrinfo 会一直占着一个阻塞线程直到系统返回。

本机 localhost 解析出来是什么顺序：

use std::time::Duration;
use hyper::Uri;
use hyper_util::client::legacy::connect::{dns::{GaiResolver, Name}, HttpConnector};
use tower::ServiceExt;

#[tokio::main]
async fn main() {
  let addrs: Vec<_> = GaiResolver::new().oneshot("localhost".parse::<Name>().unwrap()).await.unwrap().collect();
  println!("localhost -> {addrs:?}");
  let mut http = HttpConnector::new();
  http.set_connect_timeout(Some(Duration::from_millis(200)));
  let err = http.oneshot("http://127.0.0.2:9/".parse::<Uri>().unwrap()).await.unwrap_err();
  println!("{err:?}");
}

连跑 3 次输出相同：

localhost -> [[::1]:0, 127.0.0.1:0]
ConnectError("tcp connect error", 127.0.0.2:9, Custom { kind: TimedOut, error: Elapsed(()) })

第一行：macOS 上 localhost 先给 IPv6。第二行留到 connect_timeout 一节解释。

第三步：Happy Eyeballs

拿到地址列表后，ConnectingTcp::new 用 split_by_preference 把它分成首选和备用两组：

            _ => {
                let preferring_v6 = self
                    .iter
                    .as_slice()
                    .first()
                    .map(SocketAddr::is_ipv6)
                    .unwrap_or(false);

                let (preferred, fallback) = self
                    .iter
                    .partition::<Vec<_>, _>(|addr| addr.is_ipv6() == preferring_v6);

                (SocketAddrs::new(preferred), SocketAddrs::new(fallback))
            }

首选的是列表第一个地址的协议族，不一定是 IPv6。resolver 先给 IPv6 就先连 IPv6，先给 IPv4 就先连 IPv4，组内保持 resolver 给的顺序。只设了 IPv4 或只设了 IPv6 本地地址（set_local_address）时，前面两个分支只留那一族，备用组为空。

备用组为空，或者 happy_eyeballs_timeout 设成 None，就没有备用：所有地址放进一组，逐个顺序连。否则备用组带一个 tokio::time::sleep(fallback_timeout)，默认 300ms（set_happy_eyeballs_timeout）。这个 Sleep 在 new 里就创建了，计时从解析完成、开始连接时算起。

📝NOTE
冷知识：「总是先连 IPv6」只活了一个版本

hyper-util 0.1.13 的 #194 把上面这段改成了 self.iter.partition::<Vec<_>, _>(SocketAddr::is_ipv6)，注释写「Happy Eyeballs says we always give a preference to v6 if available」，不管 resolver 怎么排，都先连 IPv6。CHANGELOG 记作「Fix HttpConnector to always prefer IPv6 addresses first, if happy eyeballs is enabled.」

两周后有人报告 hyper#3900「hyper-util 0.1.13 breaks reqwest for networks where IPv6 returns a connection reset」：用户的网络上 IPv6 连接会被重置，升级到 0.1.13 后 uv 就连不上 PyPI 了，退回 0.1.12 就好。0.1.14 的 #200 把代码改回按第一个地址决定，PR 的说法是「This was changed in #194, but that change seems to be wrong. The resolver is allowed to express it's priority through the ordering.」并引了 RFC 6555 第 6 节。CHANGELOG：「Fix HttpConnector to defer address family order to resolver sort order.」test_ip_addrs_split_by_preference 里 [v4, v6] 得到「首选 IPv4」的断言，也跟着 #194 改过去、又跟着 #200 改了回来。

两组怎么赛跑在 ConnectingTcp::connect：

    async fn connect(mut self) -> Result<TcpStream, ConnectError> {
        match self.fallback {
            None => self.preferred.connect(self.config).await,
            Some(mut fallback) => {
                let preferred_fut = pin!(self.preferred.connect(self.config));
                let fallback_fut = pin!(fallback.remote.connect(self.config));
                let fallback_delay = pin!(fallback.delay);

                let (result, future) =
                    match futures_util::future::select(preferred_fut, fallback_delay).await {
                        Either::Left((result, _fallback_delay)) => {
                            (result, Either::Right(fallback_fut))
                        }
                        Either::Right(((), preferred_fut)) => {
                            // Delay is done, start polling both the preferred and the fallback
                            futures_util::future::select(preferred_fut, fallback_fut)
                                .await
                                .factor_first()
                        }
                    };

                if result.is_err() {
                    // Fallback to the remaining future (could be preferred or fallback)
                    // if we get an error
                    future.await
                } else {
                    result
                }
            }
        }
    }
  • 首选组在延迟到期前有了结果：成功就用；失败就转去 await 备用组。备用组的 future 之前从没被 poll 过，这时才开始连，不再等剩下的延迟。
  • 延迟先到期：首选和备用一起跑，谁先有结果算谁；先到的是失败，就再等另一个。

每组内部由 ConnectingTcpRemote::connect 逐个地址顺序尝试，全失败时返回第一个错误（注释：「only return the first error, we assume it's the most relevant」），错误里带上那个地址。

下面不碰外部网络，用一个忽略域名、按固定顺序返回地址的 resolver 来造场景。需要一个「连了没回应」的地址：macOS 上 127.0.0.2 走 lo0，但没有分配给任何接口，SYN 发出去没有回应，连接会一直挂着（上一节的 127.0.0.2:9 就是这样超时的）。这个行为是 macOS 的，别的系统上未必如此。

hyper = "=1.12.0"
hyper-util = { version = "=0.1.21", features = ["client-legacy", "tokio"] }
tokio = { version = "=1.54.1", features = ["full"] }
tower = { version = "=0.5.3", features = ["util"] }

use std::{convert::Infallible, error::Error, net::SocketAddr, time::{Duration, Instant}};
use hyper::Uri;
use hyper_util::client::legacy::connect::{dns::Name, HttpConnector};
use tokio::net::TcpListener;
use tower::{service_fn, ServiceExt};

// 127.0.0.2 is on loopback but not assigned on macOS: SYNs vanish, connect hangs.
async fn run(label: &str, addrs: Vec<SocketAddr>, port: u16, eyeballs: Option<Duration>, timeout: Option<Duration>) {
  // A resolver that ignores the name and returns a fixed list, in order.
  let resolver = service_fn(move |_: Name| {
    let addrs = addrs.clone();
    async move { Ok::<_, Infallible>(addrs.into_iter()) }
  });
  let mut http = HttpConnector::new_with_resolver(resolver);
  http.set_happy_eyeballs_timeout(eyeballs);
  http.set_connect_timeout(timeout);
  let uri: Uri = format!("http://dual.test:{port}/").parse().unwrap();
  let start = Instant::now();
  match http.oneshot(uri).await {
    Ok(io) => println!("{label}: connected to {} after {}ms", io.inner().peer_addr().unwrap(), start.elapsed().as_millis()),
    Err(e) => println!("{label}: {e} ({}) after {}ms", e.source().unwrap(), start.elapsed().as_millis()),
  }
}

#[tokio::main]
async fn main() {
  let v4 = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let port = v4.local_addr().unwrap().port();
  let _v6 = TcpListener::bind(("::1", port)).await.unwrap();
  let black: SocketAddr = ([127, 0, 0, 2], port).into();
  let live4: SocketAddr = ([127, 0, 0, 1], port).into();
  let live6: SocketAddr = format!("[::1]:{port}").parse().unwrap();
  let ms = Duration::from_millis;

  run("A [black4, live6], default", vec![black, live6], port, Some(ms(300)), None).await;
  run("B [black4, live6], eyeballs off, timeout 1s", vec![black, live6], port, None, Some(ms(1000))).await;
  run("C [black4, live4], timeout 1s", vec![black, live4], port, Some(ms(300)), Some(ms(1000))).await;
  run("D [black4], timeout 200ms", vec![black], port, Some(ms(300)), Some(ms(200))).await;
}

M2 Max、rustc 1.99.0，连跑 3 次，连上的地址相同，毫秒数差 9ms 以内：

A [black4, live6], default: connected to [::1]:64833 after 303ms
B [black4, live6], eyeballs off, timeout 1s: connected to [::1]:64833 after 503ms
C [black4, live4], timeout 1s: connected to 127.0.0.1:64833 after 503ms
D [black4], timeout 200ms: tcp connect error (deadline has elapsed) after 203ms
  • A：第一个地址是 IPv4，所以首选组是 [127.0.0.2]，备用组是 [::1]。首选挂住，300ms 后备用组开跑，立刻连上 ::1。这就是 Happy Eyeballs 的用处：一个协议族不通时，代价是 300ms 而不是一次系统级的连接超时。
  • B、C、D 讲的是 connect_timeout，见下一节。

hyper-util 自己的测试 client_happy_eyeballs 覆盖了更多组合：首选快、首选不可达（立即失败）、首选慢（等延迟后由备用赢）、备用里先不可达再成功，每种都断言连上的协议族和耗时（允许 ±150ms）。它用的「慢」地址是 RFC 6890 保留的 198.18.0.25 和 2001:2::fe，要经过真实网络才能表现出「慢」，所以这个测试标着 #[ignore] // TODO，还要打开内部 feature __internal_happy_eyeballs_tests 才会编译运行。本页没有运行它。

connect_timeout：按地址数平分

set_connect_timeout 默认 None，此时一次连接要多久由操作系统决定。设了以后，ConnectingTcpRemote::new 把它平分给组内每个地址：

    fn new(addrs: dns::SocketAddrs, connect_timeout: Option<Duration>) -> Self {
        let connect_timeout = connect_timeout.and_then(|t| t.checked_div(addrs.len() as u32));

        Self {
            addrs,
            connect_timeout,
        }
    }

checked_div 在地址数为 0 时返回 None，避免除零 panic：这是从 hyper 0.14 移植过来的修复（#76，hyper-util 0.1.2），PR 标题是「divide by zero error when dividing duration by scalar」。

每个地址的连接在 connect 函数末尾套一层 tokio::time::timeout，超时被转成 io::ErrorKind::TimedOut，再包成 ConnectError("tcp connect error")：

    let connect = socket.connect(*addr);
    Ok(async move {
        match connect_timeout {
            Some(dur) => match tokio::time::timeout(dur, connect).await {
                Ok(Ok(s)) => Ok(s),
                Ok(Err(e)) => Err(e),
                Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
            },
            None => connect.await,
        }
        .map_err(ConnectError::m("tcp connect error"))
    })

对照上面的输出：

  • D：只有一个挂住的地址，超时 200ms，203ms 报错。Display 只有 tcp connect error，source() 是 tokio 的 Elapsed（「deadline has elapsed」）。上一节 {:?} 的输出里还能看到地址：ConnectError("tcp connect error", 127.0.0.2:9, Custom { kind: TimedOut, error: Elapsed(()) })。
  • C：两个地址都是 IPv4，没有备用组，1s 被平分成每个 500ms。127.0.0.2 500ms 后超时，换 127.0.0.1，503ms 连上。
  • B：和 A 同样的地址，但关掉 Happy Eyeballs：两族合成一组顺序连，1s 平分成 500ms，::1 在 503ms 才连上，比 A 慢了 200ms。

平分是按组算的：开着 Happy Eyeballs 时首选组和备用组各自除以自己的地址数。平分的结果是「一个挂住的地址最多耽误 timeout / n」，代价是排在后面的正常地址也只有 timeout / n 可用，地址很多而超时很短时，慢一点的正常地址可能被误判超时。

其他 socket 选项

都在 connect 里、真正发起连接之前设到 socket 上（nodelay 例外，连上之后在 call_async 里设）：

选项默认说明
set_nodelayfalseTCP_NODELAY
set_keepalive / _interval / _retries都是 None（不开）TCP keepalive；legacy Builder::build_http 在池子开着时把时间设成 pool_idle_timeout，默认 90 秒
set_reuse_addressfalseSO_REUSEADDR
set_send_buffer_size / set_recv_buffer_sizeNoneSO_SNDBUF / SO_RCVBUF
set_local_address / set_local_addressesNone绑定本地地址；只给一族时 Happy Eyeballs 只连那一族
set_interfaceNoneLinux 系用 SO_BINDTODEVICE，macOS 系用 IP_BOUND_IF
set_mark、set_tcp_user_timeoutNone只在 Linux、Android、Fuchsia 上有

设置失败的处理分两种：keepalive、nodelay、缓冲区大小、reuse_address 失败只打一条 warn!，照常连接；创建 socket、设非阻塞、绑定接口或本地地址失败，直接返回对应的 ConnectError（tcp open error、tcp bind local error 等），这一组剩下的地址也不再尝试。

❗IMPORTANT
动脑筋：IPv6 被拒绝时要等 300ms 吗？

只在 127.0.0.1 上监听，::1 的同一端口没人监听（连过去会立刻被拒绝）。resolver 先给 ::1 再给 127.0.0.1，Happy Eyeballs 用默认的 300ms。上面的 run 函数不变，main 换成：

#[tokio::main]
async fn main() {
  // Something listens on 127.0.0.1:port, nothing on [::1]:port.
  let v4 = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let port = v4.local_addr().unwrap().port();
  let addrs = vec![format!("[::1]:{port}").parse().unwrap(), ([127, 0, 0, 1], port).into()];
  run("[::1, 127.0.0.1], default", addrs, port, Some(Duration::from_millis(300)), None).await;
}

最后连上哪个地址？用了大约多少毫秒？答案在本页末尾。这也是 http://localhost:端口 打到一个只监听 127.0.0.1 的服务时，在 macOS 上走的路径（localhost 先解析出 ::1）。

动脑筋答案

M2 Max、rustc 1.99.0，连跑 3 次，都是 0ms：

[::1, 127.0.0.1], default: connected to 127.0.0.1:64917 after 0ms

首选组是 [::1]。连接被拒绝是立即返回的错误，select(preferred_fut, fallback_delay) 里首选组先完成，结果是 Err，于是 future.await 直接去跑备用组 [127.0.0.1]，不等 300ms 的延迟。300ms 只在首选组「既不成功也不失败」时才付出，比如上面实验 A 里挂住的 127.0.0.2。

小结

  • 连接器就是 Service<Uri>，返回的 IO 要实现 hyper 的 Read + Write 和 Connection；Connected 把 ALPN、代理、HttpInfo、poison 标记告诉 Client。TLS 连接器包在 HttpConnector 外面，并把它的 enforce_http 关掉。
  • enforce_http 默认开：非 http 的 URI 在解析之前就以 invalid URL, scheme is not http 失败，经 Client 后显示为 client error (Connect)，原因在 source() 里。
  • IP 字面量不走 DNS；默认 GaiResolver 在 tokio 阻塞线程池里调 getaddrinfo，端口事后补上。
  • Happy Eyeballs 首选的是 resolver 返回的第一个地址的协议族（0.1.13 曾强制 IPv6 优先，0.1.14 改回）；首选组失败就立刻转备用组，挂住才等 happy_eyeballs_timeout（默认 300ms）。
  • connect_timeout 默认不设，设了以后在每组内按地址数平分；超时报 tcp connect error，原因是 deadline has elapsed。

💬 关联讨论

发起讨论
加载中...