4、异步数据库访问：tokio异步运行时、sqlx异步驱动、异步CRUD实战

好，咱们进入第四章。这一章可以说是整个课程的分水岭——从同步走向异步。说白了，Rust 在数据库中间件领域的核心竞争力，很大程度上就体现在它的异步能力上。你想想看，一个数据库连接池要同时处理成千上万个请求，如果每个请求都阻塞等待，那性能就全毁了。

我个人习惯把异步编程比作「高效的时间管理大师」。它不会傻等，而是利用等待的时间去干别的活。今天我们就用 tokio 和 sqlx 这两个库，把异步数据库访问这件事彻底讲透。

4.1 为什么需要异步？

先问一个问题：你在写数据库操作时，有没有遇到过这种情况？

• 一个查询耗时 200ms，期间 CPU 完全空闲
• 并发一上来，线程数暴涨，内存吃紧
• 上下文切换频繁，性能反而下降

嗯，这就是同步阻塞的典型问题。每个线程在等待数据库响应时，都在「干等」。异步的解决方案很简单：把等待的时间让出来，让其他任务先跑。

我在项目中遇到过这样一个场景：一个微服务需要同时查询 10 个分片数据库。如果用同步方式，总耗时就是 10 个查询的累加。换成异步后，10 个查询并发执行，总耗时只等于最慢的那个查询。这个优化效果，立竿见影。

4.2 tokio：Rust 异步运行时的事实标准

tokio 是什么？简单说，它就是 Rust 异步生态的「操作系统」。它负责调度任务、管理 I/O、处理时间。没有它，我们的异步代码就跑不起来。

我刚开始接触 tokio 时，觉得它挺复杂的。后来发现，你只需要掌握三个核心概念：

• Runtime（运行时）：异步任务的执行环境
• Task（任务）：被调度的最小执行单元
• Future（未来）：表示一个尚未完成的计算

来看一个最简单的 tokio 程序：

use tokio;

#[tokio::main]
async fn main() {
    println!("Hello from tokio!");
}

这个 #[tokio::main] 宏，帮我们自动创建了一个 Runtime，然后在这个 Runtime 里执行 main 函数。你想想看，如果没有这个宏，你得手动创建 Runtime、调用 block_on，多麻烦。

4.3 sqlx：异步数据库驱动的正确姿势

sqlx 是我个人最喜欢的 Rust 数据库驱动库。为什么？因为它同时支持同步和异步，而且编译时检查 SQL 语句的正确性。这一点，很多 ORM 都做不到。

sqlx 支持多种数据库：PostgreSQL、MySQL、SQLite、MSSQL。今天我们以 PostgreSQL 为例，因为它在生产环境用得最多。

先添加依赖：

[dependencies]
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.7", features = ["runtime-tokio", "postgres"] }
anyhow = "1.0"  // 错误处理，省事

连接数据库的方式有两种：

• 单连接：PgConnection::connect()，适合测试或临时任务
• 连接池：PgPool::connect()，生产环境首选

我建议你永远用连接池。为什么呢？因为创建数据库连接是个昂贵的操作——握手、认证、SSL 协商，一套下来几十毫秒。连接池提前创建好一批连接，随用随取，用完归还，效率高得多。

use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    let pool = PgPoolOptions::new()
        .max_connections(10)  // 最大连接数
        .connect("postgres://user:password@localhost/mydb")
        .await?;

    println!("连接池创建成功！");
    Ok(())
}

4.4 异步 CRUD 实战

好，理论讲完了，咱们直接上代码。假设我们有一个用户表：

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

对应的 Rust 结构体：

use sqlx::FromRow;
use chrono::{DateTime, Utc};

#[derive(Debug, FromRow)]
struct User {
    id: i32,
    name: String,
    email: String,
    created_at: DateTime<Utc>,
}

这个 FromRow 派生宏是 sqlx 的魔法——它自动把数据库行映射成 Rust 结构体。省去了手动写 row.get("name") 的繁琐。

4.4.1 创建（Create）

async fn create_user(pool: &sqlx::PgPool, name: &str, email: &str) -> Result<User, sqlx::Error> {
    let user = sqlx::query_as::<_, User>(
        "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *"
    )
    .bind(name)
    .bind(email)
    .fetch_one(pool)
    .await?;

    Ok(user)
}

这里有个细节：我们用 RETURNING * 把插入后的完整行返回回来。这样就能拿到数据库自动生成的 id 和 created_at。我在项目中经常看到有人插入后再查一次，其实完全没必要。

4.4.2 查询（Read）

async fn get_user_by_id(pool: &sqlx::PgPool, user_id: i32) -> Result<Option<User>, sqlx::Error> {
    let user = sqlx::query_as::<_, User>(
        "SELECT * FROM users WHERE id = $1"
    )
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    Ok(user)
}

async fn list_users(pool: &sqlx::PgPool) -> Result<Vec<User>, sqlx::Error> {
    let users = sqlx::query_as::<_, User>(
        "SELECT * FROM users ORDER BY created_at DESC"
    )
    .fetch_all(pool)
    .await?;

    Ok(users)
}

注意 fetch_optional 和 fetch_all 的区别。前者返回 Option，适合查单个记录；后者返回 Vec，适合查列表。如果你用 fetch_one 查不存在的记录，它会直接报错。

4.4.3 更新（Update）

async fn update_user_email(pool: &sqlx::PgPool, user_id: i32, new_email: &str) -> Result<u64, sqlx::Error> {
    let rows_affected = sqlx::query(
        "UPDATE users SET email = $1 WHERE id = $2"
    )
    .bind(new_email)
    .bind(user_id)
    .execute(pool)
    .await?
    .rows_affected();

    Ok(rows_affected)
}

这里返回的是 u64，表示影响的行数。如果返回 0，说明没有匹配的记录。这个返回值可以用来做乐观锁——比如更新前先检查版本号。

4.4.4 删除（Delete）

async fn delete_user(pool: &sqlx::PgPool, user_id: i32) -> Result<bool, sqlx::Error> {
    let result = sqlx::query("DELETE FROM users WHERE id = $1")
        .bind(user_id)
        .execute(pool)
        .await?;

    Ok(result.rows_affected() > 0)
}

删除操作返回布尔值，表示是否真的删除了记录。这个设计比返回 u64 更语义化——调用方只需要知道「删没删掉」，不需要知道具体删了几行。

4.5 事务处理

实际项目中，很少有单条 SQL 就能搞定的业务。比如转账：扣钱和加钱必须在一个事务里，要么都成功，要么都失败。

sqlx 的事务处理非常优雅：

async fn transfer_money(
    pool: &sqlx::PgPool,
    from_id: i32,
    to_id: i32,
    amount: f64,
) -> Result<(), anyhow::Error> {
    let mut tx = pool.begin().await?;

    // 扣钱
    sqlx::query("UPDATE accounts SET balance = balance - $1 WHERE id = $2")
        .bind(amount)
        .bind(from_id)
        .execute(&mut *tx)
        .await?;

    // 加钱
    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(amount)
        .bind(to_id)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(())
}

注意这里 execute(&mut *tx) 的写法。tx 是事务对象，它实现了 Executor trait，所以可以像连接池一样使用。但需要解引用再取可变引用，这个语法有点绕，记住就行。

4.6 性能调优与最佳实践

最后，分享几个我在生产环境中总结的经验：

实践	说明
连接池预热	应用启动时，先执行一个简单的 SELECT 1，让连接池建立好初始连接
超时控制	用 tokio::time::timeout 包裹数据库操作，防止慢查询拖垮整个系统
批量操作	用 sqlx::QueryBuilder 构建批量 INSERT，避免逐条插入
连接泄漏	确保每次操作后连接都归还到池中。用 drop(tx) 或 tx.rollback() 处理异常情况

关于超时控制，我举个例子：

use tokio::time::{timeout, Duration};

async fn query_with_timeout(pool: &sqlx::PgPool) -> Result<Option<User>, anyhow::Error> {
    let result = timeout(Duration::from_secs(5), async {
        sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
            .bind(1)
            .fetch_optional(pool)
            .await
    }).await;

    match result {
        Ok(Ok(user)) => Ok(user),
        Ok(Err(e)) => Err(e.into()),
        Err(_) => Err(anyhow::anyhow!("查询超时")),
    }
}

这个模式我几乎在每个项目里都用。数据库偶尔会抽风——死锁、锁等待、慢查询，超时机制就是最后的保险丝。

4.7 本章小结

异步数据库访问，说白了就是「不等待」。tokio 提供运行时，sqlx 提供驱动，两者配合，让 Rust 在数据库中间件领域如鱼得水。

我个人觉得，掌握异步的关键不在于记住 API，而在于理解「异步思维」——把 I/O 等待时间让出来，让系统做更多的事。你想想看，一个连接池 + 异步 CRUD + 事务处理，这套组合拳打出去，大部分数据库中间件的需求都能搞定。

嗯，这一章的内容就到这里。代码都在上面了，建议你打开编辑器，跟着敲一遍。遇到问题很正常，调试的过程就是学习的过程。

---

公众号:蓝海资料掘金营，微信deep3321