Зачем ты везде тащишь очереди?

Вот ты проектируешь свою систему, базы, микросервисы, кэш - все красиво. И вот встает вопрос: как это все связать, как они будет общаться?

И тут со всех утюгов кричат:

  • _У тебя микросервисы - тебе нужны очереди!_
  • _Очереди - это модно._
  • _Очереди - это полезно._
  • _Очереди - это масштаб_ и надежность.
  • _Очереди "корпоративный стандарт"._ (как и TS, ага да)

Ты берешь, допустим RabitMQ, поднимаешь брокер (+1 сервис и глобальная точка отказа), и получаешь большую асинхронную боль сквозь всю систему, рост когнитивной нагрузки и эксплуатационные издержки: лаг, дубликаты, краш консумера, ядовитые сообщения и дрейф схем. Да есть и полезное: буферизация, распределение по времени, изоляция сбоев и фан-аут. Но какой ценой?

Если у тебя CRUD между двумя сервисами и одинаковые требования по времени ответа и доступности - очередь только навредит.

Синхронный вызов честно говорит: «_либо сделал, либо нет_». С очередью ты говоришь: «_когда-нибудь, возможно, кто-то где-то это обработает_».

Отсюда вылезают взрослые темы, которые почему-то игнорируют, когда «прикручивают Kafka/Rabbit ради моды»:

  • Доставка: at-least-once - дубликаты - идемпотентность на приёмнике
  • Порядок: забудь «строгий порядок», у тебя head-of-line blocking или вообще все перемешано.
  • Нагрузка: «очередь поможет удержать пик нагрузки» — да, но хвост растёт. Хвост — это твой скрытый долг по времени отклика.
  • Мониторинг: трассировка через брокер — это чёрный ящик и пляски с кореляторами.
  • Контракты: cобытие — это публичное API. Версионирование, эволюция схем, миграции потребителей — отдельная боль.

Очередь приносит тебе асинхронность как обязательство. А это обязательство принуждает тебя проектировать идемпотентные операции, хранить входящие/исходящие события, жанглировать ретраями и дедупликацией, думать про повторный прогон истории (пересчёт из журнала). Если ты к этому не готов - будет больно, очередь сломает прод, или тебя.

Пример

Сценарий: Создать задачу и отправить ее на исполнение в другой сервис.

Вариант А — MongoDB + outbox (без внешней очереди):

Идея простая: Одной транзакцией пишем задачу и событие. Сервис исполнитель читает события напрямую, делает работу, фиксирует факт обработки. Никакой магии, никаких скрытых механизов доставки, честная атомарность.

// Сервис "Планирователь"
/**
 * @param {import('mongodb').MongoClient} client
 * @param {import('mongodb').Db} db
 * @param {Task} task
 * @returns {{status: boolean, taskId: string}}
 */
export async function createTask (client, db, task) {
  const session = client.startSession()

  try {
    const eventId = crypto.randomUUID()

    session.startTransaction()

    await db.collection('tasks').insertOne(
      {
        _id: task.id,
        type: task.type,
        payload: task.payload,
        status: 'pending',
        createdAt: Date.now()
      },
      { session }
    )

    await db.collection('outbox').insertOne(
      {
        _id: eventId,
        type: 'task.created',
        payload: { taskId: task.id },
        createdAt: Date.now(),
        pickedAt: null,
        doneAt: null
      },
      { session }
    )

    await session.commitTransaction()

    return { status: true, eventId }
  } catch (error) {
    console.error(`[createTask] ${task.id}:`, error)
    await session.abortTransaction()
  } finally {
    await session.endSession()
  }
}

// Сервис "Делатель"
/**
 * @param {import('mongodb').MongoClient} client
 * @param {import('mongodb').Db} db
 * @returns {Promise<void>}
 */
export async function executor (client, db) {
  // атомарно «забронировали» событие
  const event = await db
    .collection('outbox')
    .findOneAndUpdate(
      { type: 'task.created', pickedAt: null },
      { $set: { pickedAt: Date.now() } },
      { sort: { createdAt: 1 }, returnDocument: 'after' }
    )

  if (!event) {
    //нет событий
    delay(100)
    return
  }

  // идемпотентность на уровне обработчика
  const processedId = `${event._id}:executor`
  const processed = await db.collection('processed').findOne({ _id: processedId })

  if (!processed) {
    // уже делали — пропускаем
    return
  }

  const task = await db.collection('tasks').findOne({ _id: event.payload.taskId, status: { $ne: 'done' } })

  if (!task) {
   // таск уже кто-то сделал
    return
  }

  if (task) {
    try {
      // делаем таску, помним про идемпотентность
      await doWork(task)
      await db.collection('processed').insertOne({ _id: processedId, at: Date.now() })
      await db.collection('tasks').updateOne({ _id: task._id }, { $set: { status: 'done' } })
      await db.collection('outbox').updateOne({ _id: event._id }, { $set: { doneAt: Date.now() } })
    } catch (error) {
      // ретраи с backoff, лимит → DLQ
      console.error(`[executor] ${task.id}:`, error)
      await db.collection('outbox').updateOne({ _id: event._id }, { $set: { pickedAt: null } })
      await db.collection('processed').deleteOne({ _id: processedId })
      await delay(100)
    }
  }
}
// executor запускается setInterval или setTimeout выше по архитектуре

В примере важное: Задача и событие родились в одном коммите — значит, не потеряются. Исполнитель читает из той же БД — меньше инфраструктуры, меньше точек отказа. Цена — ты сам хозяин ретраев и DLQ, но это как раз то, что нужно контролировать.

Уже слышу крики: "А как же изоляция, границы, связность, какой "ужос" два сервиса работают с одной бд. Как же Чистая архитектура?!" Нормально, отвечу я, когда два сервиса читают одну БД: один владеет записью и инвариантами, остальные читают только стабильную проекцию/реплику по контракту — вот и вся изоляция. Ну это мы отвлеклись. Продолжим.

Вариант Б — RabbitMQ (внешняя очередь)

Наивно делать «insert в БД и затем publish в Rabbit» нельзя: между этими шагами всегда есть окна потерь/дублей. Значит… сюрприз… всё равно нужен outbox. Просто вместо прямого чтения его, будет «ретранслятор в Rabbit».

// Сервис "Планирователь"
export async function createTask (client, db, task) {
//createTask такой же как и в Варианте А, только меняем немного запись
//...
  await db.collection('outbox').insertOne(
    {
      _id: eventId,
      type: 'task.created',
      payload: { taskId: task.id },
      createdAt: Date.now(),
      pickedAt: null,
      publishedAt: null
    },
    { session }
  )
//...
}

// но тут нам нужен еще relay - штука по доставке события из бд в очередь

/**
 * @param {import('mongodb').Db} db
 * @param {import('amqplib').Channel} rmqChannel
 * @param {string} exchange
 * @returns {Promise<void>}
 */
export async function relay (db, rmqChannel, exchange) {
  const event = await db
    .collection('outbox')
    .findOneAndUpdate(
      { type: 'task.created', pickedAt: null },
      { $set: { pickedAt: Date.now() } },
      { sort: { createdAt: 1 }, returnDocument: 'after' }
    )

  if (!event) {
    await delay(50)
    return
  }

  try {
    await publishConfirm(rmqChannel, exchange, 'task.created', Buffer.from(JSON.stringify(event)))
    await db.collection('outbox').updateOne({ _id: event._id }, { $set: { publishedAt: Date.now() } })
  } catch (e) {
    await db.collection('outbox').updateOne({ _id: event._id }, { $set: { pickedAt: null } })
    await sleep(100)
  }
}

// Сервис "Делатель"
/**
 * @param {import('mongodb').Db} db
 * @param {Object} msg
 */
export async function consumeAndExecute (db, msg) {
  const event = JSON.parse(msg.content.toString())

  // идемпотентность на уровне обработчика
  const processedId = `${event._id}:executor`
  const processed = await db.collection('processed').findOne({ _id: processedId })

  if (!processed) {
    //дубль, уже исполнили
    return 'ack'
  }

  const task = await db.collection('tasks').findOne({ _id: event.payload.taskId, status: { $ne: 'done' } })

  if (!task) {
    return 'ack'
  }

  if (task) {
    try {
      // делаем таску, помним про идемпотентность
      await doWork(task)
      await db.collection('processed').insertOne({ _id: processedId, at: Date.now() })
      await db.collection('tasks').updateOne({ _id: task._id }, { $set: { status: 'done' } })
      return 'ack'
    } catch (error) {
      console.error(`[consumeAndExecute] ${task.id}:`, error)
      // ретраи с backoff, DLX/TTL, лимит → DLQ, можно ответить 'nack' получим еще раз
      await db.collection('processed').deleteOne({ _id: processedId })
      return 'ack'
    }
  }
}

А где минусы спросишь ты. Так вот же, в полтора раза больше кода, больше мест, где, что-то может пойти не так. Дополнительный сервис с брокером. Транспорт между брокером и сервисами. Если не делаешь relay — потеряешь события при падениях между БД и брокером. Если не делаешь inbox и идемпотентность — словишь дубль при redelivery/повторной публикации. Rabbit сам по себе не решает твою бизнес-консистентность, он только перевозит байты.

Внутрисистемный асинхрон: Mongo + outbox закрывает 90% кейсов без зоопарка, и даже фан-аут тоже можно сделать. Но если хочешь фан-аут на внешних потребителей из коробки, независимые группы, интеграции или просто очень хочется очередь — бери Rabbit (или другой брокер), но не забывай: outbox/inbox, idempotency, confirm-публикации, DLQ. Очередь без этих штук — просто дорогая труба с иллюзией надёжности.

Что в итоге?

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

Нужен буфер? Сделай backpressure и честные 429. Нужна обработка «не в запросе»? Outbox и воркер. Нужен событийный продукт между доменами? И тут все еще можно обойтись без брокера, но уже можно задуматься про него, выбрать, закатать рукава и решиться нырнуть в асинхронный персональный ад.

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

Хочешь поспорить — покажи, где у тебя backpressure, где идемпотентность и как ты репроцессишь хвост. Если ответ «никак» — очередь тебе не помогла, она просто спрятала проблему в тень.