跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

数据库内的精确一次：事务处理器

本指南把处理器的写入与任务的结算一起提交，因此效果是同一数据库中的行的任务，恰好生效一次。

队列把每个任务至少投递一次。处理器写入结果后、任务结算前崩溃，就会再次运行，把结果写两遍。当结果是保存队列的那个数据库中的行时，一个事务就能消除这种重复：写入和结算要么一起提交，要么都不提交。

示例是一个退款服务。一次退款写入它的记录和一个交给发信服务的收据任务，并调用银行卡处理商。

表​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS refunds (order_id INTEGER PRIMARY KEY, amount_cents INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS refund_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS receipt_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `refund_jobs`, the refunds the shop's support queues.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "refund_jobs")]
struct IssueRefund {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One row of `receipt_jobs`, the mailer's queue.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "receipt_jobs")]
struct SendReceipt {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

refund_jobs 是本服务读取的队列。receipt_jobs 是发信服务的队列：本服务只向其中写入，用派生宏生成的 insert；在手写形式中，用它自己的一条语句。

处理器​

src/main.rs
/// The refund's record and its receipt, written on the connection of the delivery's transaction.
async fn record(
    conn: &mut SqliteConnection,
    refund: &Refund,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO refunds (order_id, amount_cents) VALUES (?, ?)")
        .bind(refund.order)
        .bind(refund.amount_cents)
        .execute(&mut *conn)
        .await?;
    let receipt = SendReceipt {
        id: 0,
        locked_until: None,
        payload: serde_json::to_vec(&Receipt {
            order: refund.order,
        })?,
    };
    receipt.insert(conn).await?;
    Ok(())
}

#[subscriber(InboxQueue::<IssueRefund>::new("refunds"))]
async fn issue(
    refund: &Refund,
    Ctx(mut tx): Ctx<keys::Tx<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    if let Err(error) = record(&mut tx, refund).await {
        tracing::warn!(subscription = "refunds", refund.order, %error, "refund not recorded");
        return HandlerOutcome::drop();
    }
    // The processor takes the order as its idempotency key: a call repeated after a retry
    // refunds the card once.
    if !card_refund(refund, attempt).await {
        // The retry rolls the record and the receipt back first.
        return HandlerOutcome::retry();
    }
    // The commit keeps the record and the receipt, and finishes the refund's row.
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<IssueRefund>("refunds"), |b| {
            b.include(issue.transactional());
        })
}

挂载处的 .transactional() 把投递的事务借给处理器。处理器以第一个 Ctx 参数 Ctx<keys::Tx<Sqlite>> 接收它，并通过 &mut *tx 写入。在没有这个步骤的订阅上接收 keys::Tx 的处理器无法编译。

确认在一个事务中提交退款记录、收据任务和退款行的结算。其他任何结果都先回滚这些写入，再结算该行；重试仍然计入这次尝试。panic 什么也不提交。同一订单的第二次退款在 refunds 的主键上失败并被丢弃，因此进入队列的重复任务不会改变任何东西。

收据任务构成一条链：只有退款提交之后，发信服务才能看到它，回滚的退款则永远不会产生收据。通过 Ctx<keys::Pool> 的写入单独提交，与结果无关，经 Broker 的发布也是如此；两者都不加入这个事务。

留在事务之外的部分​

银行卡处理商在数据库之外，因此没有事务能覆盖它。处理商已经退款之后的重试会再次调用它，所以调用带上幂等键，这里是订单，处理商只退款一次。数据库之外的每种效果都以同样的方式保持至少一次：一封邮件、一次 HTTP 调用、一次向另一个 Broker 的发布。把事件中继到另一个 Broker 和 outbox 两页处理发布。

在 SQLite 上，投递的事务在第一次写入时获取数据库的写锁，一直持有到投递结算，在此期间其他所有写入者都无法进入，领取也包括在内。这里处理商的调用在两次插入之后运行，所以调用期间数据库对其他写入者保持锁定。服务器只锁住事务写入的行。在 SQLite 上，发布 outbox 所跟踪消息的处理器，要在通过 tx 第一次写入之前发布：记录在自己的连接上写入，而它需要投递持有的写锁。在这次写入之后的跟踪发布以 OutboxError::WriteLockHeld 失败，消息不会发送。

隔离级别​

每个投递的事务按表的隔离级别打开，在服务器上是 #[inbox(isolation = read_committed)]；在 SQLite 上则按它的模式打开，即 #[inbox(mode = immediate)]。两者都未声明的表采用数据库的默认级别，在 MySQL 和 MariaDB 上是 READ COMMITTED。每个处理中的投递都为它的事务占用一个连接，所以通过 keys::Pool 或一次发布再取一个连接的处理器，需要比它的 workers(n) 更大的连接池。事务模式服务于逐条投递；以这种模式挂载的批处理器无法编译。在 MySQL 和 MariaDB 上，处理器若中途截断一条通过 tx 执行的语句，例如用 select! 包住一次查询，可能扰乱随后的结算，这是 sqlx-mysql 0.9.0 的一个缺陷：让每条语句执行完。

测试​

示例文件中的测试用 TestApp::start 在内存中的 SQLite 上运行服务，每次发布后读取这些表。一次退款提交一条记录和一张收据。第一次调用处理商超时的退款会重试，第二次尝试留下一条记录和一张收据，而不是两条。同一笔退款发布两次，第二次被丢弃。