从Rocket 0.4迁移至0.5：如何同步返回sync_db_pools查询结果

我正在把一个应用从 Rocket 0.4 迁移到 0.5，之前用的是 rocket_contrib 的 diesel_postgres_pool 特性，觉得 rocket_sync_db_pools 是最佳迁移方案（也接受其他方案）。

我尝试修改了代码：

-        let quality = subjects::table
-            .find(&sub)
-            .select(subjects::quality)
-            .first::<i16>(&conn.0)
-            .ok();
+        let quality = conn.run( move |conn| {
+            subjects::table
+                .find(&sub)
+                .select(subjects::quality)
+                .first::<i16>(conn)
+                .ok();
+        });

但现在得到的不是实际结果，而是一个 Promise。文档说要用 .await，但这会让调用函数也变成异步的，我做不到——因为这段代码还要用在一个很难实现异步的 trait 里。

作为 Rust 新手，我有两个疑问：

1. 为什么用 await 等待结果时，函数必须是异步的？
2. 为什么同步数据库池会返回异步 Promise/需要异步函数？

这个思路对吗？有没有不用重构整个应用的替代方案？

编辑1 #

导入 async_trait 后有了进展：

#[async_trait]
impl Statistic for Subject {
    async fn compute(conn: &DbConn, sub: String) -> Result<Self, Error> {


[...]

        let quality = conn.run( move |conn| {
            return subjects::table
                .find(&sub)
                .select(subjects::quality)
                .first::<i16>(conn)
                .ok();
        }).await;

但调用 compute 方法时出错：

identifiers
            .map(|identifier| {
                Self::compute(&conn, identifier.clone()).map(|statistic| (identifier, statistic))
            })
            .collect()

错误信息：

23  |                 Self::compute(&conn, identifier.clone()).map(|statistic| (identifier, statistic))
    |                                                          ^^^ `Pin<Box<dyn std::future::Future<Output = Result<Self, error::Error>> + std::marker::Send>>` is not an iterator

编辑2 #

发现可能不需要 async_trait 包，Rocket 自己提供了解决方案。

编辑3 #

尝试了下面的代码：

identifiers
            .map(async move |identifier| {
                join_all(Self::compute(&conn, identifier.clone())).await.map(|statistic| (identifier, statistic))
            })
            .collect()

又出现新错误：

error[E0658]: async closures are unstable
  --> src/aggregator.rs:23:18
   |
23 |             .map(async move |identifier| {
   |                  ^^^^^
   |
   = note: see issue #62290 <https://github.com/rust-lang/rust/issues/62290> for more information
   = help: to use an async block, remove the `||`: `async {`

编辑4 #

尝试了某个方案后：

let quality = Handle::current().block_on(async move {
            conn.run(move |conn| {
                subjects::table
                    .find(&sub)
                    .select(subjects::quality)
                    .first::<i16>(conn)
                    .ok();
            })
        });
        Ok(Subject {
            sub,
            quality,
            count,
            opinion_count,
            positive_count,
            confirmed_count,
        })

出现类型不匹配错误：

error[E0308]: mismatched types
  --> src/aggregator.rs:84:13
   |
74 |             conn.run(move |conn| {
   |                      ----------- the found closure
...
84 |             quality,
   |             ^^^^^^^ expected enum `std::option::Option`, found opaque type

问题解答 #

先回答你的两个疑问 #

1. 为什么用 await 必须在异步函数里？
   Rust 里的 await 是用来暂停当前执行、等待 Future 完成的语法。异步函数的本质是返回一个 Future 类型，而 await 只能在 Future 的执行上下文中使用——普通同步函数没有这个上下文，没法处理暂停和恢复逻辑，所以必须把函数标记为 async 才能用 await。

2. 为什么同步数据库池要返回异步 Future？
   虽然 rocket_sync_db_pools 用的是同步的 Diesel 驱动，但 Rocket 0.5 是全异步架构。从池里获取连接这个操作本身是异步的（比如池里没有空闲连接时要等待），所以 conn.run() 必须返回 Future，让 Rocket 的异步 runtime 来调度这个等待过程。即使你的数据库操作是同步的，获取连接的步骤还是要适配异步架构。

解决方案：分两种场景处理 #

场景1：不想把 trait 改成异步（尽量少重构）

如果你的 trait 实在没法改成异步，可以用 Rocket 提供的 Local 阻塞运行时，直接在同步代码里阻塞等待异步操作完成：

use rocket::tokio::runtime::Handle;

// 在你的同步函数里
let quality = Handle::current().block_on(async {
    conn.run(|conn| {
        subjects::table
            .find(&sub)
            .select(subjects::quality)
            .first::<i16>(conn)
            .ok()
    }).await
});

注意：这里要把 conn.run() 的结果 await 后再返回，之前的错误是因为闭包里没明确返回表达式，且 block_on 里要等待 run() 的 Future 完成。

场景2：接受把 trait 改成异步（更符合 Rocket 0.5 架构）

如果可以调整 trait，用 Rocket 推荐的异步 trait 方式（不需要额外的 async_trait 包）：

// 先定义异步 trait
#[rocket::async_trait]
pub trait Statistic {
    async fn compute(conn: &DbConn, sub: String) -> Result<Self, Error>
    where
        Self: Sized;
}

// 实现 trait
#[rocket::async_trait]
impl Statistic for Subject {
    async fn compute(conn: &DbConn, sub: String) -> Result<Self, Error> {
        let quality = conn.run(|conn| {
            subjects::table
                .find(&sub)
                .select(subjects::quality)
                .first::<i16>(conn)
                .ok()
        }).await;

        // 其他逻辑...
        Ok(Subject {
            sub,
            quality,
            // 其他字段
        })
    }
}

然后处理批量调用的问题：之前的错误是因为 compute 返回的是 Future，不能直接用 .map()，要把所有 Future 收集起来一起等待：

use rocket::tokio::join_all;

// 把所有 compute 调用的 Future 收集成 Vec
let futures: Vec<_> = identifiers
    .into_iter()
    .map(|identifier| async move {
        let statistic = Subject::compute(&conn, identifier.clone()).await?;
        Ok((identifier, statistic))
    })
    .collect();

// 等待所有 Future 完成
let results: Result<Vec<_>, Error> = join_all(futures).await.into_iter().collect();

这样就能正确得到批量处理的结果，且符合 Rocket 0.5 的异步调度逻辑。

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