Rust in Everything
基于 reqwest 0.13.5

连接器

主线位置：连接主线的入口和出口。hyper-util 的连接池里没有可用连接时，用 pool key 拼出一个只有 scheme 和 authority 的 Uri，调 reqwest 的 Connector；Connector 走完 DNS、TCP、代理、TLS，交回一个 Conn，池子从它的 connected() 里读出这条连接是不是 HTTP/2、是不是明文代理。

flowchart TB
  pool["hyper-util 连接池"] --> conn["Connector::call"]
  conn -- Simple --> svc["ConnectorService::call"]
  conn -- WithLayers --> layers["用户的 connector_layer"]
  layers --> unname["Unnameable 转回 Uri"]
  unname --> svc
  svc --> uds{"unix_socket?"}
  uds -- 是 --> local["connect_local_transport"]
  svc --> px{"代理拦截?"}
  px -- 是 --> via["connect_via_proxy"]
  px -- 否 --> direct["connect_with_maybe_proxy"]
  direct --> https["HttpsConnector"]
  https --> http["HttpConnector：DNS 和 TCP"]
  https --> tls["TLS 握手"]
  tls --> out["Conn"]
  via --> out
  local --> out
  out --> pool

源码在 connect.rs，2139 行。真正做事的不到 500 行：ConnectorBuilder、ConnectorService 的三个连接函数、Conn。剩下大半是同一段 impl 按「TCP / Unix socket / Windows 命名管道」乘「native-tls / rustls」重复的 TlsInfoFactory 和 Connection，再加 socks 和 verbose 两个小模块。DNS 见 DNS，TLS 配置见 TLS 配置，代理隧道见 代理，连接拿回去以后怎么复用见 连接池与 HTTP 版本。

实验平台：M2 Max、macOS 27、rustc 1.99.0，reqwest 0.13.5，feature 用卷里的模板（默认 feature 加 json、stream、blocking、cookies、gzip、multipart、hickory-dns、socks、query、form），release 构建。这台机器开着系统代理（scutil --proxy 显示 HTTP/HTTPS 代理 127.0.0.1:7890，例外列表里没有 127.0.0.1），所以每个客户端都调了 .no_proxy()。服务器都是同一进程里起在 127.0.0.1 上的 hyper 服务器，HTTPS 用 rcgen 现签的证书。

三个类型

ClientBuilder::build 里先建 resolver（见 DNS），再建一个 hyper-util 的 HttpConnector：

            let mut http = HttpConnector::new_with_resolver(resolver.clone());
            http.set_connect_timeout(config.connect_timeout);

然后按 config.tls 的值把它交给 ConnectorBuilder::new_rustls_tls、new_native_tls 或 from_built_native_tls（分支怎么选见 TLS 配置），得到一个 ConnectorBuilder。接着设 connect_timeout、connection_verbose、TCP keepalive，最后一句把它变成 Connector 交给 hyper-util 的 Client：

        let hyper_client = builder.build(connector_builder.build(config.connector_layers));

三个类型的分工：

  • ConnectorBuilder：构造期的可变配置。set_keepalive 这类方法要伸进 Inner 里的 HttpConnector 去改，所以每个都是一个三分支的 match。
  • ConnectorService：真正连接的 Service<Uri>。字段和 builder 差不多：inner（TLS 分支）、proxies、verbose、simple_timeout、nodelay、tls_info、user_agent（给 CONNECT 请求用），开了 socks 还有一个 resolver。
  • Connector：交给 hyper-util 的类型，一个两分支的枚举：
#[derive(Clone)]
pub(crate) enum Connector {
    // base service, with or without an embedded timeout
    Simple(ConnectorService),
    // at least one custom layer along with maybe an outer timeout layer
    // from `builder.connect_timeout()`
    WithLayers(BoxCloneSyncService<Unnameable, Conn, BoxError>),
}

hyper-util 对连接器的要求是「一个 Service<Uri>，future 是 Unpin + Send，返回的 IO 实现 Read + Write + Connection」（见 hyper 卷的 HttpConnector）。Connector 的 Future 是 Connecting，即 Pin<Box<dyn Future<Output = Result<Conn, BoxError>> + Send>>，装箱后天然 Unpin。每建一条新连接，hyper-util 克隆一份 Connector、调一次 call；连接池命中时连接器根本不会被调用。

build：没有层就不装箱

ConnectorBuilder::build 先把字段搬进 ConnectorService，然后分两种情况：

        if layers.is_empty() {
            // we have no user-provided layers, only use concrete types
            base_service.simple_timeout = self.timeout;
            return Connector::Simple(base_service);
        }

        // otherwise we have user provided layers
        // so we need type erasure all the way through
        // as well as mapping the unnameable type of the layers back to Uri for the inner service
        let unnameable_service = ServiceBuilder::new()
            .layer(MapRequestLayer::new(|request: Unnameable| request.0))
            .service(base_service);
        let mut service = BoxCloneSyncService::new(unnameable_service);

        for layer in layers {
            service = ServiceBuilder::new().layer(layer).service(service);
        }

        // now we handle the concrete stuff - any `connect_timeout`,
        // plus a final map_err layer we can use to cast default tower layer
        // errors to internal errors
        match self.timeout {
            Some(timeout) => {
                let service = ServiceBuilder::new()
                    .layer(TimeoutLayer::new(timeout))
                    .service(service);
                let service = ServiceBuilder::new()
                    .map_err(|error: BoxError| cast_to_internal_error(error))
                    .service(service);
                let service = BoxCloneSyncService::new(service);

                Connector::WithLayers(service)
            }
            // ...
        }
  • 没有用户层（绝大多数情况）：Connector::Simple，connect_timeout 存进 simple_timeout，在 ConnectorService::call 里用 tokio::time::timeout 直接包住连接 future。字段注释说这样省掉一层 Box::pin，因为 tokio::time::Timeout 是 Unpin。
  • 有用户层：最里面一层 MapRequestLayer 把 Unnameable 拆回 Uri，再装进 BoxCloneSyncService（类型擦除，见 tower 卷的 Box 与类型擦除）。用户的层按添加顺序一层层往外套，后加的在外面；connect_timeout 变成一个 tower TimeoutLayer 套在最外面；最外面再加一个 map_err，用 cast_to_internal_error 把 tower 的 Elapsed 换成 reqwest 自己的 TimedOut，这样 Error::is_timeout() 认得出来。
📝NOTE
冷知识：connector_layer 是 0.12.10 才有的

在它之前，Connector 是一个 struct，用户只能通过 builder 上的开关影响它。#2496（「allow pluggable tower layers in connector service stack」，首次发布在 v0.12.10）加了 connector_layer，同时把 Connector 改成现在的 Simple / WithLayers 两个分支：ConnectorService 新增的 simple_timeout 字段，注释写明「When there is a single timeout layer and no other layers, we embed it directly inside our base Service::call()」，就是为了不给不用这个功能的人加一层装箱。同一个 PR 加的例子 examples/connect_via_lower_priority_tokio_runtime.rs 说明了动机：把含 TLS 握手的连接调用转到另一个低优先级的 tokio 运行时，「long-running futures during handshake crypto operations don't block other I/O futures」。同一个 PR 还在 error.rs 加了 cast_to_internal_error。

connector_layer：层看到的是 Unnameable

connector_layer 的签名：

    pub fn connector_layer<L>(mut self, layer: L) -> ClientBuilder
    where
        L: Layer<BoxedConnectorService> + Clone + Send + Sync + 'static,
        L::Service:
            Service<Unnameable, Response = Conn, Error = BoxError> + Clone + Send + Sync + 'static,
        <L::Service as Service<Unnameable>>::Future: Send + 'static,

请求类型不是 Uri，是 Unnameable；响应是 Conn。两者都定义在私有的 sealed 模块里，用户写不出它们的名字，只能写对请求和响应类型泛型的层（TimeoutLayer、ConcurrencyLimitLayer、自己的计数层）。这是有意的：reqwest 以后改连接器的内部类型，不会破坏用户代码。代价是层里拿不到目标地址做路由，只能 Debug 打印它。

一个只计数和打印的层：

#[derive(Clone)]
struct Count<S> {
  inner: S,
  n: Arc<AtomicUsize>,
}
impl<S, R> Service<R> for Count<S>
where
  S: Service<R>,
  S::Future: Send + 'static,
  R: std::fmt::Debug,
{
  type Response = S::Response;
  type Error = S::Error;
  type Future = Pin<Box<dyn Future<Output = Result<S::Response, S::Error>> + Send>>;
  fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
    self.inner.poll_ready(cx)
  }
  fn call(&mut self, req: R) -> Self::Future {
    let k = self.n.fetch_add(1, Ordering::SeqCst) + 1;
    println!("  layer: call #{k} with {req:?}");
    Box::pin(self.inner.call(req))
  }
}

// ...
  let client = Client::builder()
    .no_proxy()
    .connector_layer(layer_fn(move |inner| Count { inner, n: n2.clone() }))
    .build()
    .unwrap();
  for _ in 0..3 {
    let r = client.get(format!("http://127.0.0.1:{}/", http.port())).send().await.unwrap();
    let _ = r.text().await.unwrap();
  }
  println!("  connector called {} time(s)", n.load(Ordering::SeqCst));
❗IMPORTANT
动脑筋：同一个 Client 对同一个 HTTP/1.1 服务器顺序发 3 个请求，每个都把响应体读完。计数层会被调用几次？打印出来的请求长什么样？

ConnectorService::call：三条路

call：

    fn call(&mut self, dst: Uri) -> Self::Future {
        log::debug!("starting new connection '{:?}'", dst.host());
        let timeout = self.simple_timeout;

        // Local transports (UDS, Windows Named Pipes) skip proxies
        #[cfg(any(unix, target_os = "windows"))]
        if self.should_use_local_transport() {
            return Box::pin(with_timeout(
                self.clone().connect_local_transport(dst),
                timeout,
            ));
        }

        for prox in self.proxies.iter() {
            if let Some(intercepted) = prox.intercept(&dst) {
                return Box::pin(with_timeout(
                    self.clone().connect_via_proxy(dst, intercepted),
                    timeout,
                ));
            }
        }

        Box::pin(with_timeout(
            self.clone().connect_with_maybe_proxy(dst, false),
            timeout,
        ))
    }
  1. 设了 unix_socket(path)（Windows 上是 windows_named_pipe）：不管 URL 里写的是哪个主机，都连这个本地套接字，代理全部忽略（build 里还会清空 proxies，打一条 trace 日志）。URL 是 https 时照样在套接字上做 TLS。
  2. 代理列表里第一个拦截这个 Uri 的代理：走 connect_via_proxy。HTTPS 目标先发 CONNECT 建隧道再做 TLS，HTTP 目标直接连代理、以绝对形式写请求行；SOCKS 走另一条。细节见 代理。系统代理在 build 时作为 ProxyMatcher::system() 追加在列表末尾，所以 .no_proxy() 之后这个循环是空的。
  3. 其余：connect_with_maybe_proxy(dst, false)。

三条路都包在 with_timeout 里，timeout 为 None 时直接 await。

Inner：TLS 分支在构造时就定了

#[derive(Clone)]
enum Inner {
    #[cfg(not(feature = "__tls"))]
    Http(HttpConnector),
    #[cfg(feature = "__native-tls")]
    NativeTls(HttpConnector, TlsConnector),
    #[cfg(any(feature = "__rustls"))]
    RustlsTls {
        http: HttpConnector,
        tls: Arc<rustls::ClientConfig>,
        tls_proxy: Arc<rustls::ClientConfig>,
    },
}

Inner 选哪个变体由 ClientBuilder::build 决定，一个 Client 一辈子只用一种 TLS 实现。两个细节：

  • 编进任何 TLS 实现后，Http 变体就不存在了。 明文 http:// 请求也走 NativeTls 或 RustlsTls 分支：hyper-tls / hyper-rustls 的 HttpsConnector 对 http scheme 原样透传 TCP，返回 MaybeHttpsStream::Http。为此三个构造函数都对 HttpConnector 调了 enforce_http(false)（hyper-util 默认拒绝非 http 的 scheme，见 hyper 卷的 HttpConnector）。
  • rustls 分支存两份配置。 new_rustls_tls 在有代理时克隆一份 ClientConfig，清空 ALPN，作为 tls_proxy：
        let (tls, tls_proxy) = if proxies.is_empty() {
            let tls = Arc::new(tls);
            (tls.clone(), tls)
        } else {
            let mut tls_proxy = tls.clone();
            tls_proxy.alpn_protocols.clear();
            (Arc::new(tls), Arc::new(tls_proxy))
        };

tls_proxy 用于和 HTTPS 代理本身握手：隧道里的 CONNECT 请求按 HTTP/1.1 写，不能让代理在 ALPN 里选成 h2。和目标服务器握手仍用带 ALPN 的 tls。这份拷贝来自 #466（「fix proxy missing alpn」，提交里的一条是「disable alpn for proxy connection」）。

connect_with_maybe_proxy 的 rustls 分支：

            Inner::RustlsTls { http, tls, .. } => {
                let mut http = http;

                // Disable Nagle's algorithm for TLS handshake
                //
                // https://www.openssl.org/docs/man1.1.1/man3/SSL_connect.html#NOTES
                if !self.nodelay && (dst.scheme() == Some(&Scheme::HTTPS)) {
                    http.set_nodelay(true);
                }

                let mut http = hyper_rustls::HttpsConnector::from((http, tls));
                let io = http.call(dst).await?;

                if let hyper_rustls::MaybeHttpsStream::Https(stream) = io {
                    if !self.nodelay {
                        let (io, _) = stream.inner().get_ref();
                        io.inner().inner().set_nodelay(false)?;
                    }
                    Ok(Conn {
                        inner: self.verbose.wrap(RustlsTlsConn { inner: stream }),
                        is_proxy,
                        tls_info: self.tls_info,
                    })
                } else {
                    Ok(Conn {
                        inner: self.verbose.wrap(io),
                        is_proxy,
                        tls_info: false,
                    })
                }
            }

DNS 和 TCP 都在 HttpsConnector 内部调用的 HttpConnector 里完成；reqwest 自己只做两件事。一是 Nagle：tcp_nodelay 默认就是 true，这段只在用户关掉它时起作用：握手期间临时打开 TCP_NODELAY，握手完再关回去（注释指向 OpenSSL 文档里 SSL_connect 的说明，这段来自 2019 年的提交 f5e7e88「disable nagle algorithm for TLS handshake」）。二是把结果包成 Conn：HTTPS 包一层 RustlsTlsConn，明文直接用。native-tls 分支结构相同，换成 hyper_tls::HttpsConnector 和 NativeTlsConn。

Conn：一个装箱的连接，加两个标记

    pin_project! {
        /// Note: the `is_proxy` member means *is plain text HTTP proxy*.
        /// This tells hyper whether the URI should be written in
        /// * origin-form (`GET /just/a/path HTTP/1.1`), when `is_proxy == false`, or
        /// * absolute-form (`GET http://foo.bar/and/a/path HTTP/1.1`), otherwise.
        #[allow(missing_debug_implementations)]
        pub struct Conn {
            #[pin]
            pub(super)inner: BoxConn,
            pub(super) is_proxy: bool,
            // Only needed for __tls, but #[cfg()] on fields breaks pin_project!
            pub(super) tls_info: bool,
        }
    }

Conn 的 inner 是 Box<dyn AsyncConnWithInfo>：TCP、TLS over TCP、TLS over 隧道、Unix socket 都擦成同一个类型，hyper-util 的池子里只有一种连接。Read / Write 全部转发给 inner。

is_proxy 只在「明文 HTTP 经代理」时为 true（connect_via_proxy 对 HTTP 目标调 connect_with_maybe_proxy(proxy_dst, true)）；走 CONNECT 隧道的 HTTPS 是 false，因为隧道里就是直连。

hyper-util 拿到连接后调 Connection::connected：

    impl Connection for Conn {
        fn connected(&self) -> Connected {
            let connected = self.inner.connected().proxy(self.is_proxy);
            #[cfg(feature = "__tls")]
            if self.tls_info {
                if let Some(tls_info) = self.inner.tls_info() {
                    connected.extra(tls_info)
                } else {
                    connected
                }
            } else {
                connected
            }
            #[cfg(not(feature = "__tls"))]
            connected
        }
    }

inner.connected() 一路转到最里层：TCP 的实现放进 HttpInfo（本地和对端地址，Response::remote_addr() 读的就是它），RustlsTlsConn 在 ALPN 协商出 h2 时再调 negotiated_h2()：

    impl Connection for RustlsTlsConn<TokioIo<MaybeHttpsStream<TokioIo<TcpStream>>>> {
        fn connected(&self) -> Connected {
            if self.inner.inner().get_ref().1.alpn_protocol() == Some(b"h2") {
                self.inner
                    .inner()
                    .get_ref()
                    .0
                    .inner()
                    .connected()
                    .negotiated_h2()
            } else {
                self.inner.inner().get_ref().0.inner().connected()
            }
        }
    }

hyper-util 据此决定在这条连接上做 HTTP/1 还是 HTTP/2 握手（ALPN 列表怎么来的见 TLS 配置，池子怎么处理 HTTP/2 连接见 连接池与 HTTP 版本）。Connected 的 extra 由 legacy Client 插进每个响应的 extensions，同一条连接上的所有响应都带着它。

verbose 是另一层包装：connection_verbose(true) 且 log 开了 TRACE 级别时，Wrapper::wrap 把连接包进 Verbose，每次读写打一行带随机连接 id 的 trace 日志，内容是转义后的原始字节（TLS 连接上是解密后的明文）。两个条件缺一个就不包，没有额外开销。

TlsInfo

tls_info(true) 打开上面那个 if self.tls_info。TlsInfo 两个字段：对端叶子证书的 DER、协商出的 TLS 版本。rustls 的 TlsInfoFactory 从 peer_certificates() 和 protocol_version() 取；native-tls 拿不到版本，version 恒为 None。版本字段是 0.13.5 才加的（#3067，「expose the negotiated TLS version via TlsInfo」）。

取法是 resp.extensions().get::<reqwest::tls::TlsInfo>()。同一个 rustls 服务器，分别用默认 rustls 后端和 native-tls 后端（另一个同时打开 native-tls feature 的 crate，见 TLS 配置）：

client: 200 OK HTTP/2.0, tls version Some(Version(Tls1_3)), peer cert 343 bytes

client: 200 OK HTTP/2.0, tls version None, peer cert 343 bytes

经 CONNECT 隧道的 HTTPS 连接拿不到 TlsInfo。connect_via_proxy 的两个 TLS 分支在构造 Conn 时把 tls_info 写死成 false：

                    return Ok(Conn {
                        inner: self.verbose.wrap(RustlsTlsConn {
                            inner: TokioIo::new(io),
                        }),
                        is_proxy: false,
                        tls_info: false,
                    });

SOCKS 的 rustls 分支也一样（native-tls 的 SOCKS 分支却用了 self.tls_info）。用一个三十行的 CONNECT 代理验证（它对任何 CONNECT 都回 200，再把字节转发到本地 HTTPS 服务器）：

via_proxy=false: 200 OK HTTP/2.0, TlsInfo present: true
proxy: CONNECT api.test:57561 HTTP/1.1
via_proxy=true: 200 OK HTTP/2.0, TlsInfo present: false

连接本身没问题，HTTP/2 也照常协商出来了，只是扩展里没有 TlsInfo。要靠 TlsInfo 做证书钉扎（pinning）的代码，一旦环境里配了代理（包括系统代理），就会拿到 None。

connect_timeout 管两处

connect_timeout 的值在 build 里用了两次：

  1. http.set_connect_timeout(config.connect_timeout)：hyper-util 的 HttpConnector 只拿它约束 TCP 建连，并且按解析出的地址数平分（见 hyper 卷 HttpConnector 的「connect_timeout：按地址数平分」，在 reqwest 里的效果见 DNS）。
  2. connector_builder.set_timeout(config.connect_timeout)：ConnectorService 外面的 with_timeout 或最外层的 TimeoutLayer，约束整个连接器调用：DNS、TCP、代理握手、TLS 握手，加上用户的 connector_layer。

第二处是关键。一个接受 TCP、读掉 ClientHello、永远不回的服务器，三种超时各设 300ms：

  timed("connect_timeout(300ms)", Client::builder().connect_timeout(ms(300)), https_silent.clone()).await;
  timed("timeout(300ms)", Client::builder().timeout(ms(300)), https_silent.clone()).await;
  timed(
    "connector_layer(TimeoutLayer 300ms)",
    Client::builder().connector_layer(tower::timeout::TimeoutLayer::new(ms(300))),
    https_silent.clone(),
  )
  .await;

连跑 3 次，毫秒数差 4ms 以内：

== TLS server that never answers
connect_timeout(300ms): after 306ms is_connect=true is_timeout=true is_dns=false
    error sending request for url (https://127.0.0.1:64697/) <- client error (Connect) <- operation timed out
timeout(300ms): after 302ms is_connect=false is_timeout=true is_dns=false
    error sending request for url (https://127.0.0.1:64697/) <- operation timed out
connector_layer(TimeoutLayer 300ms): after 302ms is_connect=true is_timeout=true is_dns=false
    error sending request for url (https://127.0.0.1:64697/) <- client error (Connect) <- operation timed out
  • connect_timeout 在 TCP 早已连上之后的 TLS 握手阶段触发，错误链里有 hyper-util 的 client error (Connect)，所以 is_connect() 为真。
  • 总超时 timeout 也在 300ms 触发，但它在连接器外面（请求的 future 上），错误链里没有 Connect，is_connect() 为假。各个超时的完整分工见 超时。
  • 自己套的 TimeoutLayer 被 cast_to_internal_error 换成了同一个 TimedOut，和 connect_timeout 的错误看不出区别。

三个都不设时，这个请求永远不会结束：同一个服务器，外面套 5 秒的 tokio::time::timeout，输出 still pending after 5s: true。Client::new() 的 connect_timeout、timeout、read_timeout 默认都是 None。

动脑筋答案

== a counting connector_layer, HTTP/1.1 server, 3 sequential requests
  layer: call #1 with Unnameable(http://127.0.0.1:64705/)
  connector called 1 time(s)
  • 1 次。 第一个请求时池子是空的，hyper-util 调连接器；响应体读完后连接回到池里，后两个请求直接复用，连接器不再被调用。connector_layer 是「每条新连接一次」，不是「每个请求一次」，拿它做请求计数、限流会数错。
  • 层收到的是 Unnameable(http://127.0.0.1:64705/)：hyper-util 用 pool key 拼出的 Uri，只有 scheme 和 authority，路径固定是 /，请求的真实路径在这一层看不到。

小结

  • ClientBuilder::build 建 HttpConnector，按 TLS 后端选 ConnectorBuilder 的构造函数，build(layers) 得到 Connector 交给 hyper-util；连接器只在池子里没有可用连接时被调用。
  • 没有 connector_layer 时是 Connector::Simple，不装箱；有层时类型擦除，请求类型是不可命名的 Unnameable，后加的层在外，connect_timeout 在最外。
  • ConnectorService::call 三条路：本地套接字、代理、直连；TLS 实现由 Inner 在构造时定死，编进 TLS 后明文请求也走 HttpsConnector 透传。
  • Conn 把各种连接擦成一个 Box<dyn ...>，connected() 带回明文代理标记、negotiated_h2 和可选的 TlsInfo；经 CONNECT 隧道或 rustls 的 SOCKS 时 TlsInfo 恒缺。
  • connect_timeout 一份值用两处：hyper-util 里按地址平分的 TCP 超时，和包住整个连接器（DNS、TCP、代理、TLS）的外层超时；三种超时都不设，卡在握手的请求永远不结束。

💬 关联讨论

发起讨论
加载中...