Что такое RabbitMQ Streams?

Что такое RabbitMQ Streams?

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

Как это работает?

Streams основаны на концепции append-only лога. Каждое новое сообщение добавляется в конец лога. Потребители могут начать чтение с начала, с конца или с определённого смещения (offset). Это напоминает Apache Kafka, но встроено прямо в RabbitMQ.

Пример использования

Допустим, вы хотите логировать действия пользователей и затем анализировать их. С помощью Streams вы можете записывать события, а несколько сервисов (аналитика, аудит, мониторинг) могут читать их независимо.

// Пример на Node.js с amqplib
const amqp = require('amqplib');

async function producer() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createConfirmChannel();
  // Создаём stream
  await channel.assertQueue('user_actions', { arguments: { 'x-queue-type': 'stream' } });
  // Отправляем сообщение
  channel.sendToQueue('user_actions', Buffer.from('User logged in'), {}, (err, ok) => {
    if (err) console.error(err);
    else console.log('Message sent');
  });
}

async function consumer() {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  await channel.assertQueue('user_actions', { arguments: { 'x-queue-type': 'stream' } });
  // Читаем с начала
  channel.consume('user_actions', (msg) => {
    console.log('Received:', msg.content.toString());
    channel.ack(msg);
  }, { arguments: { 'x-stream-offset': 'first' } });
}

Когда применять?

RabbitMQ Streams полезны, когда нужно хранить историю сообщений для повторного чтения, например, для аудита, аналитики или восстановления состояния. Они также подходят для сценариев, где несколько потребителей должны обрабатывать одни и те же данные независимо. Однако для высоконагруженного стриминга с большим количеством партиций лучше рассмотреть Kafka.

Вывод: RabbitMQ Streams — это удобное расширение для сценариев, где требуется долговременное хранение сообщений и независимое чтение несколькими потребителями, особенно если вы уже используете RabbitMQ и не хотите внедрять отдельный брокер.