Работает ли Kafka по push-модели

# Работает ли Kafka по push-модели?

Это важный вопрос о том, как Kafka доставляет сообщения потребителям. Прямой ответ: Нет, Kafka работает по PULL-модели, а не push-модели. Это одна из ключевых особенностей архитектуры Kafka, которая отличает её от других систем обмена сообщениями.

Push-модель vs Pull-модель

Push-модель (сервер отправляет)

Сервер сам решает, когда отправить сообщение потребителю:

[Broker] ----push----> [Consumer1]
         ----push----> [Consumer2]
         ----push----> [Consumer3]

Проблемы:

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

Pull-модель (потребитель забирает)

Потребитель сам запрашивает сообщения, когда готов:

[Consumer1] ----pull----> [Broker]
[Consumer2] ----pull----> [Broker]
[Consumer3] ----pull----> [Broker]

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

• Потребитель сам контролирует темп
• Легче обрабатывать backpressure
• Можно эффективнее использовать батчи

Как Kafka работает (pull-модель)

1. Потребитель запрашивает сообщения

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);
consumer.subscribe(Arrays.asList("topic1", "topic2"));

// Потребитель активно забирает сообщения
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    // poll() — это PULL запрос к брокеру
    
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Offset: " + record.offset() + ", Value: " + record.value());
    }
}

2. Broker возвращает батч сообщений

Броker не отправляет сообщения самостоятельно. Вместо этого:

Потребитель: "Дай мне сообщения из партиции 0 начиная с offset 100"
Броker: "Вот 10 сообщений от offset 100 до 109"
Потребитель: "Спасибо, обработаю и вернусь за следующей порцией"

3. Потребитель контролирует темп

// Сколько сообщений за раз забирать
props.put("fetch.min.bytes", "1");        // Минимум 1 байт
props.put("fetch.max.wait.ms", "500");    // Максимум 500мс ожидания
props.put("max.poll.records", "500");     // Максимум 500 записей за раз

// Потребитель сам решает, когда обработал батч
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    processRecord(record);
}
consumer.commitSync();  // Фиксируем прогресс

Преимущества pull-модели в Kafka

1. Backpressure (обработка перегрузки)

Потребитель может замедлить темп обработки:

// Если потребитель медленный, он просто реже вызывает poll()
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    
    for (ConsumerRecord<String, String> record : records) {
        // Тяжелая обработка
        Thread.sleep(100);  // Медленно
        processRecord(record);
    }
}

2. Батчирование

Можно эффективнее обрабатывать данные:

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    
    // Батчевая вставка в БД (быстрее, чем по одной)
    List<String> batch = new ArrayList<>();
    for (ConsumerRecord<String, String> record : records) {
        batch.add(record.value());
    }
    database.batchInsert(batch);  // Одна операция для всего батча
}

3. Контроль смещения (offset)

Потребитель полностью контролирует, какие сообщения он обработал:

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

for (ConsumerRecord<String, String> record : records) {
    try {
        processRecord(record);
        // Коммитим только после успешной обработки
        consumer.commitSync();
    } catch (Exception e) {
        // Если ошибка — сообщение останется необработанным
        // Потребитель вернётся к старому offset при перезагрузке
    }
}

4. Масштабируемость

Потребители не перегружают друг друга, каждый работает со своим темпом

Как это выглядит на практике

@Component
public class KafkaMessageConsumer {
    private static final Logger logger = LoggerFactory.getLogger(KafkaMessageConsumer.class);
    
    private final KafkaConsumer<String, String> consumer;
    private final OrderService orderService;
    
    public KafkaMessageConsumer(OrderService orderService) {
        this.orderService = orderService;
        this.consumer = createConsumer();
    }
    
    public void startConsuming() {
        consumer.subscribe(Collections.singletonList("orders"));
        
        // PULL-модель: потребитель сам запрашивает сообщения
        while (true) {
            // Это PULL запрос! Потребитель сам запрашивает сообщения
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            
            if (records.isEmpty()) {
                logger.info("No messages available");
                continue;
            }
            
            for (ConsumerRecord<String, String> record : records) {
                try {
                    logger.info("Received: {}", record.value());
                    Order order = parseOrder(record.value());
                    
                    // Обработка может быть медленной
                    orderService.process(order);
                    
                    // Коммитим смещение после обработки
                    consumer.commitSync();
                } catch (Exception e) {
                    logger.error("Error processing message", e);
                    // Не коммитим — сообщение будет обработано заново
                }
            }
        }
    }
    
    private KafkaConsumer<String, String> createConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-service-group");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("max.poll.records", 100);      // Забирать не более 100 записей
        props.put("session.timeout.ms", 30000);  // 30 сек таймаут
        return new KafkaConsumer<>(props);
    }
}

Почему именно pull-модель?

1. Простота масштабирования — потребители не зависят от брокера
2. Лучший контроль — потребитель сам решает, что обрабатывать
3. Надёжность — потребитель может повторить обработку
4. Эффективность — батчирование и оптимизация
5. Меньше накладных расходов — брокер не помнит статус потребителя

Сравнение с RabbitMQ (push-модель)

Kafka (pull):              RabbitMQ (push):
Consumer активно          Consumer ждёт
запрашивает сообщения     сообщения от брокера

Потребитель ориентирован  Брокер ориентирован
на темп обработки         на доставку

Дольше батчи             Быстрая доставка

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

• Хотя Kafka использует pull-модель, её API скрывает эту сложность
• Consumer сам вызывает poll() — это и есть pull
• Offset коммит — это подтверждение обработки
• Группа потребителей (group.id) позволяет масштабировать
• Каждая партиция обслуживается только одним потребителем в группе

Выводы

Kafka работает исключительно по pull-модели, где потребитель активно запрашивает сообщения у брокера. Это позволяет потребителям контролировать темп обработки, предотвращать перегрузки и использовать батчирование. Pull-модель — одна из причин популярности Kafka для обработки больших объёмов данных в масштабируемых системах.