如何在Hyper/Axum Web服务器中处理带计算密集型任务的高吞吐量请求

• 消费端串行执行任务：你在rx.recv()循环中直接await了spawn_blocking返回的JoinHandle，导致同一时间只能处理1个计算任务，完全没有发挥并行能力。容量仅32的通道会被瞬间打满，后续所有请求都会阻塞在tx.send(data).await处，大量请求上下文、请求字节、解码后的WriteRequest对象堆积在内存中，直接触发内存暴涨。
• 无负载保护机制：2000万QPS的流量远超单节点CPU密集型任务的处理上限，没有限流降级策略的情况下，请求堆积速度始终快于处理速度，内存占用会持续升高直到OOM。
• 未限制阻塞任务并发数：Tokio默认spawn_blocking线程池大小为CPU核数*50，无限制提交任务会导致线程数过多，上下文切换开销飙升的同时也会占用更多内存。

1. 修复消费逻辑实现并行处理 #

移除消费循环中对spawn_blocking的await操作，配合信号量控制同时运行的计算任务数量，避免线程池过载：

use tokio::sync::Semaphore;
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel::<WriteRequest>(1024); // 可根据实际场景适当调大通道容量
    // CPU密集型任务建议最大并行数设为CPU核数的1~2倍，避免上下文切换开销过高
    let max_concurrent = num_cpus::get() * 2;
    let semaphore = Arc::new(Semaphore::new(max_concurrent));

    tokio::spawn(async move {
        while let Some(payload) = rx.recv().await {
            let permit = semaphore.acquire().await.unwrap();
            tokio::task::spawn_blocking(move || {
                let _permit_guard = permit; // 任务完成后自动释放信号量
                heavy_computation(payload)
            });
            // 不要await spawn_blocking的结果，避免阻塞消费流程
        }
    });

    // 其余初始化逻辑见下文
    // ...
}

2. 增加限流降级策略 #

在发送请求到通道时增加超时逻辑，通道满时直接返回503错误，避免请求在服务端堆积：

use axum::{http::StatusCode, response::IntoResponse};
use std::time::Duration;

let app = Router::new()
    .route("/write", post(move |req: Bytes| async move {
         let data: WriteRequest = Message::decode(req);
         // 超时时间可根据实际场景调整，避免请求长时间阻塞占用内存
         match tokio::time::timeout(Duration::from_millis(5), tx.send(data)).await {
             Ok(Ok(_)) => "ok".into_response(),
             _ => (StatusCode::SERVICE_UNAVAILABLE, "server busy, please retry later").into_response(),
         }
     }));

3. 内存细节优化 #

• 避免WriteRequest的不必要拷贝，尽量复用axum传入的Bytes零拷贝特性
• 若heavy_computation仅需要部分字段，只传递必要字段到闭包中，减少无效内存占用
• 对于大体积的请求payload，可以考虑流式解码，避免一次性加载全量数据到内存

4. 架构层面水平扩展 #

单节点CPU密集型任务的处理上限通常为几万到几十万QPS，远低于2000万的要求，需要在服务前面增加负载均衡层，将流量分发到多台服务器并行处理，才能支撑目标吞吐量。

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