跳到主要内容
本页尚未翻译

下面显示的是英文原文。

Workers

workers(n) keeps a claim in flight for each free worker, bounded by the pool; with a partition_key column workers_by_key(n) keeps each key in order.

A subscription with n workers runs up to n deliveries at once. It keeps a claim in flight for each free worker, so its workers take rows at once instead of one after another. It never holds more than the pool's size less one connection, so the pool keeps one for the handlers' own queries, their publishes and the settlements that take one (Connections). A subscription keeps one claim in flight where claims that run at once would cost more than they give. On SQLite, which takes one writer at a time, they only collide. On a table whose groups keep their order, or with a partition_key column, they could finish out of claim order. On MySQL and MariaDB a lease claim locks the rows it reads, leases in work included, until it commits. One claim in flight still feeds every worker: the next claim starts as soon as the last one returns. On SQLite a lease claim takes a row for every worker in one statement. The rows past the free workers wait in the subscription for the next free worker, and their leases are extended while they wait. Two claims never take the same row.

How many workers a SQLite queue uses depends on what its handlers write. In the lease and advisory lock forms the throughput grows with the workers. In transactional mode a handler's first write takes the database's write lock, and the lock stays until the delivery settles. Every other writer then waits in SQLite's busy handler, which sleeps a millisecond or more instead of queueing it. A transactional subscription on SQLite whose handlers write gains little from a second worker in the lease form, and loses most of its throughput to these waits in the advisory lock form: it runs one worker there.

A #[field(partition_key)] column holds each row's key: text or bytes, optional or not (KeyColumn). workers_by_key(n) sends the deliveries of one key to one worker, so the rows of a key are handled in the order they were claimed, and different keys run side by side. By hand, .partition_key(..) names the column, Key goes into the description's type, and the row reads its key through KeyRow.

Example​

Images resize on four workers in any order, and the ledger's postings of one account stay in order on one of four workers.

examples/sqlx_subscriptions_workers.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `image_jobs`: images resize in any order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "image_jobs")]
struct Resize {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One row of `ledger_jobs`: the postings of one account apply in the order they were written.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "ledger_jobs")]
struct Post {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    account: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<Resize>::new("images"))]
async fn resize(image: &Image) -> HandlerOutcome {
    tracing::info!(path = %image.path, "resizing");
    HandlerOutcome::ack()
}

#[subscriber(InboxQueue::<Post>::new("postings"))]
async fn post(posting: &Posting) -> HandlerOutcome {
    tracing::info!(account = %posting.account, posting.n, "posting");
    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());
    let broker = broker.route::<Resize>("images").route::<Post>("postings");
    RustStream::new(AppInfo::new("back-office", "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, |b| {
            // Up to four images in work at once.
            b.include(resize.workers(nonzero!(4)));
            // Four workers, and the postings of one account on one of them, one after another.
            b.include(post.workers_by_key(nonzero!(4)));
        })
}