В чем разница между топик и очередь?

Различия между топиком и очередью

Топик и очередь — это два основных паттерна обмена сообщениями в системах, работающих с брокерами сообщений (Kafka, RabbitMQ, NATS). Они предназначены для разных сценариев и имеют принципиально разное поведение.

1. Очередь (Queue)

Определение: Очередь — это способ доставки сообщений "один-в-один" (one-to-one). Каждое сообщение обрабатывается ровно одним консьюмером.

import pika

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

# Объявляем очередь
channel.queue_declare(queue='task_queue', durable=True)

# Producer отправляет сообщение в очередь
def send_message(message):
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"Отправлено: {message}")

# Consumer получает сообщение
def receive_message():
    def callback(ch, method, properties, body):
        print(f"Получено: {body.decode()}")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(queue='task_queue', on_message_callback=callback)
    channel.start_consuming()

send_message('Задача 1')
receive_message()

Характеристики:

• Один-в-один (point-to-point)
• Сообщение удаляется после обработки
• Может быть несколько консьюмеров, но каждое сообщение обрабатывается одним
• Гарантирует, что сообщение будет обработано
• Использует load balancing между консьюмерами

2. Топик (Topic)

Определение: Топик — это способ доставки сообщений "один-ко-многим" (one-to-many). Одно сообщение отправляется всем подписчикам.

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer отправляет сообщение в топик
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def send_event(topic, event):
    producer.send(topic, event)
    print(f"Событие отправлено в {topic}: {event}")

send_event('user_events', {'type': 'signup', 'user_id': 123})

# Несколько консьюмеров получают одно сообщение
def subscribe_to_events(topic, consumer_group):
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=['localhost:9092'],
        group_id=consumer_group,
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    
    for message in consumer:
        print(f"[{consumer_group}] Получено событие: {message.value}")

# Запускаем несколько консьюмеров
subscribe_to_events('user_events', 'analytics_group')
subscribe_to_events('user_events', 'notification_group')
subscribe_to_events('user_events', 'logging_group')

Характеристики:

• Один-ко-многим (publish-subscribe)
• Сообщение доставляется всем активным подписчикам
• Каждый подписчик получает копию сообщения
• Подписчики работают независимо
• Идеален для распределения событий

3. Сравнение

4. Практические примеры

Пример 1: Очередь для обработки заказов

from celery import Celery

app = Celery('tasks', broker='redis://localhost')

@app.task
def process_order(order_id):
    """Задача обработки заказа в очереди"""
    print(f"Обработка заказа: {order_id}")
    # ... логика обработки
    return f"Заказ {order_id} обработан"

# Отправка задачи в очередь
process_order.delay(12345)

# Несколько воркеров будут обрабатывать задачи
# celery -A tasks worker --loglevel=info

Пример 2: Топик для распространения событий

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer отправляет событие
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

def on_user_created(user_id, email):
    event = {'user_id': user_id, 'email': email, 'timestamp': str(datetime.now())}
    producer.send('user_created', json.dumps(event).encode('utf-8'))
    print(f"Событие user_created отправлено: {user_id}")

# Консьюмер 1: отправить приветственное письмо
group_email = KafkaConsumer('user_created', group_id='email_service')
for msg in group_email:
    event = json.loads(msg.value)
    print(f"Email сервис: отправить письмо {event['email']}")

# Консьюмер 2: добавить в аналитику
group_analytics = KafkaConsumer('user_created', group_id='analytics')
for msg in group_analytics:
    event = json.loads(msg.value)
    print(f"Analytics: записать нового пользователя {event['user_id']}")

# Консьюмер 3: запустить рекомендационный движок
group_ml = KafkaConsumer('user_created', group_id='ml_pipeline')
for msg in group_ml:
    event = json.loads(msg.value)
    print(f"ML Pipeline: инициализировать профиль {event['user_id']}")

5. Когда использовать очередь

Используй очередь когда:

• Задача должна быть обработана одной системой
• Нужна балансировка нагрузки между несколькими воркерами
• Важна гарантия обработки каждого сообщения
• Это долгоживущие задачи (отправка email, обработка данных)
• Нужна задержка между обработкой сообщений

Примеры:

• Обработка заказов
• Отправка email и SMS
• Генерация отчетов
• Обработка изображений
• Резервное копирование

6. Когда использовать топик

Используй топик когда:

• Много систем должны реагировать на одно событие
• Нужна распространение события всем заинтересованным сервисам
• Системы работают асинхронно и независимо
• Нужна история событий для новых подписчиков
• Это информационные события (не критические задачи)

Примеры:

• Уведомления о создании пользователя
• События изменения статуса заказа
• Метрики системы
• Логи приложения
• Аналитические события

7. Гибридный подход

# Топик для событий
# -> Консьюмер 1: Analytics (группа 'analytics')
# -> Консьюмер 2: Email service (группа 'email')
# -> Консьюмер 3: Logging service (группа 'logging')

# Каждый консьюмер может затем отправить задачу в очередь
from kafka import KafkaConsumer
from celery import current_app
import json

consumer = KafkaConsumer(
    'user_events',
    group_id='email_service',
    bootstrap_servers=['localhost:9092']
)

for message in consumer:
    event = json.loads(message.value)
    if event['type'] == 'signup':
        # Отправляем в очередь для обработки
        send_welcome_email.delay(event['user_id'])

8. RabbitMQ: Очередь с паттернами

import pika

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

# Очередь (point-to-point)
channel.queue_declare(queue='tasks', durable=True)

# Топик через exchange (publish-subscribe)
channel.exchange_declare(exchange='events', exchange_type='fanout')

# Создаём очередь и привязываем к топику
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='events', queue=queue_name)

# Producer отправляет в топик
channel.basic_publish(exchange='events', routing_key='', body='Событие!')

# Consumer получает из топика
def callback(ch, method, properties, body):
    print(f"Получено: {body}")

channel.basic_consume(queue=queue_name, on_message_callback=callback)
channel.start_consuming()

9. Выбор между Kafka, RabbitMQ, Redis

Kafka (топик):

• Высокая пропускная способность
• Сохранение истории сообщений
• Отлично для больших объёмов
• Для событийных систем

RabbitMQ (очередь/топик):

• Гибкая маршрутизация
• Поддержка разных паттернов
• Гарантирует доставку
• Для классических задач

Redis (очередь):

• Простота
• Высокая скорость
• Для простых задач
• В памяти (потеря при перезагрузке)

10. Вывод

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