Могут ли несколько консьюмер читать одну очередь в RabbitMQ

Могут ли несколько консьюмер читать одну очередь в RabbitMQ?

Прямой ответ: ДА

Несколько потребителей (consumer) могут читать из одной очереди в RabbitMQ. Это один из основных паттернов использования RabbitMQ для масштабирования обработки сообщений.

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

Когда несколько консьюмеров подписаны на одну очередь, RabbitMQ распределяет сообщения между ними. Каждое сообщение обрабатывает ровно один консьюмер.

import pika
import time

# КОНСЬЮМЕР 1
def consumer_1():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    channel = connection.channel()
    
    # Подписываемся на одну очередь
    channel.queue_declare(queue='tasks', durable=True)
    
    def callback(ch, method, properties, body):
        print(f"Consumer 1 получил: {body.decode()}")
        time.sleep(1)  # Обработка
        ch.basic_ack(delivery_tag=method.delivery_tag)
    
    channel.basic_consume(queue='tasks', on_message_callback=callback)
    print('Consumer 1 ждет сообщений...')
    channel.start_consuming()

# КОНСЬЮМЕР 2 (в отдельном процессе/потоке)
def consumer_2():
    connection = pika.BlockingConnection(
        pika.ConnectionParameters('localhost')
    )
    channel = connection.channel()
    
    channel.queue_declare(queue='tasks', durable=True)
    
    def callback(ch, method, properties, body):
        print(f"Consumer 2 получил: {body.decode()}")
        time.sleep(1)  # Обработка
        ch.basic_ack(delivery_tag=method.delivery_tag)
    
    channel.basic_consume(queue='tasks', on_message_callback=callback)
    print('Consumer 2 ждет сообщений...')
    channel.start_consuming()

# Запустить обоих консьюмеров параллельно
# python consumer.py (запустить дважды)

Визуально как это работает

Производитель (Publisher)
    |
    v
┌─────────────────────┐
│   RabbitMQ Queue    │
│   tasks             │
│ [msg1][msg2][msg3]  │
└─────────────────────┘
   /          |
  /           |
v            v
Consumer 1  Consumer 2
получает    получает
msg1        msg2

Этот процесс повторяется для msg3, msg4...

Важный параметр: prefetch_count (QoS)

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

channel.queue_declare(queue='tasks', durable=True)

# ВАЖНО: Устанавливаешь prefetch_count
# Это сколько сообщений консьюмер получает одновременно
channel.basic_qos(prefetch_count=1)  # Получать по 1 сообщению

def callback(ch, method, properties, body):
    print(f"Обрабатываю: {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)  # Подтверждаю обработку

channel.basic_consume(queue='tasks', on_message_callback=callback)
channel.start_consuming()

# prefetch_count=1: Consumer получает новое сообщение
#                   только после подтверждения (ack) старого
# prefetch_count=0: Неограниченное предварительное получение (ПЛОХО!)

Сценарий работы с несколькими консьюмерами

# ПРОДЮСЕР: отправляет 10 сообщений
import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

channel.queue_declare(queue='tasks', durable=True)

for i in range(10):
    channel.basic_publish(
        exchange='',
        routing_key='tasks',
        body=f'Task {i}',
        properties=pika.BasicProperties(delivery_mode=2)  # persistent
    )
    print(f'Отправлено Task {i}')

connection.close()

# Если запустить 2 консьюмера:
# Consumer 1 получит: Task 0, Task 2, Task 4, Task 6, Task 8
# Consumer 2 получит: Task 1, Task 3, Task 5, Task 7, Task 9

Важные концепции

1. Message Acknowledgment (подтверждение)

# AUTO_ACK (опасно)
channel.basic_consume(queue='tasks', auto_ack=True)
# Сообщение удаляется сразу, даже если обработка упала!

# MANUAL_ACK (правильно)
channel.basic_consume(queue='tasks', auto_ack=False)
def callback(ch, method, properties, body):
    try:
        # Обработка
        process(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)  # Успех
    except Exception as e:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)  # Вернуть в очередь

2. Fair dispatch (справедливое распределение)

# Без fair dispatch (prefetch_count=0)
channel.basic_qos(prefetch_count=0)
# Consumer 1 получает все сообщения если он быстрее

# С fair dispatch (prefetch_count=1)
channel.basic_qos(prefetch_count=1)
# Сообщения распределяются равномерно

Пример с обработкой ошибок

import pika
import time

class TaskConsumer:
    def __init__(self, consumer_name):
        self.consumer_name = consumer_name
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost')
        )
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='tasks', durable=True)
        self.channel.basic_qos(prefetch_count=1)
    
    def process_message(self, ch, method, properties, body):
        try:
            message = body.decode()
            print(f"[{self.consumer_name}] Обрабатываю: {message}")
            
            # Имитация обработки
            time.sleep(2)
            
            # Подтверждаем обработку
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"[{self.consumer_name}] Готово")
            
        except Exception as e:
            print(f"[{self.consumer_name}] Ошибка: {e}")
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    
    def start(self):
        self.channel.basic_consume(
            queue='tasks',
            on_message_callback=self.process_message
        )
        print(f"[{self.consumer_name}] Ждет сообщений...")
        self.channel.start_consuming()

# Запустить
if __name__ == '__main__':
    consumer = TaskConsumer('Consumer-1')
    consumer.start()

Преимущества

• Масштабируемость - добавляешь консьюмеров, система быстрее обрабатывает сообщения
• Отказоустойчивость - если один консьюмер упал, другие продолжают работать
• Распределение нагрузки - сообщения распределяются между консьюмерами

Вывод

Да, несколько консьюмеров могут читать одну очередь в RabbitMQ. Каждое сообщение обрабатывает ровно один консьюмер. Важно использовать prefetch_count для справедливого распределения и manual ack для надежной обработки.