Rust中如何在单个连接上并发发送多个HTTP/2请求？

问题描述 #

当前使用hyper legacy版本编写的代码，每次循环都会新建TLS连接。期望复用单个连接（仅一次TLS握手）并并发发送请求，尝试用Arc共享Client实例但未达到预期效果，同时新hyper crate文档不完善，寻求解决方案。

当前代码 #

use bytes::Bytes;
use http_body_util::Empty;
use hyper_tls::HttpsConnector;
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
use std::sync::Arc;
use std::time::Duration;
use tokio::{task::JoinHandle, time::Instant};

async fn connection_request(
    requests: u32,
    client: Arc<
        Client<HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>, Empty<Bytes>>,
    >,
) {
    for _ in 0..requests {
        if let Ok(_) = client
            .get("https://www.example.com/".parse().unwrap())
            .await
        {
            //  println!("status: {}", res.status());
        }
    }
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let requests = 100;
    let https = HttpsConnector::new();

    let client = Client::builder(TokioExecutor::new())
        .pool_idle_timeout(Duration::from_secs(30))
        .build::<_, Empty<Bytes>>(https);
    let shared_client = Arc::new(client);

    let mut handles: Vec<JoinHandle<()>> = vec![];
    let total_connections = 50;

    let start = Instant::now();

    for _ in 0..total_connections {
        let client_clone = shared_client.clone();
        handles.push(tokio::spawn(async move {
            connection_request(requests / total_connections, client_clone).await;
        }));
    }

    for handle in handles {
        let _ = handle.await;
    }

    let elapsed = start.elapsed();
    println!("Total time: {:?}", elapsed);

    Ok(())
}

解决方案 #

要实现单个TLS连接并发请求，核心是利用HTTP/2的多路复用特性（HTTP/1.1无法在单个连接上并发处理请求，仅能串行复用），同时配置hyper Client强制使用HTTP/2并限制连接池为单个连接。

步骤1：配置TLS支持HTTP/2 #

修改HttpsConnector的构造，启用ALPN协议优先选择HTTP/2：

// 添加必要依赖：rustls、rustls-native-certs
use rustls::ClientConfig;
use rustls_native_certs::load_native_certs;

// 创建支持HTTP/2的TLS配置
let mut tls_config = ClientConfig::builder()
    .with_safe_defaults()
    .with_root_certificates(load_native_certs().expect("Failed to load root certificates"))
    .with_no_client_auth();
// 设置ALPN协议，优先使用HTTP/2
tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];

// 用自定义TLS配置构建HttpsConnector
let https = HttpsConnector::new_with_tls_config(Arc::new(tls_config));

步骤2：限制Client连接池为单个连接 #

在构建Client时，添加max_idle_per_host(1)配置，确保每个目标主机仅保留1个空闲连接：

let client = Client::builder(TokioExecutor::new())
    .pool_idle_timeout(Duration::from_secs(30))
    .max_idle_per_host(1) // 限制单主机空闲连接数为1
    .build::<_, Empty<Bytes>>(https);

修改后的完整代码 #

use bytes::Bytes;
use http_body_util::Empty;
use hyper_tls::HttpsConnector;
use hyper_util::{client::legacy::Client, rt::TokioExecutor};
use rustls::ClientConfig;
use rustls_native_certs::load_native_certs;
use std::sync::Arc;
use std::time::Duration;
use tokio::{task::JoinHandle, time::Instant};

async fn connection_request(
    requests: u32,
    client: Arc<
        Client<HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>, Empty<Bytes>>,
    >,
) {
    for _ in 0..requests {
        if let Ok(_) = client
            .get("https://www.example.com/".parse().unwrap())
            .await
        {
            // println!("status: {}", res.status());
        }
    }
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let requests = 100;

    // 配置TLS支持HTTP/2
    let mut tls_config = ClientConfig::builder()
        .with_safe_defaults()
        .with_root_certificates(load_native_certs().expect("Failed to load root certificates"))
        .with_no_client_auth();
    tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];

    let https = HttpsConnector::new_with_tls_config(Arc::new(tls_config));

    // 构建Client并限制单主机连接数
    let client = Client::builder(TokioExecutor::new())
        .pool_idle_timeout(Duration::from_secs(30))
        .max_idle_per_host(1)
        .build::<_, Empty<Bytes>>(https);
    let shared_client = Arc::new(client);

    let mut handles: Vec<JoinHandle<()>> = vec![];
    let total_connections = 50;

    let start = Instant::now();

    for _ in 0..total_connections {
        let client_clone = shared_client.clone();
        handles.push(tokio::spawn(async move {
            connection_request(requests / total_connections, client_clone).await;
        }));
    }

    for handle in handles {
        let _ = handle.await;
    }

    let elapsed = start.elapsed();
    println!("Total time: {:?}", elapsed);

    Ok(())
}

关键说明 #

1. HTTP/2多路复用：同一TLS连接可同时处理多个请求，实现真正的并发，而HTTP/1.1只能在单个连接上串行处理请求。
2. 连接池配置：max_idle_per_host(1)确保Client不会为同一主机创建多余的空闲连接，强制复用已建立的单个连接。
3. ALPN协议：通过配置ALPN让TLS握手时优先协商HTTP/2，若服务器不支持则降级为HTTP/1.1。

内容的提问来源于stack exchange，提问作者PremadeS