如何在Lambda处理器中处理多个异步函数？编译报错求助

问题场景 #

在AWS Lambda的Rust函数中，使用try_join!并行执行create_db_eid_cache和create_table_eid_cache两个异步函数时，编译报错：

future cannot be sent between threads safely
the trait std::marker::Send is not implemented for (dyn StdError + 'static)

尝试给function_handler的返回类型添加Send约束后问题仍未解决。

错误原因 #

Lambda运行时要求handler返回的future必须实现Send trait（因为任务会在多线程环境中调度）。使用try_join!组合的future，其Send特性依赖于所有子future及其错误类型都实现Send。

默认的Box<dyn std::error::Error>没有Send约束，导致整个Result类型的future不满足Lambda运行时的要求。

解决方案 #

方案1：显式指定带Send+Sync约束的错误类型 #

定义包含Send + Sync约束的错误类型，确保所有异步函数的错误返回值都遵循该类型：

use lambda_http::{run, service_fn, Request, Response, Body};
use std::error::Error;
use tokio::try_join;

// 定义带Send和Sync约束的动态错误类型
type DynError = Box<dyn Error + Send + Sync + 'static>;

async fn function_handler(_event: Request) -> Result<Response<Body>, DynError> {
    let db_eid_cache = create_db_eid_cache();
    let table_eid_cache = create_table_eid_cache();

    // 并行执行两个异步任务，处理联合错误
    try_join!(db_eid_cache, table_eid_cache)?;

    Ok(Response::builder()
        .status(200)
        .header("content-type", "application/json")
        .body("ok".into())
        .map_err(Box::new)?)
}

// 确保异步函数的错误返回类型与DynError一致
async fn create_db_eid_cache() -> Result<(), DynError> {
    // 替换为你的实际业务逻辑
    Ok(())
}

async fn create_table_eid_cache() -> Result<(), DynError> {
    // 替换为你的实际业务逻辑
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), DynError> {
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        .with_target(false)
        .without_time()
        .init();

    run(service_fn(function_handler)).await
}

方案2：使用anyhow简化错误处理 #

anyhow::Error默认实现了Send + Sync，可以直接作为错误类型使用，无需手动添加约束：

use lambda_http::{run, service_fn, Request, Response, Body};
use anyhow::{Result, Context};
use tokio::try_join;

async fn function_handler(_event: Request) -> Result<Response<Body>> {
    let db_eid_cache = create_db_eid_cache();
    let table_eid_cache = create_table_eid_cache();

    try_join!(db_eid_cache, table_eid_cache)?;

    Ok(Response::builder()
        .status(200)
        .header("content-type", "application/json")
        .body("ok".into())
        .context("Failed to build HTTP response")?)
}

async fn create_db_eid_cache() -> Result<()> {
    // 替换为你的实际业务逻辑，用anyhow的?自动转换错误
    Ok(())
}

async fn create_table_eid_cache() -> Result<()> {
    // 替换为你的实际业务逻辑
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        .with_target(false)
        .without_time()
        .init();

    run(service_fn(function_handler)).await
}

关键说明 #

• Lambda的run函数对传入的service future有Send约束，这是多线程调度的必要条件。
• 无论是手动定义带约束的错误类型，还是使用anyhow，核心都是确保错误类型实现Send，从而让整个future满足Lambda的要求。

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