Сколько было потребителей сообщений в RabbitMQ в одну очередь?

Опыт работы с потребителями RabbitMQ

На конкретных проектах работал с очередями RabbitMQ, где было от 2 до 8 потребителей на одну очередь в зависимости от нагрузки и требований.

Архитектура с несколькими потребителями

Типичный сценарий: задачи обработки заказов, отправка email, обработка платежей. Для каждого типа задач была отдельная очередь с несколькими потребителями для параллельной обработки.

const amqp = require('amqplib');

const QUEUE_NAME = 'order_processing';
const CONSUMER_COUNT = 4; // 4 потребителя на очередь

async function setupConsumers() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  await channel.assertQueue(QUEUE_NAME, { durable: true });
  
  // Каждый потребитель обрабатывает одно сообщение за раз
  await channel.prefetch(1);
  
  // Запускаем несколько инстансов приложения/потребителей
  for (let i = 0; i < CONSUMER_COUNT; i++) {
    channel.consume(QUEUE_NAME, async (msg) => {
      if (msg) {
        try {
          const data = JSON.parse(msg.content.toString());
          console.log(`Consumer ${i}: Processing order ${data.orderId}`);
          
          // Бизнес-логика
          await processOrder(data);
          
          // Подтверждаем обработку
          channel.ack(msg);
        } catch (error) {
          console.error(`Consumer ${i}: Error`, error);
          // Отправляем обратно в очередь при ошибке
          channel.nack(msg, false, true);
        }
      }
    });
  }
}

Правильный способ масштабирования

Важный момент: в RabbitMQ один потребитель может быть представлен несколькими инстансами одного приложения, а не разными потребителями в одном процессе:

# Запускаем несколько инстансов одного сервиса
npm start  # Инстанс 1 подключается к очереди
npm start  # Инстанс 2 подключается к очереди
npm start  # Инстанс 3 подключается к очереди
npm start  # Инстанс 4 подключается к очереди

# RabbitMQ распределяет сообщения между инстансами

К одной очереди подключается несколько инстансов сервиса, а не один инстанс с несколькими потребителями. Это автоматически обеспечивает балансировку нагрузки.

Конфигурация Consumer Group

В production используется паттерн Consumer Group:

async function startOrderConsumer() {
  const connection = await amqp.connect(process.env.RABBITMQ_URL);
  const channel = await connection.createChannel();
  
  const queue = 'orders.new';
  const exchange = 'orders';
  
  // Объявляем очередь как durable (пережидает перезагрузку)
  await channel.assertQueue(queue, {
    durable: true,
    arguments: {
      'x-max-length': 10000,  // Макс размер очереди
      'x-dead-letter-exchange': 'dead_letters'
    }
  });
  
  // prefetch(1) — важно!
  // Каждый потребитель обрабатывает по одному сообщению
  await channel.prefetch(1);
  
  console.log(`[*] Waiting for messages in ${queue}`);
  
  channel.consume(queue, async (msg) => {
    if (msg) {
      const order = JSON.parse(msg.content.toString());
      
      try {
        await processOrder(order);
        channel.ack(msg);
        console.log(`[✓] Order ${order.id} processed`);
      } catch (error) {
        console.error(`[✗] Error processing order ${order.id}`);
        // NACK с requeue — сообщение вернётся в очередь
        channel.nack(msg, false, true);
      }
    }
  });
}

Мониторинг и масштабирование

В production мониторим очередь:

async function getQueueStats() {
  const connection = await amqp.connect(process.env.RABBITMQ_URL);
  const channel = await connection.createChannel();
  
  const queueStats = await channel.checkQueue('orders.new');
  
  console.log(`Messages in queue: ${queueStats.messageCount}`);
  console.log(`Active consumers: ${queueStats.consumerCount}`);
  
  // Если очередь растёт быстрее чем обрабатывается — запускаем больше инстансов
  if (queueStats.messageCount > queueStats.consumerCount * 100) {
    console.warn('Queue is growing! Scale up consumers');
  }
}

Когда запускаю новый инстанс сервиса, он автоматически подключается к очереди и начинает получать сообщения вместе с другими инстансами. RabbitMQ распределяет сообщения fair-dispatch по всем подключённым потребителям.

На практике в моих проектах

Малые очереди (очередь обработки не критична):

• 1-2 потребителя (инстанса)
• Простое, легко отлаживать

Средние очереди (например, email, уведомления):

• 3-4 инстанса
• Хороший баланс между пропускной способностью и ресурсами

Критичные очереди (платежи, заказы):

• 5-8 инстансов + ещё несколько на backup
• Высокая доступность
• Dead Letter Exchange для ошибок

Важные практики

1. Всегда используй prefetch(1) — не нагружай потребителей сразу
2. Durable очереди — сообщения пережидают перезагрузку
3. Acknowledgements — подтверждай обработку
4. Dead Letter Queue — для ошибок
5. TTL на сообщения — если нужно
6. Мониторинг — следи за размером очереди

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