Rocket.rs异步流仅处理6个并发请求，与24 worker配置不符问题排查

这不是Rocket的固有限制，大概率是你的实现细节或默认配置导致的并发瓶颈，以下是具体分析和解决思路：

常见原因 #

1. 客户端HTTP/1.1连接限制 #

如果用浏览器测试，HTTP/1.1协议默认限制单个客户端最多同时建立6个TCP连接，超出的请求会排队等待连接释放。这是浏览器的通用限制，和Rocket无关。

2. 路由逻辑阻塞Worker线程 #

Rocket的Worker线程是处理请求的核心，如果你的路由里包含同步阻塞操作（比如std::thread::sleep、同步文件IO、长时间CPU计算），哪怕用了async语法，也会直接占用Worker线程，导致无法处理新请求。24个Worker如果被6个阻塞请求占满，剩下的请求自然会挂起。

3. 错误使用同步锁 #

如果你的通知逻辑依赖全局同步锁（比如std::sync::Mutex），在异步代码中持有这类锁会阻塞Worker线程。若锁的竞争逻辑导致同一时间仅能处理6个请求，就会出现你看到的现象。

4. Tokio Runtime配置不足 #

Rocket基于Tokio运行，每个Worker对应一个Tokio Runtime实例。默认的Runtime线程数或IO池大小可能无法支撑高并发请求，导致任务排队。

解决方法 #

1. 验证客户端限制 #

用多客户端工具（如wrk、ab）或不同浏览器发起请求，排除客户端连接限制：

# 用wrk发起30个并发请求测试
wrk -c 30 -t 10 http://localhost:8000/notify

2. 替换阻塞操作为异步实现 #

将所有同步阻塞代码替换为Tokio的异步版本，或用spawn_blocking将阻塞逻辑移到专用线程池：

use std::time::Duration;
use rocket::response::stream::EventStream;

#[get("/notify")]
async fn notify() -> EventStream![] {
    // 错误：同步睡眠阻塞Worker
    // std::thread::sleep(Duration::from_secs(10));
    
    // 正确：异步睡眠，不占用Worker线程
    tokio::time::sleep(Duration::from_secs(10)).await;

    // 若必须调用阻塞代码，用spawn_blocking隔离
    let data = tokio::task::spawn_blocking(|| {
        std::fs::read_to_string("notification_data.txt").unwrap()
    }).await.unwrap();

    // 你的流处理逻辑示例
    EventStream! {
        yield "notification: ".to_string() + &data;
    }
}

3. 改用异步锁替代同步锁 #

将std::sync::Mutex替换为Tokio的异步锁tokio::sync::Mutex，避免阻塞Worker线程：

use tokio::sync::Mutex;

struct NotificationService;
impl NotificationService {
    fn new() -> Self { NotificationService }
    // 示例通知方法
    async fn send(&mut self) -> String { "sent".to_string() }
}

static NOTIFIER: Mutex<NotificationService> = Mutex::const_new(NotificationService::new());

#[get("/notify")]
async fn notify() -> EventStream![] {
    let mut service = NOTIFIER.lock().await;
    let result = service.send().await;
    
    EventStream! {
        yield result;
    }
}

4. 调整Rocket与Tokio配置 #

通过代码或Rocket.toml调整Worker数量、Tokio线程池大小及HTTP/2并发流限制：

use rocket::Config;

#[launch]
fn rocket() -> _ {
    let config = Config::figment()
        .merge(("workers", 24))
        .merge(("tokio_threads", 64)) // 每个Worker的Tokio核心线程数
        .merge(("max_concurrent_streams", 100)); // HTTP/2并发流上限

    rocket::custom(config)
        .mount("/", routes![notify])
}

总结 #

只要正确实现异步逻辑、调整合理配置，Rocket可以支撑大量并发请求，不存在硬编码的6个并发限制。优先排查客户端连接限制和阻塞代码，这是最常见的问题根源。

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