Типы подписки и порядок

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

Пример - сервис доставки. Он упаковывает заказы, отслеживает отправления, проводит платежи и выставляет счета по заказам, и каждая из этих задач читает через подписку своего типа.

Четыре типа

Каждый консьюмер подписки - это обработчик в одном из экземпляров сервиса. Тип подписки решает, как эти консьюмеры делят сообщения:

Тип	Кто получает сообщение	Порядок
Shared (по умолчанию)	любой консьюмер, по очереди	не сохраняется
KeyShared	консьюмер, которому принадлежит ключ сообщения	для каждого ключа
Exclusive	единственный консьюмер; второй ждёт его ухода	порядок топика
Failover	активный консьюмер; остальные в резерве	порядок топика

Тип задаёт subscription_type в дескрипторе. Все консьюмеры одной подписки называют один и тот же тип.

Shared: любой консьюмер

Макросы

src/main.rs

/// One step in the life of a shipment.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order: u64,
    status: String,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// Packs an order. Every instance of the service shares the subscription `packing`.
#[subscriber(PulsarSubscription::new("orders", "packing"))]
async fn pack(
    order: &Order,
    Out(shipments): Out<impl Publisher<Options = PulsarPublishOptions>, Shipments>,
) -> HandlerOutcome {
    let packed = Shipment {
        order: order.id,
        status: "packed".to_owned(),
    };
    // The order's id keys the event, so every event of one order keeps its place in line.
    let sent = shipments
        .message(&packed)
        .partition_key(order.id.to_string())
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, "order packed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// One step in the life of a shipment.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[outgoing(name = "shipments")]
struct Shipment {
    order: u64,
    status: String,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// Packs an order. Every instance of the service shares the subscription `packing`.
struct Pack;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Pack
where
    Entry: OutEntry<Shipments, Wire: Publisher<Options = PulsarPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let packed = Shipment {
            order: order.id,
            status: "packed".to_owned(),
        };
        // The order's id keys the event, so every event of one order keeps its place in line.
        outs.get(Shipments)
            .message(&packed)
            .partition_key(order.id.to_string())
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        tracing::info!(order.id, "order packed");
        Ok(())
    }
}

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

pack публикует событие отправления для каждого заказа. Шаг partition_key задаёт ключом события номер заказа. Этот шаг берётся из прелюдии брокера, поэтому слот ограничен Publisher<Options = PulsarPublishOptions>, чтобы шаг был доступен. Публикация возвращает управление, когда брокер сохранил сообщение.

KeyShared: по порядку для каждого ключа

Макросы

src/main.rs

/// Follows each shipment. Events of one order arrive in the order they were published.
#[subscriber(
    PulsarSubscription::new("shipments", "tracking")
        .subscription_type(SubscriptionType::KeyShared),
    workers(8, by_key)
)]
async fn track(event: &Shipment) -> HandlerOutcome {
    tracing::info!(event.order, event.status, "shipment tracked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Follows each shipment. Events of one order arrive in the order they were published.
struct Track;

impl Handle<Shipment> for Track {
    fn handle(
        &self,
        event: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.order, event.status, "shipment tracked");
        ready(Ok(()))
    }
}

fn tracking() -> PulsarSubscription {
    PulsarSubscription::new("shipments", "tracking").subscription_type(SubscriptionType::KeyShared)
}

KeyShared отдаёт все сообщения одного ключа одному консьюмеру, поэтому события одного заказа приходят в track в том порядке, в каком их опубликовал pack. Разные ключи распределяются по консьюмерам, поэтому экземпляры делят работу. Сервер назначает ключи и переносит часть из них, когда консьюмер подключается или уходит.

Внутри одного экземпляра workers(8, by_key) запускает восемь партиций воркеров, и доставки одного ключа выполняются друг за другом на одной партиции воркеров. Партицию воркеров выбирает ключ партиционирования сообщения, поэтому порядок каждого ключа сохраняется и внутри процесса. В ручной форме та же настройка - workers_by_key. Ключ приходит в обработчик заголовком partition-key.

Exclusive: один консьюмер

Макросы

src/main.rs

/// Books payments one after another, in the order they were taken.
#[subscriber(
    PulsarSubscription::new("payments", "ledger").subscription_type(SubscriptionType::Exclusive)
)]
async fn book(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.order, payment.amount_cents, "payment booked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Books payments one after another, in the order they were taken.
struct Book;

impl Handle<Payment> for Book {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.order, payment.amount_cents, "payment booked");
        ready(Ok(()))
    }
}

fn ledger() -> PulsarSubscription {
    PulsarSubscription::new("payments", "ledger").subscription_type(SubscriptionType::Exclusive)
}

Exclusive даёт читать подписку одному консьюмеру, поэтому book видит платежи в том порядке, в каком их приняли. Второму консьюмеру сервер отказывает как занятой, а клиент повторяет запрос, пока первый не уйдёт или не кончатся попытки OperationRetries, а по умолчанию они не кончаются никогда. Поэтому второй экземпляр сервиса не запускается и ничего не сообщает, пока ждёт. Подписка Exclusive принадлежит сервису, который работает в одном экземпляре.

Failover: один активный консьюмер

Макросы

src/main.rs

/// Invoices orders on one instance at a time; the others stand by.
#[subscriber(
    PulsarSubscription::new("orders", "invoicing").subscription_type(SubscriptionType::Failover)
)]
async fn invoice(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order invoiced");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Invoices orders on one instance at a time; the others stand by.
struct Invoice;

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

fn invoicing() -> PulsarSubscription {
    PulsarSubscription::new("orders", "invoicing").subscription_type(SubscriptionType::Failover)
}

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

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    // A second ledger consumer reports the refusal after three tries instead of waiting.
    let retries = OperationRetries::attempts(nonzero!(3u32));
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").operation_retries(retries),
        |b| {
            b.include(pack).out(Shipments, Publish).build();
            b.include(track);
            b.include(book);
            b.include(invoice);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    // A second ledger consumer reports the refusal after three tries instead of waiting.
    let retries = OperationRetries::attempts(nonzero!(3u32));
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650").operation_retries(retries),
        |b| {
            b.include(subscriber(PulsarSubscription::new("orders", "packing"), Pack).build())
                .out(Shipments, Publish)
                .build();
            b.include(
                subscriber(tracking(), Track)
                    .workers_by_key(nonzero!(8usize))
                    .build(),
            );
            b.include(subscriber(ledger(), Book).build());
            b.include(subscriber(invoicing(), Invoice).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

OperationRetries::attempts ограничивает, сколько раз клиент повторяет операцию, которую сервер не принял. С тремя попытками второй консьюмер ledger останавливает сервис при запуске с ошибкой, которая называет топик. Ограничение действует на каждую операцию соединения, включая переподключение после перезапуска брокера, поэтому сервис получает раннюю ошибку вместо того, чтобы переждать сбой.

Партиционированные топики

Партиционированный топик - это несколько топиков под одним именем, и каждая партиция обслуживается отдельно. Администратор создаёт его с заданным числом партиций:

docker compose exec pulsar bin/pulsar-admin topics create-partitioned-topic \
  persistent://public/default/shipments -p 3

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

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

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