如何将Actix Web Payload传递至函数并解决线程安全报错

我希望从路由处理程序中抽离部分Payload处理逻辑，但遇到类型安全问题：尝试将Payload标记为Send安全后，编译器报错无法跨线程发送Payload。请问如何保证该Payload仅由相关工作线程处理？

现有代码示例 #

路由定义 #

#[post("/upload")]
pub async fn upload(mut body: Payload) -> Result<HttpResponse> {
    Handler {}.handle_payload(body).await?;
    Ok(HttpResponse::Ok().finish())
}

处理Trait及实现 #

#[async_trait::async_trait]
pub trait Handle {
    async fn handle_payload(&self, s: impl Stream + Send) -> Result<()>;
}

pub struct Handler {}

#[async_trait::async_trait]
impl Handle for Handler {
    async fn handle_payload(&self, s: impl Stream + Send) -> Result<()> {
        Ok(())
    }
}

编译错误信息 #

|     Handler {}.handle_payload(body).await?;
    |                  -------------- ^^^^ `Rc<RefCell<actix_http::h1::payload::Inner>>` cannot be sent between threads safely
    |                  |
    |                  required by a bound introduced by this call
    |
    = help: within `actix_web::web::Payload`, the trait `std::marker::Send` is not implemented for `Rc<RefCell<actix_http::h1::payload::Inner>>`
    = note: required because it appears within the type `actix_http::h1::payload::Payload`
    = note: required because it appears within the type `actix_web::dev::Payload`
    = note: required because it appears within the type `actix_web::web::Payload`

以及：

|     Handler {}.handle_payload(body).await?;
    |                  -------------- ^^^^ `(dyn futures_util::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)` cannot be sent between threads safely
    |                  |
    |                  required by a bound introduced by this call
    |
    = help: the trait `std::marker::Send` is not implemented for `(dyn futures_util::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)`
    = note: required for `Unique<(dyn futures_util::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>` to implement `std::marker::Send`
    = note: required because it appears within the type `Box<(dyn futures_util::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>`
    = note: required because it appears within the type `Pin<Box<(dyn futures_util::Stream<Item = Result<actix_web::web::Bytes, PayloadError>> + 'static)>>`
    = note: required because it appears within the type `actix_web::dev::Payload`
    = note: required because it appears within the type `actix_web::web::Payload`

解决方案 #

问题核心在于：Actix Web的Payload内部使用了非线程安全的Rc<RefCell>，因此本身不实现Send trait；而async_trait宏默认会强制异步方法返回的future是Send的，这就导致了冲突。

要解决这个问题，只需让trait的异步方法允许非Send的future，同时去掉对Stream的Send约束，让Payload在当前工作线程内处理即可：

修改后的Trait及实现 #

#[async_trait::async_trait(?Send)] // 添加?Send参数，允许非Send的future
pub trait Handle {
    async fn handle_payload(&self, s: impl Stream) -> Result<()>; // 移除Send约束
}

pub struct Handler {}

#[async_trait::async_trait(?Send)]
impl Handle for Handler {
    async fn handle_payload(&self, s: impl Stream) -> Result<()> {
        // 示例：流式处理Payload内容
        while let Some(result) = s.next().await {
            let bytes = result?;
            // 在这里添加你的业务处理逻辑
        }
        Ok(())
    }
}

路由代码保持不变 #

#[post("/upload")]
pub async fn upload(mut body: Payload) -> Result<HttpResponse> {
    Handler {}.handle_payload(body).await?;
    Ok(HttpResponse::Ok().finish())
}

补充说明 #

如果确实需要跨线程处理Payload（不推荐大文件场景），可以先将整个Payload读取到内存中，转换为Bytes或Vec<u8>这类实现了Send的类型，再传递给其他线程。示例代码如下：

#[post("/upload")]
pub async fn upload(mut body: Payload) -> Result<HttpResponse> {
    // 读取整个Payload到内存
    let bytes = body.try_concat().await?;
    // 跨线程处理（比如用spawn）
    tokio::spawn(async move {
        // 处理bytes的逻辑
    });
    Ok(HttpResponse::Ok().finish())
}

但这种方式会将整个请求体加载到内存，对于大文件上传可能导致内存占用过高，因此优先推荐第一种在当前线程流式处理的方案。

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