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

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

Ключи упорядочивания

Это руководство сохраняет порядок сообщений одного ключа: при получении, при параллельной обработке внутри сервиса и при публикации.

По умолчанию Pub/Sub доставляет сообщения в произвольном порядке. Сообщение, опубликованное с ключом упорядочивания в подписку с упорядочиванием сообщений, приходит после всех более ранних сообщений того же ключа. Сообщения разных ключей по-прежнему приходят в любом порядке, и поэтому их можно обрабатывать параллельно. Гарантия действует для сообщений, опубликованных в одном регионе.

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

Упорядоченная доставка на партициях воркеров​

src/main.rs
/// The entries of one account reach one lane in publish order; eight lanes run side by side.
#[subscriber(
    GooglePubSub::new("ledger-posting").create_with_topic("ledger"),
    workers(8, by_key)
)]
async fn post(
    entry: &Entry,
    Out(out): Out<impl Publisher<Options = PubSubPublishOptions>>,
) -> HandlerOutcome {
    let posted = Posted {
        account: entry.account.clone(),
        amount_cents: entry.amount_cents,
    };
    let sent = out
        .message(&posted)
        .to("postings")
        .ordering_key(entry.account.clone())
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(entry.account, error = %err, "posting not published, entry retried");
            HandlerOutcome::retry()
        }
    }
}

create_with_topic создаёт ledger-posting с упорядочиванием сообщений. Pub/Sub фиксирует эту настройку при создании подписки, поэтому у подписки, которой владеет эксплуатация, она есть, только если её включила эксплуатация.

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

При повторной доставке сообщения одного ключа Pub/Sub снова доставляет и более поздние сообщения этого ключа, поэтому обработчик упорядоченной подписки должен быть идемпотентным.

Ключ на том, что публикует сервис​

post публикует проведённую проводку в postings под ключом счёта, который называет в вызове шаг ordering_key. Шаг приходит с прелюдией. Чтобы он был доступен, параметр Out называет тип настроек крейта: Publisher<Options = PubSubPublishOptions>. Публикация без ключа не упорядочивается.

У ответа нет вызова, в котором можно назвать ключ. Ключ для каждой доставки задаёт преобразование публикации:

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "statements")]
struct Statement {
    account: String,
    amount_cents: i64,
}

/// Answers each entry with a statement line for the account's owner.
#[subscriber(GooglePubSub::new("ledger-statements").create_with_topic("ledger"), reply)]
async fn notify(entry: &Entry) -> Statement {
    Statement {
        account: entry.account.clone(),
        amount_cents: entry.amount_cents,
    }
}

/// Sends each reply under the key of the entry it answers, which the delivery reports in its
/// `partition-key` header.
struct UnderTheEntrysKey;

impl<C> PublishTransform<ForReply<C>, PubSubPublishOptions> for UnderTheEntrysKey {
    type Destination = Reads;

    fn apply(
        &self,
        _out: &mut Outgoing<'_>,
        options: &mut Option<PubSubPublishOptions>,
        cx: &PublishContext<'_, C>,
    ) {
        if let Some(key) = cx.headers().get_str(PARTITION_KEY_HEADER) {
            options
                .get_or_insert_with(PubSubPublishOptions::default)
                .ordering_key = Some(key.to_owned());
        }
    }
}

UnderTheEntrysKey читает ключ, с которым пришла проводка, и передаёт его выписке, поэтому выписки одного счёта тоже сохраняют порядок. Ключ, одинаковый для всех сообщений точки монтирования, задаётся вместо этого в политике: Publish::default().ordering_key(..). Если ключ задан в нескольких местах, вызов важнее заголовка partition-key, а заголовок важнее политики.

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            b.include(post).out(DefaultSlot, Publish::default()).build();
            b.include(notify)
                .out_reply(Publish::default())
                .transform(UnderTheEntrysKey);
        },
    )
}

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

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

Тесты публикуют проводку с ключом счёта в заголовке partition-key: так ключ сервиса счетов доходит до доставки. Ключ проведённой проводки они проверяют через tb.out::<DefaultSlot>().with_options(..), а ключ выписки - через published(..).with_options(..). Внутрипроцессный Pub/Sub не моделирует упорядоченную доставку. Тест на эмуляторе для неё есть на странице компонента упорядочивания.