Actix-Web中async-sqlite多行列查询触发跨线程安全报错求助

问题场景 #

使用Actix-web编写全异步API，基于rusqlite的async-sqlite crate操作SQLite数据库。单条数据查询/写入可正常运行，但执行多行列查询使用prepare语句时，触发4个E0277编译错误，提示*mut sqlite3_stmt、RefCell等类型无法安全跨线程共享。

用户代码示例：

use actix_web::{web, App, HttpResponse, HttpServer, post};
use async_sqlite::{JournalMode, PoolBuilder};


async fn database_lookup(sql_pool: &async_sqlite::Pool) -> Result<String, String> {

    let _result = match sql_pool.conn(move|conn| {
        let mut stmt = conn.prepare("SELECT attribute WHERE resource = :resource").unwrap();
        let rows = stmt.query_map(&[":resource", "https://test.com"], |row| row.get(0));
        let mut attributes = Vec::new();
        for attr in rows {
            attributes.push(attr);
        }
        Ok(attributes)
    }).await {
        Ok(attrib) => Ok(attrib),
        Err(_e) => return Err("There was an error".to_string())
    };

    return Ok("abc".to_string());
}


#[post("/endpoint")]
async fn endpoint() -> HttpResponse {
    
    let sql_pool = match PoolBuilder::new()
            .path("db.sqlite3")
            .journal_mode(JournalMode::Wal)
            .open()
            .await {
        Ok(con) => con,
        Err(_e) => panic!("Failed to open database")
    };

    let _result = match database_lookup(&sql_pool).await {
        Ok(c) => c,
        Err(_e) => return HttpResponse::BadRequest().body(format!("bad things happen"))
    };

    return HttpResponse::Ok().body(format!("Good stuff"));
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {

    HttpServer::new(|| {
        App::new().service(
            web::scope("/base")
                .service(endpoint),
        )
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

解决方案 #

1. 核心问题根源 #

async-sqlite的conn方法会将闭包提交到后台线程执行，要求闭包返回的类型必须实现Send（可安全跨线程传递）。但rusqlite的Stmt、Rows等类型内部包含RefCell、裸指针等非线程安全结构，不满足Send约束，直接返回这些类型会触发编译错误。

2. 具体修正步骤 #

（1）在闭包内完成所有数据提取和转换

不要返回Stmt或Rows，而是在闭包内部将查询结果转换成普通的Send类型（如Vec<String>），确保返回值满足线程安全要求。

修正后的database_lookup函数：

async fn database_lookup(sql_pool: &async_sqlite::Pool) -> Result<Vec<String>, String> {
    sql_pool.conn(|conn| {
        // 替换unwrap为错误处理，避免panic
        let mut stmt = conn.prepare("SELECT attribute WHERE resource = :resource")
            .map_err(|e| format!("Prepare statement failed: {}", e))?;
        
        // 直接在闭包内将查询结果收集为Vec<String>
        let attributes: Result<Vec<String>, _> = stmt.query_map(
            &[":resource", "https://test.com"], 
            |row| row.get(0)
        ).collect();
        
        attributes.map_err(|e| format!("Query failed: {}", e))
    }).await
}

（2）修正数据库池的生命周期

当前代码在每个请求的endpoint函数中创建新数据库池，会导致严重性能问题和连接泄漏。应在main函数中创建一次池，通过Actix-web的应用数据（App Data）注入到请求处理函数中。

修正后的main和endpoint：

#[post("/endpoint")]
async fn endpoint(sql_pool: web::Data<async_sqlite::Pool>) -> HttpResponse {
    match database_lookup(&sql_pool).await {
        Ok(_c) => HttpResponse::Ok().body("Good stuff"),
        Err(e) => HttpResponse::BadRequest().body(format!("bad things happen: {}", e))
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    // 在main中初始化数据库池，全局复用
    let sql_pool = PoolBuilder::new()
        .path("db.sqlite3")
        .journal_mode(JournalMode::Wal)
        .open()
        .await
        .expect("Failed to open database");

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(sql_pool.clone())) // 注入应用数据
            .service(
                web::scope("/base")
                    .service(endpoint),
            )
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

3. 关键说明 #

• 所有数据库操作的中间结果（如Stmt、Rows）必须限制在conn闭包内部，不能跨线程传递。
• 数据库池应全局初始化并复用，避免每个请求创建新池。
• 替换unwrap为显式错误处理，提升代码健壮性。

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