Очереди и журналы — Примеры

Пример 1 — Очередь для списания, журнал для события

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

нагрузка A: списать с карты за заказ 8842, один раз.
нагрузка B: события order.placed, на которые реагируют биллинг, аналитика и комплектация.

Решение

A -> очередь
  [списать за заказ 8842] -> один потребитель -> подтверждено -> исчезла
  второму потребителю перечитывать задачу не нужно, иначе будет двойное списание.

B -> журнал, с хранением
  [order.placed 8841][order.placed 8842][order.placed 8843] ...
  биллинг читает с позиции 120, аналитика с 98, комплектация с 141
  четвёртая система в следующем месяце начнёт с 0 и воспроизведёт пропущенное.

Определяющее свойство — нужно ли кому-нибудь прочитать элемент снова: подтверждение завершает жизнь сообщения очереди, а журнал хранит запись и даёт каждому потребителю позицию. Задача адресована тому, кто свободен; событие же — факт, который несколько читателей потребляют каждый в своём темпе. Однако обычно ошибаются, выбирая очередь из-за более простой мысленной модели, и позже обнаруживают, что история, нужная новому потребителю, была подтверждена и исчезла.

Пример 2 — Одно и то же событие доставлено дважды

Обработчик order.placed переводит заказ в оплаченные и отправляет квитанцию. Брокер доставляет заново всякий раз, когда подтверждение потеряно, — поэтому одно и то же событие может прийти дважды, и уйдут две квитанции. Сделайте вторую обработку безвредной: запишите идентификатор сообщения до применения эффекта.

async function onOrderPlaced(msg) {
  await db.query("update orders set status = 'paid' where id = $1", [msg.orderId]);
  await mailer.send(msg.customerEmail, "receipt");
}
// потерянное подтверждение приносит то же сообщение снова: две квитанции.

Решение

async function onOrderPlaced(msg) {
  const first = await db.query(
    "insert into processed_messages (message_id) values ($1) on conflict do nothing",
    [msg.id]
  );
  if (first.rowCount === 0) return;         // это событие уже обработано

  await db.query("update orders set status = 'paid' where id = $1", [msg.orderId]);
  await mailer.send(msg.customerEmail, "receipt");
}
// позиция потребителя фиксируется после работы, поэтому сбой воспроизводит
// сообщение, а не теряет его.

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