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

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

任务入队

本指南向队列表写入任务：与任务所服务的数据在同一个事务中写入、排期到以后执行，以及按优先级执行。

与独立的 Broker 相比，服务数据库中的队列有一个优势：任务与它所服务的数据一起提交。回滚的订单不会留下确认邮件，没有保存的订单也绝不会发出邮件。

示例是一个网店。结账订阅负责下单，邮件订阅立即发送订单确认，并在三天后发送评价邀请。

表​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, email TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS checkout_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS email_jobs (
    id INTEGER PRIMARY KEY,
    priority INTEGER NOT NULL,
    retry_after TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    processed_at TEXT,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqlitePool};

/// One row of `email_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(priority)]
    priority: i64,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A smaller priority is claimed first.
const URGENT: i64 = 0;
const ROUTINE: i64 = 10;

impl SendEmail {
    /// A job for `email`, due at `due`.
    fn new(email: &Email, priority: i64, due: DateTime<Utc>) -> serde_json::Result<Self> {
        Ok(Self {
            id: 0,
            priority,
            retry_after: due,
            attempt: 1,
            processed_at: None,
            locked_until: None,
            payload: serde_json::to_vec(email)?,
        })
    }
}

每一列开启队列的一种行为（角色）。retry_after 保存一行到期的时间：领取只取时间已到的行。priority 决定领取的顺序：值越小越早，然后是 retry_after 较早的行，最后是 id 较小的行。processed_at 保留已完成的行：确认把时间写入这一列，而不删除该行。generated 标记由数据库填充的列，因此插入语句不包含这些列。在手写形式中，服务以 Insert 实现自己编写这条插入语句。

订单事务中的任务​

src/main.rs
/// The order, its confirmation and a reminder three days later commit together, or not at all.
async fn place_order(pool: &SqlitePool, order: &Order) -> Result<(), Box<dyn Error>> {
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO orders (id, email) VALUES (?, ?)")
        .bind(order.id)
        .bind(&order.email)
        .execute(&mut *tx)
        .await?;
    let now = Utc::now();
    let confirmation = Email::confirmation(order);
    SendEmail::new(&confirmation, URGENT, now)?
        .insert(&mut *tx)
        .await?;
    let reminder = Email::review_request(order);
    SendEmail::new(&reminder, ROUTINE, now + TimeDelta::days(3))?
        .insert(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(())
}

place_order 打开一个 sqlx 事务，在其中写入三行：订单、确认邮件和提醒邮件。派生宏生成的 insert 接受该表所在数据库的任意连接，因此 &mut *tx 把任务放进订单的事务。它们要么一起提交，要么一起回滚：插入失败的订单不会在表中留下邮件。下单的 HTTP 端点用服务自己的连接池调用同一个函数；端点和队列共用一个连接池的服务，把它交给 SqlxBroker::new。

确认邮件现在就到期，而且紧急。提醒邮件三天后到期，在那之前没有领取会取走它。周期性任务也是同样的做法：它的处理器插入下一行，并设置更晚的 retry_after。只能存在一次的任务，则由服务自己的唯一索引加上一条单独的 INSERT .. ON CONFLICT DO NOTHING 语句来保证。

订阅​

src/main.rs
#[subscriber(InboxQueue::<Checkout>::new("checkouts"))]
async fn checkout(order: &Order, Ctx(pool): Ctx<keys::Pool<Sqlite>>) -> HandlerOutcome {
    match place_order(&pool, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "checkouts", order.id, %error, "order not placed");
            HandlerOutcome::drop()
        }
    }
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(order = email.order, template = %email.template, "sending");
    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("shop", "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::<Checkout>("checkouts"), |b| {
            b.include(checkout);
            b.include(send);
        })
}

checkout 从上下文中取出连接池并下单；下单失败的订单以一条警告被丢弃。它自己的事务与结账行的领取相互独立：写入必须与所在行的结算一起提交的处理器，应以事务模式运行。send 按领取顺序读取 email_jobs 的每一行。它的确认用 processed_at 标记该行，因此已发送的邮件留在表中以备审计，领取会跳过它们。没有这一列的表在确认时删除该行。

任务何时被发现​

经 Broker 的发布会立即唤醒同一进程中订阅了该表的订阅：路由、Repository、应答或 Out 槽位都是如此。前端发布的结账请求就以这种方式唤醒 checkout。服务用自己的 SQL 写入的行，例如 place_order 写入的行，不唤醒任何订阅。订阅在它的轮询间隔之后发现这一行，未设置时为一秒。Broker 的发布不加入服务的事务，所以必须与订单一起提交的任务通过 insert 写入，并等待这个间隔。在 Postgres 上，由服务或触发器发出的通知会提前结束等待。

测试​

示例文件中的测试通过测试框架发布结账请求，然后检查表和处理器。邮件行在下一次轮询时到达它的订阅，所以等待邮件的测试用 TestApp::start_live 启动应用，并用 tb.advance 让两秒过去。一个测试检查确认邮件已经发出，而提醒邮件仍在等待。一个测试把同一个订单下两次，发现只有第一次下单的邮件。一个测试在服务启动之前写入一封普通邮件和一封紧急邮件，检查紧急邮件先发出。在 SQL 上测试服务介绍了测试框架。