如何在Rust Hyper中分离请求处理与响应返回逻辑？

我正在使用Rust Hyper实现HTTP监听器，目前handle_request()函数同时负责接收客户端请求并向客户端返回响应。我希望让handle_request()不返回响应，转而在另一个线程（如tokio::spawn(response_worker(/*...*/))）中处理响应返回。但我了解Hyper的service_fn()必须返回Result<Response<Body>, _>类型，因此想询问是否存在分离请求与响应的可行方案。

以下是我的代码片段：

#[tokio::main]
async fn main() -> Result<()> {

//...//

tokio::spawn(process_data(rx.clone(), processor_args, tokenizer.clone()));

    let addr = ([172, 17, 0, 4], 6006).into();
    let cookie_jar = CookieJar::new();
    let make_svc = make_service_fn(|_conn| {
        let cookie_jar = cookie_jar.clone();
        let vec_queue = vec_queue.clone(); 
        let tx = tx.clone();
        let tokenizer = tokenizer.clone();
        async move {
            let handle_request_task = tokio::task::spawn(async move {
                service_fn(move |req| {
                    handle_request(req, cookie_jar.clone(), tx.clone(), vec_queue.clone(), tokenizer.clone())
                })
            });
            core::result::Result::Ok::<_, hyper::Error>(handle_request_task.await.expect("failed to spawn handle_request task"))
        }
    });

}


async fn handle_request(
    req: Request<Body>,
    mut cookie_jar: CookieJar,
    tx: Sender<IdsWithToken, i32>,
    vec_queue: Arc<Mutex<VecDeque<IdWithMessage>>>,
    mut tokenizer: Tokenizer,
) -> Result<Response<Body>, anyhow::Error> {
    
    //...//

    let request_message = IdsWithToken {
        session_id: session_id.clone(),
        token: token.clone(),
    };
    println!("request_message: {:?}", request_message);
    let _ = tx.send(request_message.clone(), 0).await;


    if let Some(final_response) = process_response_from_queue(vec_queue.clone(), request_message.session_id.clone()).await {
        Ok(Response::new(Body::from(format!(
            "Data from priority queue: {}",
            final_response.message
        ))))
    } else {
        Ok(Response::new(Body::from(
            "No data available in the priority queue",
        )))
    }
}

我期望将响应返回逻辑转移至类似以下的独立线程中：

tokio::spawn(response_worker(/*...*/));
// 希望在此处处理请求的响应返回

核心思路是利用Tokio通道结合Hyper的响应机制，让handle_request立即返回一个挂起的响应，后续由worker线程异步填充响应内容。以下是两种可行方案：

方案1：流式Body + MPSC通道（推荐） #

通过MPSC通道让worker线程向响应流发送数据，客户端会持续等待直到流结束，适合需要异步返回响应的场景：

use hyper::{Body, Request, Response, service::{make_service_fn, service_fn}};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use bytes::Bytes;
use std::convert::Infallible;

async fn handle_request(
    req: Request<Body>,
    mut cookie_jar: CookieJar,
    tx: Sender<IdsWithToken, i32>,
    vec_queue: Arc<Mutex<VecDeque<IdWithMessage>>>,
    mut tokenizer: Tokenizer,
) -> Result<Response<Body>, anyhow::Error> {
    // 创建MPSC通道，用于worker和响应流通信
    let (resp_tx, resp_rx) = mpsc::channel(1);
    // 将Receiver转换为Stream，再包装成Hyper的Body
    let body = Body::wrap_stream(ReceiverStream::new(resp_rx));

    // 提取请求相关参数
    let session_id = /* 从请求中获取session_id的逻辑 */;
    let request_message = IdsWithToken {
        session_id: session_id.clone(),
        token: token.clone(),
    };
    println!("request_message: {:?}", request_message);
    let _ = tx.send(request_message.clone(), 0).await;

    // 启动响应处理worker
    tokio::spawn(response_worker(
        resp_tx,
        vec_queue.clone(),
        session_id.clone()
    ));

    // 立即返回包含流式Body的响应
    Ok(Response::new(body))
}

// 独立的响应处理worker
async fn response_worker(
    resp_tx: mpsc::Sender<Result<Bytes, hyper::Error>>,
    vec_queue: Arc<Mutex<VecDeque<IdWithMessage>>>,
    session_id: String,
) {
    // 从队列中等待响应数据
    let response_content = if let Some(final_response) = process_response_from_queue(vec_queue, session_id).await {
        format!("Data from priority queue: {}", final_response.message)
    } else {
        "No data available in the priority queue".to_string()
    };

    // 发送响应内容到通道
    let _ = resp_tx.send(Ok(Bytes::from(response_content))).await;
    // 通道关闭后，Hyper会自动结束响应流
}

方案2：Oneshot通道（适合一次性响应） #

如果每个请求只需要返回一次响应，可以用专门用于单消息传递的Oneshot通道：

use tokio::sync::oneshot;

async fn handle_request(
    req: Request<Body>,
    mut cookie_jar: CookieJar,
    tx: Sender<IdsWithToken, i32>,
    vec_queue: Arc<Mutex<VecDeque<IdWithMessage>>>,
    mut tokenizer: Tokenizer,
) -> Result<Response<Body>, anyhow::Error> {
    // 创建Oneshot通道
    let (resp_tx, resp_rx) = oneshot::channel();

    // 提取请求参数
    let session_id = /* 从请求中获取session_id的逻辑 */;
    let request_message = IdsWithToken {
        session_id: session_id.clone(),
        token: token.clone(),
    };
    println!("request_message: {:?}", request_message);
    let _ = tx.send(request_message.clone(), 0).await;

    // 启动响应worker
    tokio::spawn(response_worker_oneshot(
        resp_tx,
        vec_queue.clone(),
        session_id.clone()
    ));

    // 等待worker返回响应内容
    let content = resp_rx.await.expect("worker failed to send response");
    Ok(Response::new(Body::from(content)))
}

async fn response_worker_oneshot(
    resp_tx: oneshot::Sender<String>,
    vec_queue: Arc<Mutex<VecDeque<IdWithMessage>>>,
    session_id: String,
) {
    let content = if let Some(final_response) = process_response_from_queue(vec_queue, session_id).await {
        format!("Data from priority queue: {}", final_response.message)
    } else {
        "No data available in the priority queue".to_string()
    };
    // 发送响应内容
    let _ = resp_tx.send(content);
}

关键注意事项 #

• 务必处理通道发送失败的情况（比如客户端提前断开连接），避免任务泄漏
• 使用流式Body时，worker最终要确保通道关闭，否则客户端会一直处于等待状态
• Hyper的service_fn必须返回Result<Response<Body>, _>，无法完全不返回响应，只能通过异步方式延迟填充响应内容

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