多个实例与唤醒 worker

本指南在服务的多个实例上运行一个队列，每个实例中有多个 worker，并设置它们发现新任务的速度。

读取一张队列表的所有实例共享这张表。除了数据库，没有任何东西协调这些实例：每个实例各自领取行，表的领取方式保证任何一行同一时间只属于一次领取。无论采用哪种方式、使用哪种数据库，两个领取者都绝不会取走同一行。

示例渲染发票。网店用自己的 SQL 写入发票任务，渲染服务按负载需要运行任意多个实例。

服务

宏
手写

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, Sqlite, SqliteConnection};

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

#[subscriber(InboxQueue::<RenderInvoice>::new("invoices"))]
async fn render(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.number, "rendering");
    HandlerOutcome::ack()
}

fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // A row another program writes is seen within a quarter of a second.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new())
        .poll_interval(Duration::from_millis(250))
        .route::<RenderInvoice>("invoices");
    RustStream::new(AppInfo::new("invoices", "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 invoices in work at once in this instance.
            b.include(render.workers(nonzero!(4)));
        })
}

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, Sqlite, SqliteConnection};

/// One row of `invoice_jobs`.
#[derive(sqlx::FromRow)]
struct RenderInvoice {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for RenderInvoice {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl PayloadRow for RenderInvoice {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

struct Render;

impl Handle<Invoice> for Render {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.number, "rendering");
        ready(Ok(()))
    }
}

fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // A row another program writes is seen within a quarter of a second.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new())
        .poll_interval(Duration::from_millis(250))
        .route::<RenderInvoice>("invoices");
    RustStream::new(AppInfo::new("invoices", "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 invoices in work at once in this instance.
            let render = subscriber("invoices", Render)
                .map_source(|_| InboxQueue::<RenderInvoice>::new("invoices"))
                .workers(nonzero!(4));
            b.include(render.build());
        })
}

workers(4) 让每个实例同时处理最多四张发票。订阅为每个空闲 worker 保持一次进行中的领取，所以它的 worker 同时取行，而不是一个接一个地取。SQLite 同一时间只接受一个写入者，订阅在那里只保持一次进行中的领取；在租约形式下，这次领取为每个 worker 各取一行。同一张表上的第二个实例再增加四个 worker，任何发票都不会同时交给两者。

新任务多快被发现

只要领取总是取满，订阅就持续领取，因此积压的任务无需停顿即可排空。取到的行少于请求数的领取之后，订阅等待它的轮询间隔，未设置时为一秒。新行的等待时间受三件事限制：

经 Broker 的发布会立即唤醒同一进程中订阅了该表的订阅，代价是每个被唤醒的订阅一次原子操作。
以其他途径写入的行要等待轮询间隔：另一个程序的插入、触发器、服务的 insert、另一个实例的发布。这里的 poll_interval(250 ms) 把这段等待限制在四分之一秒以内，代价是在空表上每个订阅每秒执行四条领取语句。
在 Postgres 上，一条通知会结束等待。

Postgres 上的通知

宏
手写

src/main.rs
use sqlx::PgConnection;
use sqlx::postgres::PgPoolOptions;

/// The same service on Postgres: notifications wake it, and the poll each minute reaches a row
/// whose notification was lost.
fn service_on_postgres(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    let broker = SqlxBroker::from_url(url, PgPoolOptions::new())
        .poll_interval(Duration::from_mins(1))
        .listen_notify();
    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        // The table and its trigger, created before the subscription checks them.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect(&schema_url).await?;
            sqlx::raw_sql(POSTGRES_SCHEMA).execute(&mut conn).await?;
            sqlx::raw_sql(ANNOUNCE).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(render.workers(nonzero!(4)));
        })
}

/// The trigger announces every row another program inserts: the channel is the table, and an
/// empty payload wakes every subscription of it.
const ANNOUNCE: &str = "
CREATE OR REPLACE FUNCTION announce_invoice() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(TG_TABLE_NAME, '');
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER invoice_announced AFTER INSERT ON invoice_jobs
    FOR EACH ROW EXECUTE FUNCTION announce_invoice();
";

src/main.rs
use sqlx::PgConnection;
use sqlx::postgres::PgPoolOptions;

/// The same service on Postgres: notifications wake it, and the poll each minute reaches a row
/// whose notification was lost.
fn service_on_postgres(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    let broker = SqlxBroker::from_url(url, PgPoolOptions::new())
        .poll_interval(Duration::from_mins(1))
        .listen_notify();
    RustStream::new(AppInfo::new("invoices", "0.1.0"))
        // The table and its trigger, created before the subscription checks them.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect(&schema_url).await?;
            sqlx::raw_sql(POSTGRES_SCHEMA).execute(&mut conn).await?;
            sqlx::raw_sql(ANNOUNCE).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            let render = subscriber("invoices", Render)
                .map_source(|_| InboxQueue::<RenderInvoice>::new("invoices"))
                .workers(nonzero!(4));
            b.include(render.build());
        })
}

/// The trigger announces every row another program inserts: the channel is the table, and an
/// empty payload wakes every subscription of it.
const ANNOUNCE: &str = "
CREATE OR REPLACE FUNCTION announce_invoice() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(TG_TABLE_NAME, '');
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER invoice_announced AFTER INSERT ON invoice_jobs
    FOR EACH ROW EXECUTE FUNCTION announce_invoice();
";

listen_notify() 在连接池的一个连接上监听，每个订阅监听它的表的通道。Broker 的每次发布在写入之后立即执行 pg_notify(channel, group)，所以另一个实例中的订阅一旦能看到这一行就领取它。RustStream 之外的写入者自己发出同样的通知；上面的触发器为每次插入发出通知，它的空载荷唤醒该表的每个订阅。有了通知，轮询间隔就成为兜底，用来接住通知遗漏的行，这里是一分钟。

通知有代价，这个开关在未设置时关闭。Broker 在整个生命周期内占用连接池的一个连接。每次发布多执行一条语句。发出通知的事务在提交时获取一个服务器全局的锁，因此发出通知的事务依次提交；有大量并发写入者的服务，应在启用通知之前测量这一点（NOTIFY）。

worker 与连接池

一个订阅占用的连接从不超过连接池大小减一，所以连接池始终为处理器、处理器的发布和结算留出一个连接。连接池太小，容不下它的 worker 时，同时运行的投递就少一些，但不会失败。在租约方式下，n + 1 的连接池让 workers(n) 的每个空闲 worker 同时领取；行锁方式和咨询锁方式还要为每个处理中的任务占用一个连接（连接）。在生产环境中运行一页为整个服务确定连接池大小。

时钟

每个实例从它所在主机的时钟读取“现在”，用来领取、推迟和租用行（时间）。服务同一张表的各台主机要让时钟保持同步，误差远小于一个租约，NTP 就能做到。做不到时，使用 DatabaseClock 的表在每条语句中读取数据库的时钟；租约方式始终使用主机的时钟，所以这样的表采用行锁方式或咨询锁方式。

测试

示例文件中的测试在内存中的 SQLite 上运行服务。两个实例各有自己的连接池，在一张存有二十张发票的表上启动，它们合起来把每张发票渲染一次。另一个程序写入的行，在四分之一秒的轮询间隔内被发现。经 Broker 的发布立即到达它自己的实例。这些行不唤醒任何订阅，所以等待它们的测试用 TestApp::start_live 在真实时钟上运行。最后一个测试让 Postgres 服务连接它自己的数据库，发现一次插入远在一分钟的轮询之前就被领取；它从 POSTGRES_TEST_URL 读取服务器地址，没有这个变量时跳过。

服务​

新任务多快被发现​

Postgres 上的通知​

worker 与连接池​

时钟​

测试​

服务

新任务多快被发现

Postgres 上的通知

worker 与连接池

时钟

测试