Может ли consumer читать несколько партийций?

Может ли Kafka Consumer читать несколько партиций?

Да, абсолютно! Kafka consumer может и обычно читает сообщения из НЕСКОЛЬКИХ партиций одновременно. Это один из главных механизмов масштабирования Kafka.

Базовое понимание

Kafka Topic структура:

Topic: "orders"
├── Partition 0: [msg1, msg2, msg3, ...]
├── Partition 1: [msg4, msg5, msg6, ...]
├── Partition 2: [msg7, msg8, msg9, ...]
└── Partition 3: [msg10, msg11, msg12, ...]

Один consumer может читать из всех 4 партиций одновременно!

Пример 1: Один consumer, несколько партиций

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import java.util.Collections;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Подписываемся на один topic
consumer.subscribe(Collections.singletonList("orders"));

// Если topic имеет 4 партиции, этот consumer будет читать из всех 4!
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

for (ConsumerRecord<String, String> record : records) {
    System.out.printf("Topic: %s, Partition: %d, Offset: %d, Value: %s%n",
        record.topic(), record.partition(), record.offset(), record.value());
    
    // Вывод:
    // Topic: orders, Partition: 0, Offset: 0, Value: Order123
    // Topic: orders, Partition: 1, Offset: 5, Value: Order456
    // Topic: orders, Partition: 2, Offset: 10, Value: Order789
    // Topic: orders, Partition: 3, Offset: 15, Value: Order999
}

Как это работает: Consumer Group

Консцепция Consumer Group:

Topic "orders" (4 партиции):
├── Partition 0
├── Partition 1
├── Partition 2
└── Partition 3

Consumer Group "my-group":
├── Consumer 1 -> читает из Partition 0, 1
├── Consumer 2 -> читает из Partition 2, 3
└── Consumer 3 -> не работает (переизбыток)

Правило распределения:

• Партиции распределяются между consumers в группе
• Каждая партиция читается ТОЛЬКО одним consumer в группе
• Если consumers < партиций -> некоторые читают несколько
• Если consumers > партиций -> некоторые простаивают

Пример 2: Несколько consumers в группе

Сценарий: Topic с 3 партициями, 2 consumer в группе

Topic "payments" (3 партиции):
├── Partition 0 -> Consumer A
├── Partition 1 -> Consumer B
└── Partition 2 -> Consumer A (балансировка)

// Consumer A
KafkaConsumer<String, String> consumerA = new KafkaConsumer<>(props);
consumerA.subscribe(Collections.singletonList("payments"));
// Consumer A будет читать из Partition 0 и Partition 2

// Consumer B
KafkaConsumer<String, String> consumerB = new KafkaConsumer<>(props);
consumerB.subscribe(Collections.singletonList("payments"));
// Consumer B будет читать из Partition 1

Это балансирует нагрузку между consumers.

Пример 3: Явное указание партиций

Вместо subscribe() можно использовать assign() для явного выбора партиций:

import org.apache.kafka.common.TopicPartition;

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// Явно присваиваем партиции
TopicPartition partition0 = new TopicPartition("orders", 0);
TopicPartition partition1 = new TopicPartition("orders", 1);
TopicPartition partition2 = new TopicPartition("orders", 2);

consumer.assign(Arrays.asList(partition0, partition1, partition2));
// Теперь этот consumer читает ИЗ ТРЁХ ПАРТИЦИЙ!

consumer.seek(partition0, 0);  // Начать с offset 0
consumer.seek(partition1, 0);
consumer.seek(partition2, 0);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Partition: %d, Offset: %d, Value: %s%n",
            record.partition(), record.offset(), record.value());
    }
}

Важная деталь: Порядок сообщений

ВНУТРИ партиции порядок гарантирован:

Partition 0: [msg1, msg2, msg3]  <- порядок гарантирован
Partition 1: [msg4, msg5, msg6]  <- порядок гарантирован
Partition 2: [msg7, msg8, msg9]  <- порядок гарантирован

МЕЖДУ партициями порядок НЕ гарантирован:

Consumer читает:
Partition 0: msg1
Partition 2: msg7
Partition 1: msg4
Partition 0: msg2
Partition 1: msg5
...

Может быть любой порядок!

Если нужен глобальный порядок -> используй 1 партицию (но теряешь масштабируемость).

Partition Assignment Strategies

1. RangeAssignor (по умолчанию)

Topic: orders (3 партиции)
Consumers: A, B, C

Распределение:
Consumer A: [Partition 0]
Consumer B: [Partition 1]
Consumer C: [Partition 2]

2. RoundRobinAssignor

Consumer A: [Partition 0, 2]
Consumer B: [Partition 1]
Consumer C: []

3. StickyAssignor

Минимизирует переассоциацию при добавлении/удалении consumers.

props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");

Rebalancing: Что происходит при добавлении consumer

Сценарий: Topic с 3 партициями

Шаг 1: Был 1 consumer

Consumer A: [Partition 0, Partition 1, Partition 2]

Шаг 2: Добавили Consumer B

Отправляется сигнал rebalance

NOВОЕ распределение:
Consumer A: [Partition 0, Partition 1]
Consumer B: [Partition 2]

Шаг 3: Во время rebalance

- Consumer A ОСТАНАВЛИВАЕТСЯ
- Consumer B ОСТАНАВЛИВАЕТСЯ
- Kafka переоценивает распределение партиций
- Оба consumer ВОЗОБНОВЛЯЮТ работу с новым распределением

// Отслеживание rebalance
consumer.subscribe(Collections.singletonList("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Rebalance: потеряли партиции");
        // Сохрани offset
    }
    
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        System.out.println("Rebalance: получили новые партиции");
        // Восстанови offset
    }
});

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

Начальная setup:

Topic "user-events" (10 партиций)
Consumer Group "analytics" (3 consumers)

Распределение:
Consumer 1: Partitions 0, 1, 2, 3
Consumer 2: Partitions 4, 5, 6, 7
Consumer 3: Partitions 8, 9

Если нагрузка растёт, добавляем ещё consumer:

Consumer Group "analytics" (4 consumers)

Новое распределение (более сбалансированное):
Consumer 1: Partitions 0, 1, 2
Consumer 2: Partitions 3, 4, 5
Consumer 3: Partitions 6, 7, 8
Consumer 4: Partitions 9

Каждый consumer читает МЕНЬШЕ партиций -> быстрее обрабатывает.

Вопросы производительности

Вопрос: Может ли один consumer читать одновременно из 100 партиций?

Да, технически может, но:

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

// Увеличить max.poll.records если много партиций
props.put("max.poll.records", 500);  // по умолчанию 500

// Увеличить буффер
props.put("fetch.max.bytes", 52428800);  // 50MB по умолчанию

Типичная ошибка

Неправильный код: Использование нескольких consumer groups

// НЕПРАВИЛЬНО - разные groups!
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(propsWithGroup("group1"));
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(propsWithGroup("group2"));

consumer1.subscribe(Collections.singletonList("orders"));
consumer2.subscribe(Collections.singletonList("orders"));

// Результат: ОБА consumer получат ВСЕ партиции!
// Это дублирование, не параллелизм

Правильный код: Один group

KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(propsWithGroup("my-group"));
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(propsWithGroup("my-group"));

consumer1.subscribe(Collections.singletonList("orders"));
consumer2.subscribe(Collections.singletonList("orders"));

// Результат: партиции распределены между ними

Итоговая таблица

Вывод

Да, Kafka consumer ЧАСТО читает из НЕСКОЛЬКИХ партиций:

1. Consumer группа автоматически распределяет партиции
2. Один consumer может читать из многих партиций
3. Это позволяет масштабировать параллельную обработку
4. Порядок внутри партиции гарантирован, между партициями нет
5. Rebalancing происходит при изменении количества consumers

Это фундаментальная особенность Kafka, которая делает её мощной для распределённой обработки сообщений!