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

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

Рассылка на PUB/SUB

Это руководство рассылает оповещения о событиях на PUB/SUB: один сервис публикует, и каждый подписавшийся наблюдатель получает копию.

В примере два сервиса. Воркер берёт задания и оповещает о каждом выполненном задании через широковещательную конечную точку. Наблюдатель подписывается на все оповещения о заданиях.

Источник оповещений​

src/main.rs
// One file, two patterns: the crate prelude names each policy after its pattern.
use ruststream_zeromq::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
    items: u32,
}

/// The announcement, under a name every `jobs.` watcher matches.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "jobs.done")]
struct JobDone {
    id: u64,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> JobDone {
    tracing::info!(job.id, job.items, "job worked");
    JobDone { id: job.id }
}

#[ruststream::app]
fn app() -> impl App {
    // The watchers dial this endpoint; the announcer is the stable side, so it binds.
    let events = ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")).bindable();
    let announce = events.bind(ZmqFanoutPublish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                b.include(work).out_reply(announce);
            },
        )
        .with_broker(events, |_b| {})
}

Воркер читает задания из ZmqQueue и оповещает через ZmqFanout, паттерн PUB/SUB. Файл, который монтирует два паттерна, импортирует прелюдию крейта. В ней каждая политика названа по своему паттерну, здесь это ZmqFanoutPublish. Источник оповещений - устойчивая сторона, поэтому он привязывается к адресу, а наблюдатели подключаются к нему. .out_reply(announce) публикует каждый JobDone под его именем, jobs.done.

Наблюдатель​

src/main.rs
use ruststream_zeromq::prelude::*;
use serde::Deserialize;

#[derive(Debug, PartialEq, Deserialize)]
struct JobEvent {
    id: u64,
}

/// Every job announcement: `jobs.done`, `jobs.failed`, and whatever `jobs.` name comes next.
#[subscriber("jobs.")]
async fn watch(event: &JobEvent, ctx: &mut Context) -> HandlerOutcome {
    tracing::info!(event.id, name = ctx.name(), "job announced");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    // A subscription that dials names where its retry copies go: a queue this service binds.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5560")).bindable();
    let copies = retries.bind(ZmqQueuePublish);
    RustStream::new(AppInfo::new("watcher", "0.1.0"))
        .with_broker(
            ZmqFanout::new(ZmqEndpoint::connect("tcp://worker:5556")),
            |b| {
                b.include(watch).out_retry(copies).to("jobs.");
            },
        )
        .with_broker(retries, |b| {
            b.include(watch);
        })
}

На широковещательной конечной точке имя подписки - префикс имени сообщения, и они сравниваются побайтно. jobs. получает jobs.done, jobs.failed и любое другое имя на jobs., и больше ничего. ctx.name() - имя, под которым пришло сообщение. Каждый наблюдатель, подключённый к источнику, получает свою копию каждого оповещения.

Наблюдатель подключается, поэтому его регистрация называет, куда уходят копии для повтора, хотя watch никогда не просит повтора. Копии уходят в очередь, которую наблюдатель привязывает для себя, и её читает тот же обработчик. Это правило разбирает руководство Повторы и очереди dead-letter.

Опоздавшие наблюдатели​

Широковещательная конечная точка ничего не хранит. Наблюдатель получает то, что опубликовано после того, как его подписка дошла до издателя, и ничего из более раннего. Наблюдатель, который запустился поздно или переподключился, пропускает оповещения за это время. Оповещение, которому не подошёл ни один наблюдатель, отбрасывается без ошибки.

Сервис, который оповещает собственную подписку, так ничего не теряет: его издатель читает фильтр этой подписки до первой отправки.

Медленные наблюдатели​

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

Повторы на широковещательной конечной точке​

Копия для повтора на широковещательной конечной точке - такая же публикация, как любая другая: она доходит до каждой подписки с совпавшим префиксом. Если конечную точку привязывает сервис, копия снова уходит всем, кто получил оригинал. Имя dead-letter на той же конечной точке тоже совпадает с префиксом: jobs.dead приходит на jobs. и обрабатывается заново. Поэтому dead-letter широковещательной конечной точки уходят на отдельную конечную точку.

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

Тест источника оповещений отправляет задание и читает оповещение, опубликованное в jobs.done. Тест наблюдателя публикует jobs.done, jobs.failed и metrics.cpu так, как это сделал бы источник, и проверяет, что watch получил первые два. Внутрипроцессный режим применяет правило префикса. Опоздавшие подписчики и обратное давление есть только на настоящих сокетах. Оба режима разбирает руководство Тестирование сервиса на ZeroMQ.