Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Несколько экземпляров и пробуждение воркеров

Это руководство запускает очередь на нескольких экземплярах сервиса с несколькими воркерами в каждом и задаёт, как быстро они видят новую задачу.

Таблица очереди общая для всех экземпляров, которые её читают. Экземпляры координирует только база данных: каждый захватывает строки, а форма таблицы держит любую строку не больше чем в одном захвате одновременно. Два захвата никогда не берут одну и ту же строку, в любой форме и на любой базе данных.

Пример строит счета-фактуры. Магазин записывает задачи на счета собственным 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)));
        })
}

workers(4) строит в каждом экземпляре до четырёх счетов одновременно. Подписка держит по захвату на каждого свободного воркера, поэтому её воркеры берут строки сразу, а не друг за другом. На SQLite, который допускает одного писателя за раз, она держит один захват, и в форме аренды этот захват берёт по строке на каждый воркер. Второй экземпляр на той же таблице добавляет ещё четыре воркера, и ни один счёт не попадает к обоим.

Как быстро видна новая задача​

Подписка захватывает, пока её захваты возвращаются полными, поэтому накопившиеся строки разбираются без пауз. После захвата, который нашёл меньше строк, чем просил, она ждёт интервал опроса, по умолчанию одну секунду. Сколько ждёт новая строка, определяют три вещи:

  • Публикация через брокер сразу будит подписки своей таблицы в том же процессе, ценой одной атомарной операции на каждую разбуженную подписку.
  • Строка, записанная иначе, ждёт интервал опроса: вставка другой программы, триггер, 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();
";

listen_notify() слушает на одном соединении пула, а каждая подписка слушает канал своей таблицы. Каждая публикация брокера выполняет pg_notify(channel, group) сразу после своей записи, поэтому подписка в другом экземпляре захватывает строку, как только та становится видна. Писатель вне RustStream отправляет такое же уведомление сам; триггер выше делает это при каждой вставке, а его пустая полезная нагрузка будит каждую подписку таблицы. С уведомлениями интервал опроса становится запасным путём для того, что уведомление пропустило, здесь минута.

У уведомлений есть цена, и если переключатель не задан, они выключены. Брокер держит одно соединение пула на всё время своей жизни. Каждая публикация выполняет ещё один оператор. Транзакция, которая отправляет уведомление, при фиксации берёт блокировку, глобальную для сервера, поэтому фиксации таких транзакций выполняются одна за другой; сервис с множеством одновременных писателей измеряет это, прежде чем включать уведомления (NOTIFY).

Воркеры и пул​

Подписка никогда не держит больше соединений, чем размер пула минус одно, поэтому в пуле всегда остаётся одно для обработчиков, их публикаций и завершений. Пул, слишком маленький для своих воркеров, выполняет меньше доставок одновременно, но не возвращает ошибку. В форме аренды пул из n + 1 соединений позволяет каждому свободному воркеру workers(n) захватывать одновременно; формы блокировки строк и рекомендательной блокировки к тому же держат соединение на каждую задачу в работе (Соединения). Пул для всего сервиса рассчитывает руководство Работа в промышленной среде.

Часы​

Каждый экземпляр читает «сейчас» по часам своего хоста, чтобы захватывать, откладывать и арендовать строки (Время). Хосты, которые обслуживают одну таблицу, держат часы согласованными, с расхождением намного меньше срока аренды, как это делает NTP. Где это невозможно, таблица на DatabaseClock читает часы базы данных в каждом операторе; форма аренды сохраняет часы хоста, поэтому такая таблица использует форму блокировки строк или рекомендательной блокировки.

Тестирование​

Тесты в файле примера запускают сервис на SQLite в памяти. Два экземпляра, каждый со своим пулом, стартуют на одной таблице с двадцатью счетами и вместе строят каждый счёт один раз. Строку, которую записала другая программа, сервис видит в пределах четверти секунды интервала опроса. Публикация через брокер сразу доходит до своего экземпляра. Эти строки не будят ни одну подписку, поэтому тесты, которые их ждут, работают на реальных часах через TestApp::start_live. Последний тест запускает сервис для Postgres на собственной базе данных и находит, что вставка захвачена задолго до опроса раз в минуту; он читает адрес сервера из POSTGRES_TEST_URL и без него пропускается.