从单机到分布式：一致性实战（一）单机时代：ACID 的庇护

pub async fn create_order(
    pool: &PgPool,
    buyer_id: i64,
    book_id: i64,
    price: Decimal,
) -> Result<Order> {
    let mut tx = pool.begin().await?;

    // 1. 创建订单
    let order = sqlx::query_as!(
        Order,
        r#"INSERT INTO orders (buyer_id, book_id, price, status)
           VALUES ($1, $2, $3, 'created')
           RETURNING *"#,
        buyer_id, book_id, price
    )
    .fetch_one(&mut *tx)
    .await?;

    // 2. 扣减库存（标记书籍为已售）
    let rows = sqlx::query!(
        r#"UPDATE books SET status = 'sold' WHERE id = $1 AND status = 'available'"#,
        book_id
    )
    .execute(&mut *tx)
    .await?
    .rows_affected();

    if rows == 0 {
        return Err(Error::BookNotAvailable);
    }

    // 3. 扣减买家余额
    let rows = sqlx::query!(
        r#"UPDATE accounts SET balance = balance - $1 WHERE user_id = $2 AND balance >= $1"#,
        price, buyer_id
    )
    .execute(&mut *tx)
    .await?
    .rows_affected();

    if rows == 0 {
        return Err(Error::InsufficientBalance);
    }

    // 4. 增加卖家余额
    let rows = sqlx::query!(
        r#"UPDATE accounts SET balance = balance + $1 WHERE user_id = (
            SELECT seller_id FROM books WHERE id = $2
        )"#,
        price, book_id
    )
    .execute(&mut *tx)
    .await?
    .rows_affected();

    if rows == 0 {
        return Err(Error::SellerAccountNotFound);
    }

    // 提交事务
    tx.commit().await?;

    Ok(order)
}

保护力度：低 ──────────────────────────────────→ 高
         READ UNCOMMITTED → READ COMMITTED → REPEATABLE READ → SERIALIZABLE
                              ↑
                              │
                         PostgreSQL 默认

事务 A：张三余额从 1000 改成 500（还没提交）
事务 B：读取张三余额 = 500 ← 脏读！
事务 A：回滚
结果：事务 B 读到了一个从未存在过的值

事务 A：读取张三余额 = 1000
事务 B：张三余额改成 500，提交
事务 A：再次读取张三余额 = 500 ← 不可重复读！

事务 A：SELECT COUNT(*) FROM orders WHERE user_id = 1  → 10 条
事务 B：INSERT INTO orders (user_id, ...) VALUES (1, ...)，提交
事务 A：SELECT COUNT(*) FROM orders WHERE user_id = 1  → 11 条 ← 幻读！

// 默认：READ COMMITTED，适合大多数场景
sqlx::query!("SELECT * FROM books WHERE id = $1", book_id)
    .fetch_one(pool)
    .await?;

// 需要可重复读：金融对账、报表统计
sqlx::query!("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
    .execute(&mut *tx)
    .await?;

// 需要严格一致性读：统计、审计（注意：会增加序列化失败风险）
sqlx::query!("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
    .execute(&mut *tx)
    .await?;

// FOR UPDATE：锁定这一行，其他事务等着
let book = sqlx::query_as!(
    Book,
    r#"SELECT * FROM books WHERE id = $1 FOR UPDATE"#,
    book_id
)
.fetch_one(&mut *tx)
.await?;

// 现在可以安全地修改了
sqlx::query!(
    r#"UPDATE books SET status = 'sold' WHERE id = $1"#,
    book_id
)
.execute(&mut *tx)
.await?;

// 用版本号实现乐观锁
let book = sqlx::query_as!(
    Book,
    r#"SELECT * FROM books WHERE id = $1"#,
    book_id
)
.fetch_one(pool)
.await?;

// 更新时检查版本号
let rows = sqlx::query!(
    r#"UPDATE books
       SET status = 'sold', version = version + 1
       WHERE id = $1 AND version = $2"#,
    book_id, book.version
)
.execute(pool)
.await?
.rows_affected();

if rows == 0 {
    return Err(Error::ConcurrentModification);
}

┌─────────────────────────────────────────────────────────┐
│                      用户请求                           │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│                    Nginx (反向代理)                      │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│               Rust + Axum (单实例)                       │
│                                                         │
│   ┌─────────────────────────────────────────────────┐   │
│   │              业务逻辑                            │   │
│   │  create_order() / get_books() / ...            │   │
│   └─────────────────────────────────────────────────┘   │
└─────────────────────────┬───────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│                   PostgreSQL (单实例)                    │
│                                                         │
│   ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐   │
│   │  users  │  │  books  │  │ orders  │  │accounts │   │
│   └─────────┘  └─────────┘  └─────────┘  └─────────┘   │
└─────────────────────────────────────────────────────────┘

use sqlx::{PgPool, Postgres, Transaction};
use uuid::Uuid;

pub struct UserService {
    pool: PgPool,
}

impl UserService {
    /// 用户注册：创建用户 + 初始化账户余额
    pub async fn register(&self, username: &str, email: &str) -> Result<User> {
        let mut tx: Transaction<'_, Postgres> = self.pool.begin().await?;

        // 1. 检查用户名是否已存在
        let exists = sqlx::query_scalar!(
            r#"SELECT EXISTS(SELECT 1 FROM users WHERE username = $1)"#,
            username
        )
        .fetch_one(&mut *tx)
        .await?
        .unwrap_or(false);

        if exists {
            return Err(Error::UsernameExists);
        }

        // 2. 创建用户
        let user = sqlx::query_as!(
            User,
            r#"INSERT INTO users (id, username, email, created_at)
               VALUES ($1, $2, $3, NOW())
               RETURNING *"#,
            Uuid::new_v4(),
            username,
            email
        )
        .fetch_one(&mut *tx)
        .await?;

        // 3. 初始化账户（余额为 0）
        sqlx::query!(
            r#"INSERT INTO accounts (user_id, balance) VALUES ($1, 0)"#,
            user.id
        )
        .execute(&mut *tx)
        .await?;

        // 4. 记录注册日志
        sqlx::query!(
            r#"INSERT INTO user_logs (user_id, action, created_at)
               VALUES ($1, 'register', NOW())"#,
            user.id
        )
        .execute(&mut *tx)
        .await?;

        // 提交事务
        tx.commit().await?;

        Ok(user)
    }
}