Что такое Consumer group в Kafka?

Consumer group в Kafka

Consumer group — это фундаментальная концепция Apache Kafka, которая позволяет масштабировать обработку сообщений и обеспечивает гарантии доставки. Это группа потребителей (consumers), которые вместе читают из одной или нескольких тем (topics).

Основная идея

Когда несколько потребителей объединены в одну группу, Kafka автоматически распределяет партиции (partitions) между ними так, чтобы каждая партиция читалась только одним потребителем в группе. Это обеспечивает параллельную обработку без дублирования.

from kafka import KafkaConsumer
import json

# Создаем потребителя, входящего в группу 'my-group'
consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',  # Consumer group ID
    auto_offset_reset='earliest',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    print(f'Offset: {message.offset}, Value: {message.value}')

Как работает распределение партиций

Представим, что:

• Тема имеет 4 партиции (0, 1, 2, 3)
• Consumer group состоит из 2 потребителей

Распределение:

• Consumer 1: читает партиции 0 и 1
• Consumer 2: читает партиции 2 и 3

Если добавить третьего потребителя:

• Consumer 1: партиция 0
• Consumer 2: партиция 1
• Consumer 3: партиции 2 и 3

Этот процесс называется rebalancing (перебалансировка).

Ключевые характеристики

1. Параллелизм

Несколько потребителей в одной группе обрабатывают разные партиции одновременно. Это обеспечивает горизонтальное масштабирование и позволяет быстрее обрабатывать большие объемы данных.

2. Гарантия, что сообщение обработано ровно одним потребителем

• Каждое сообщение из партиции будет обработано только одним потребителем в группе
• Это предотвращает дублирование обработки

3. Управление offset

# Kafka отслеживает, какие сообщения уже обработаны (offset)
consumer = KafkaConsumer(
    'my-topic',
    group_id='my-group',
    bootstrap_servers=['localhost:9092'],
    auto_commit_interval_ms=5000,  # Автоматически коммитить offset каждые 5 сек
    enable_auto_commit=True
)

# Или явный коммит
for message in consumer:
    process(message)
    consumer.commit()  # Сохраняем offset вручную

Rebalancing (Переbalансировка)

Что происходит при добавлении/удалении потребителя:

1. Все потребители в группе останавливают обработку
2. Kafka переераспределяет партиции
3. Потребители возобновляют обработку

Это может привести к brief downtime. Длительность зависит от количества сообщений и rebal time.

Практический пример: обработка заказов

from kafka import KafkaConsumer
import json

def process_order(order):
    print(f"Processing order {order['id']}")
    return True

consumer = KafkaConsumer(
    'orders',
    group_id='order-processors',
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=False,
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    try:
        order = message.value
        if process_order(order):
            consumer.commit()
    except Exception as e:
        print(f'Error: {e}')

Различия между consumer groups

Если у вас две разные группы читают из одной темы, обе группы получат ВСЕ сообщения из этой темы. Каждая группа отслеживает свой offset независимо.

Важные моменты

1. Количество партиций = максимальный параллелизм

   • Если партиций 4, то максимум 4 потребителя в группе смогут работать параллельно
   • Добавление 5-го потребителя не даст результата

2. Offset хранится в Kafka

   • Обычно в специальной теме __consumer_offsets
   • При перезагрузке потребителя он может продолжить с того же места

3. Сессии и heartbeat

   • Если потребитель не отправит heartbeat, его признают мертвым
   • Это инициирует rebalancing

Consumer group — это мощный механизм, который делает Kafka масштабируемой и надежной системой для распределенной обработки событий.