Rust 2026 经验谈 - 数据库与 ORM 选型

1
use sqlx::postgres::PgPool;
2

3
async fn create_user(pool: &PgPool, name: &str, email: &str) -> Result<User, sqlx::Error> {
4
    // 编译时检查：如果 users 表不存在或列名错误，cargo build 直接报错
5
    let user = sqlx::query_as!(
6
        User,
7
        r#"INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id, name, email, created_at"#,
8
        name, email
9
    )
10
    .fetch_one(pool)
11
    .await?;
12
    Ok(user)
13
}
14

15
async fn list_users(pool: &PgPool) -> Result<Vec<User>, sqlx::Error> {
16
    let users = sqlx::query_as!(
17
        User,
18
        r#"SELECT id, name, email, created_at FROM users ORDER BY id LIMIT 100"#
19
    )
20
    .fetch_all(pool)
21
    .await?;
22
    Ok(users)
23
}

1
// query!：返回匿名列，字段名与 SQL 列名一致
2
let row = sqlx::query!(
3
    r#"SELECT id, name FROM users WHERE id = $1"#,
4
    user_id
5
)
6
.fetch_one(pool)
7
.await?;
8
// row.id: i32, row.name: String
9

10
// query_as!：映射到自定义结构体
11
#[derive(sqlx::FromRow)]
12
struct UserRow {
13
    id: i32,
14
    name: String,
15
}
16
let user = sqlx::query_as!(
17
    UserRow,
18
    r#"SELECT id, name FROM users WHERE id = $1"#,
19
    user_id
20
)
21
.fetch_one(pool)
22
.await?;
23

24
// query_scalar!：返回单个值
25
let count: i64 = sqlx::query_scalar!(
26
    r#"SELECT COUNT(*) FROM users"#
27
)
28
.fetch_one(pool)
29
.await?;

1
use sea_orm::{Database, EntityTrait, QueryFilter, Set, ActiveModelTrait};
2
use entity::user;
3

4
let db = Database::connect("postgres://user:pass@localhost/db").await?;
5

6
// 查询
7
let users = user::Entity::find().all(&db).await?;
8

9
// 条件查询
10
let alice = user::Entity::find()
11
    .filter(user::Column::Name.eq("Alice"))
12
    .one(&db)
13
    .await?;
14

15
// 插入
16
let new_user = user::ActiveModel {
17
    name: Set("Bob".into()),
18
    email: Set("bob@example.com".into()),
19
    ..Default::default()
20
};
21
let result = new_user.insert(&db).await?;
22

23
// 更新
24
let mut model: user::ActiveModel = alice.unwrap().into();
25
model.name = Set("Alice Updated".into());
26
model.update(&db).await?;
27

28
// 删除
29
user::Entity::delete_by_id(1).exec(&db).await?;

1
use diesel::prelude::*;
2
use diesel::result::QueryResult;
3

4
// schema.rs 由 diesel print-schema 生成
5
table! {
6
    users (id) {
7
        id -> Int4,
8
        name -> Varchar,
9
        email -> Varchar,
10
        created_at -> Timestamp,
11
    }
12
}
13

14
#[derive(Queryable, Selectable)]
15
#[diesel(table_name = users)]
16
struct User {
17
    id: i32,
18
    name: String,
19
    email: String,
20
    created_at: chrono::NaiveDateTime,
21
}
22

23
#[derive(Insertable)]
24
#[diesel(table_name = users)]
25
struct NewUser<'a> {
26
    name: &'a str,
27
    email: &'a str,
28
}
29

30
fn create_user(conn: &mut PgConnection, name: &str, email: &str) -> QueryResult<User> {
31
    let new_user = NewUser { name, email };
32
    diesel::insert_into(users::table)
33
        .values(&new_user)
34
        .returning(User::as_returning())
35
        .get_result(conn)
36
}
37

38
fn list_users(conn: &mut PgConnection) -> QueryResult<Vec<User>> {
39
    users::table
40
        .order(users::id.asc())
41
        .limit(100)
42
        .load(conn)
43
}

1
# SQLx 编译时检查需要数据库——CI 中如何处理？
2

3
# 方案 1：离线模式（推荐）
4
# 先在本地生成离线查询数据
5
# DATABASE_URL=postgres://... cargo sqlx prepare --check
6

7
# 提交 .sqlx 目录到 git
8
# CI 中不需要数据库
9

10
# .github/workflows/ci.yml
11
- name: Check with offline queries
12
  run: cargo check
13
  env:
14
    SQLX_OFFLINE: true
15

16
# 方案 2：CI 中启动数据库
17
- name: Start Postgres
18
  run: |
19
    docker run -d -p 5432:5432 \
20
      -e POSTGRES_PASSWORD=password \
21
      postgres:16
22
- name: Wait for DB
23
  run: sleep 5
24
- name: Check
25
  run: cargo check
26
  env:
27
    DATABASE_URL: postgres://postgres:password@localhost/postgres

1
use sqlx::postgres::{PgPool, PgPoolOptions};
2
use std::time::Duration;
3

4
let pool = PgPoolOptions::new()
5
    .max_connections(20)
6
    .min_connections(5)
7
    .acquire_timeout(Duration::from_secs(5))
8
    .idle_timeout(Duration::from_secs(600))
9
    .max_lifetime(Duration::from_secs(1800))
10
    .connect("postgres://user:pass@localhost/db")
11
    .await?;
12

13
// 使用
14
let row = sqlx::query!("SELECT 1")
15
    .fetch_one(&pool)
16
    .await?;

1
use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
2
use tokio_postgres::NoTls;
3

4
let mgr_config = ManagerConfig {
5
    recycling_method: RecyclingMethod::Verified,
6
    ..Default::default()
7
};
8

9
let mgr = Manager::from_config(
10
    "postgres://user:pass@localhost/db".parse()?,
11
    NoTls,
12
    mgr_config,
13
);
14

15
let pool = Pool::builder(mgr)
16
    .max_size(20)
17
    .build()?;
18

19
let client = pool.get().await?;
20
let rows = client.query("SELECT 1", &[]).await?;

1
// ❌ 获取连接后 panic 或忘记归还
2
async fn bad(pool: &PgPool) {
3
    let conn = pool.acquire().await.unwrap();
4
    // 如果这里 panic，连接可能不归还
5
    panic!("oops");
6
}
7

8
// ✓ 使用作用域守卫（SQLx 自动管理）
9
async fn good(pool: &PgPool) {
10
    // pool.acquire() 返回 PoolConnection
11
    // Drop 时自动归还连接
12
    let mut conn = pool.acquire().await?;
13
    sqlx::query!("SELECT 1").execute(&mut *conn).await?;
14
    // conn Drop → 归还连接
15
}

1
use sqlx::migrate::Migrator;
2

3
// 编译时嵌入 migration 文件
4
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
5

6
async fn run_migrations(pool: &PgPool) -> Result<(), sqlx::migrate::MigrateError> {
7
    MIGRATOR.run(pool).await
8
}
9

10
#[tokio::main]
11
async fn main() -> Result<(), Box<dyn std::error::Error>> {
12
    let pool = PgPool::connect(&env::var("DATABASE_URL")?).await?;
13
    run_migrations(&pool).await?;
14
    // 启动服务...
15
    Ok(())
16
}

1
use sea_orm_migration::prelude::*;
2

3
#[derive(DeriveMigrationName)]
4
pub struct Migration;
5

6
#[async_trait::async_trait]
7
impl MigrationTrait for Migration {
8
    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
9
        manager
10
            .create_table(
11
                Table::create()
12
                    .table(User::Table)
13
                    .if_not_exists()
14
                    .col(ColumnDef::new(User::Id).integer().not_null().auto_increment().primary_key())
15
                    .col(ColumnDef::new(User::Name).string().not_null())
16
                    .col(ColumnDef::new(User::Email).string().not_null().unique_key())
17
                    .col(ColumnDef::new(User::CreatedAt).timestamp().not_null().default(Expr::current_timestamp()))
18
                    .to_owned(),
19
            )
20
            .await
21
    }
22

23
    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
24
        manager
25
            .drop_table(Table::drop().table(User::Table).to_owned())
26
            .await
27
    }
28
}
29

30
#[derive(Iden)]
31
enum User {
32
    Table,
33
    Id,
34
    Name,
35
    Email,
36
    CreatedAt,
37
}

1
use rusqlite::{Connection, params, Result};
2

3
fn main() -> Result<()> {
4
    let conn = Connection::open("data.db")?;
5

6
    conn.execute_batch(
7
        "CREATE TABLE IF NOT EXISTS users (
8
            id INTEGER PRIMARY KEY AUTOINCREMENT,
9
            name TEXT NOT NULL,
10
            email TEXT UNIQUE NOT NULL
11
        )"
12
    )?;
13

14
    conn.execute(
15
        "INSERT INTO users (name, email) VALUES (?1, ?2)",
16
        params!["Alice", "alice@example.com"],
17
    )?;
18

19
    let mut stmt = conn.prepare("SELECT id, name, email FROM users")?;
20
    let users = stmt.query_map([], |row| {
21
        Ok(User {
22
            id: row.get(0)?,
23
            name: row.get(1)?,
24
            email: row.get(2)?,
25
        })
26
    })?;
27

28
    for user in users {
29
        println!("{:?}", user?);
30
    }
31

32
    Ok(())
33
}

1
use redb::{Database, TableDefinition};
2

3
const TABLE: TableDefinition<&str, &[u8]> = TableDefinition::new("my_data");
4

5
let db = Database::create("data.redb")?;
6

7
let write_txn = db.begin_write()?;
8
{
9
    let mut table = write_txn.open_table(TABLE)?;
10
    table.insert("key1", b"value1")?;
11
    table.insert("key2", b"value2")?;
12
}
13
write_txn.commit()?;
14

15
let read_txn = db.begin_read()?;
16
let table = read_txn.open_table(TABLE)?;
17
let value = table.get("key1")?.unwrap();
18
assert_eq!(value.value(), b"value1");

1
// SQLx
2
async fn transfer_funds(
3
    pool: &PgPool,
4
    from: i32,
5
    to: i32,
6
    amount: i64,
7
) -> Result<(), sqlx::Error> {
8
    let mut tx = pool.begin().await?;
9

10
    let balance: i64 = sqlx::query_scalar!(
11
        "SELECT balance FROM accounts WHERE id = $1 FOR UPDATE",
12
        from
13
    )
14
    .fetch_one(&mut *tx)
15
    .await?;
16

17
    if balance < amount {
18
        tx.rollback().await?;
19
        return Err(sqlx::Error::RowNotFound);
20
    }
21

22
    sqlx::query!(
23
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
24
        amount, from
25
    )
26
    .execute(&mut *tx)
27
    .await?;
28

29
    sqlx::query!(
30
        "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
31
        amount, to
32
    )
33
    .execute(&mut *tx)
34
    .await?;
35

36
    tx.commit().await?;
37
    Ok(())
38
}

1
use sea_orm::{DatabaseTransaction, TransactionTrait};
2

3
async fn transfer_funds(
4
    db: &DatabaseConnection,
5
    from: i32,
6
    to: i32,
7
    amount: i64,
8
) -> Result<(), DbErr> {
9
    let txn = db.begin().await?;
10

11
    let from_account = account::Entity::find_by_id(from)
12
        .lock(LockType::Update)
13
        .one(&txn)
14
        .await?
15
        .ok_or(DbErr::Custom("Account not found".into()))?;
16

17
    if from_account.balance < amount {
18
        txn.rollback().await?;
19
        return Err(DbErr::Custom("Insufficient funds".into()));
20
    }
21

22
    let mut from_active: account::ActiveModel = from_account.into();
23
    from_active.balance = Set(from_active.balance.unwrap() - amount);
24
    from_active.update(&txn).await?;
25

26
    let to_account = account::Entity::find_by_id(to)
27
        .lock(LockType::Update)
28
        .one(&txn)
29
        .await?
30
        .ok_or(DbErr::Custom("Account not found".into()))?;
31

32
    let mut to_active: account::ActiveModel = to_account.into();
33
    to_active.balance = Set(to_active.balance.unwrap() + amount);
34
    to_active.update(&txn).await?;
35

36
    txn.commit().await?;
37
    Ok(())
38
}

1
async fn nested_transaction(pool: &PgPool) -> Result<(), sqlx::Error> {
2
    let mut tx = pool.begin().await?;
3

4
    sqlx::query!("INSERT INTO logs (msg) VALUES ('outer')")
5
        .execute(&mut *tx)
6
        .await?;
7

8
    // 嵌套事务（SAVEPOINT）
9
    let mut savepoint = tx.begin().await?;
10

11
    sqlx::query!("INSERT INTO logs (msg) VALUES ('inner')")
12
        .execute(&mut *savepoint)
13
        .await?;
14

15
    // 可以回滚内层而不影响外层
16
    if something_went_wrong {
17
        savepoint.rollback().await?;
18
    } else {
19
        savepoint.commit().await?;
20
    }
21

22
    tx.commit().await?;
23
    Ok(())
24
}

1
// 乐观锁：用 version 字段检测并发冲突
2
async fn update_user_optimistic(
3
    pool: &PgPool,
4
    id: i32,
5
    name: String,
6
    expected_version: i32,
7
) -> Result<bool, sqlx::Error> {
8
    let result = sqlx::query!(
9
        r#"UPDATE users SET name = $1, version = version + 1
10
           WHERE id = $2 AND version = $3"#,
11
        name, id, expected_version
12
    )
13
    .execute(pool)
14
    .await?;
15

16
    Ok(result.rows_affected() > 0)
17
}
18

19
// 使用
20
match update_user_optimistic(pool, 1, "New Name".into(), 3).await? {
21
    true => println!("Updated"),
22
    false => println!("Conflict! Someone else modified the record"),
23
}

1
async fn update_user_pessimistic(
2
    pool: &PgPool,
3
    id: i32,
4
    name: String,
5
) -> Result<(), sqlx::Error> {
6
    let mut tx = pool.begin().await?;
7

8
    // 加行锁
9
    let user = sqlx::query_as!(
10
        User,
11
        "SELECT * FROM users WHERE id = $1 FOR UPDATE",
12
        id
13
    )
14
    .fetch_one(&mut *tx)
15
    .await?;
16

17
    // 此时其他事务无法读取此行（等待锁释放）
18
    sqlx::query!(
19
        "UPDATE users SET name = $1 WHERE id = $2",
20
        name, id
21
    )
22
    .execute(&mut *tx)
23
    .await?;
24

25
    tx.commit().await?;
26
    Ok(())
27
}

1
// ❌ 事务中混用 pool 和 tx
2
async fn bad(pool: &PgPool) -> Result<(), sqlx::Error> {
3
    let mut tx = pool.begin().await?;
4

5
    // 这个查询在 pool 上执行，不在事务中！
6
    sqlx::query!("INSERT INTO logs (msg) VALUES ('outside tx')")
7
        .execute(pool)  // ← 错误！
8
        .await?;
9

10
    // 应该在事务中执行
11
    sqlx::query!("INSERT INTO logs (msg) VALUES ('inside tx')")
12
        .execute(&mut *tx)  // ← 正确
13
        .await?;
14

15
    tx.commit().await?;
16
    Ok(())
17
}

1
// ❌ 在事务中做耗时操作
2
async fn bad_long_txn(pool: &PgPool) -> Result<(), sqlx::Error> {
3
    let mut tx = pool.begin().await?;
4

5
    // 锁定了行，然后做 30 秒的 HTTP 调用
6
    let user = sqlx::query_as!(User, "SELECT * FROM users WHERE id = 1 FOR UPDATE")
7
        .fetch_one(&mut *tx)
8
        .await?;
9

10
    let external = call_external_api(&user).await?;  // 30 秒！
11
    // 期间其他事务全部阻塞
12

13
    sqlx::query!("UPDATE users SET data = $1 WHERE id = 1", external)
14
        .execute(&mut *tx)
15
        .await?;
16

17
    tx.commit().await?;
18
    Ok(())
19
}
20

21
// ✓ 事务中只做数据库操作
22
async fn good_short_txn(pool: &PgPool) -> Result<(), sqlx::Error> {
23
    // 先在事务外做耗时操作
24
    let user = sqlx::query_as!(User, "SELECT * FROM users WHERE id = 1")
25
        .fetch_one(pool)
26
        .await?;
27
    let external = call_external_api(&user).await?;
28

29
    // 事务中只做快速更新
30
    let mut tx = pool.begin().await?;
31
    sqlx::query!("UPDATE users SET data = $1 WHERE id = 1", external)
32
        .execute(&mut *tx)
33
        .await?;
34
    tx.commit().await?;
35

36
    Ok(())
37
}