Перейти к основному содержимому

Kafka: высокопроизводительный распределенный стриминг

Apache Kafka — это распределенное хранилище событий и платформа для потоковой обработки данных. Она разработана для обработки огромных объемов данных в реальном времени с низкой задержкой и высокой пропускной способностью.

1. Основные концепции и EOS​

Семантика Exactly-Once (EOS)​

Kafka обеспечивает семантику «ровно один раз» (exactly-once) с помощью:

  • Идемпотентные продюсеры: Гарантируют, что повторные попытки отправки не приведут к дублированию сообщений за счет присвоения порядкового номера каждому сообщению.
  • Транзакции: Позволяют объединять несколько операций Kafka (в разных разделах/топиках) в единую атомарную единицу.
  • Управление смещениями (offsets): При использовании Kafka Streams или транзакционных потребителей смещения фиксируются атомарно вместе с результатами обработки.

Проблемы:

  • Требуется идемпотентность на стороне потребителя (обработка одного и того же сообщения дважды, если фиксация смещения не удалась).
  • Продюсеры должны включить acks=all и enable.idempotence=true.
  • Увеличение задержки из-за накладных расходов на транзакционные записи и координацию.

2. High-Water Mark (HW)​

  • High-Water Mark (HW) — это последнее смещение, которое считается «зафиксированным» (committed) и может быть прочитано потребителями.
  • Это обеспечивает согласованность: потребители не могут читать незафиксированные данные, которые могут исчезнуть в случае сбоя лидера.
  • Если реплика-последователь (follower) отстает от HW, она не может быть выбрана лидером до тех пор, пока не догонит его.

3. Сбой лидера и сообщения в полете​

  • Kafka использует ZooKeeper или KRaft для выбора нового лидера из In-Sync Replicas (ISR).
  • In-flight сообщения (с acks=1 или acks=0) могут быть потеряны, если они не были реплицированы на последователя до падения лидера.
  • С acks=all сообщение считается успешно отправленным только если оно было реплицировано на все ISR.
  • Крайний случай: Если нет доступных ISR, Kafka может приостановить запись, чтобы предотвратить потерю данных (в зависимости от настройки min.insync.replicas).

4. Log Compaction vs. Retention​

  • Log Compaction (Уплотнение логов): Сохраняет как минимум последнюю версию для каждого ключа. Старые версии удаляются в процессе уплотнения. Полезно для хранения текущего состояния (changelogs).
  • Retention Policies (Политики удержания): Удаляют сообщения на основе времени (log.retention.ms) или общего размера (log.retention.bytes), независимо от ключа.
  • Сложный случай: Если потребитель был отключен дольше периода удержания, он может пропустить исторические обновления или изменения состояния.

5. Нечистый выбор лидера (Unclean Leader Election)​

  • Когда unclean.leader.election.enable=true, последователь, не находящийся в синхронизации (не в ISR), может стать лидером.
  • Опасность: Это может привести к потере данных и несогласованности, так как новый лидер может не иметь всех зафиксированных сообщений от предыдущего лидера.
  • В продакшене это значение обычно должно быть false, если только доступность не важнее согласованности.

6. Предотвращение дубликатов (Идемпотентность)​

  • Каждое сообщение имеет Producer ID (PID) и Sequence Number.
  • Брокер отслеживает последний подтвержденный порядковый номер для каждого PID.
  • Если продюсер повторно отправляет сообщение (например, из-за таймаута сети), брокер отбрасывает его как дубликат.

7. Backpressure и отставание (Lag)​

  • Consumer Lag (Отставание потребителя): Возникает, когда скорость продюсера превышает скорость обработки потребителем.
  • Обработка:
    • Используйте методы pause() и resume() в API потребителя.
    • Масштабируйтесь горизонтально, добавляя больше потребителей в Consumer Group.
    • Оптимизируйте логику обработки или используйте пакетную обработку (batching).
    • Внедрите ограничение скорости (rate limiting) на стороне продюсера.

8. ISR, OSR и недореплицированные разделы​

  • ISR (In-Sync Replicas): Последователи, которые полностью синхронизированы с лидером.
  • OSR (Out-of-Sync Replicas): Последователи, которые отстают.
  • Under-Replicated Partitions (URP): Разделы, в которых количество ISR меньше заданного фактора репликации.

9. Kafka Streams и обработка состояний​

  • Использует RocksDB для эффективного локального хранения состояния.
  • Changelog Topics: Каждое изменение состояния записывается в топик Kafka, что позволяет восстановить состояние после сбоя.
  • Punctuation: Позволяет выполнять периодические задачи (например, оконные функции) без повторного чтения всех данных.

10. Мониторинг отставания потребителя​

  • Используйте CLI kafka-consumer-groups.sh.
  • Мониторьте метрики JMX, такие как records-lag-max.
  • Используйте инструменты вроде Prometheus, Grafana или Burrow для оповещений.

Примеры реализации на PHP 8.4​

Для работы с Kafka в PHP стандартом де-факто является расширение php-rdkafka.

1. Разделение топика (Divider/Partitioner)​

Использование ключей позволяет гарантировать, что сообщения, относящиеся к одной сущности (например, user_id), всегда попадают в один и тот же раздел.

<?php

declare(strict_types=1);

use RdKafka\Conf;
use RdKafka\Producer;

$conf = new Conf();
$conf->set('metadata.broker.list', 'localhost:9092');

$producer = new Producer($conf);
$topic = $producer->newTopic("orders-topic");

// PHP 8.4: Использование ключей для обеспечения порядка внутри раздела
$orderId = "order_12345";
$payload = json_encode(['status' => 'created', 'id' => $orderId]);

// Ключ 'order_12345' гарантирует, что все сообщения для этого заказа попадут в один раздел
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload, $orderId);

$producer->flush(10000);

2. Стратегия с несколькими потребителями (Consumer Groups)​

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

<?php

declare(strict_types=1);

use RdKafka\Conf;
use RdKafka\KafkaConsumer;

$conf = new Conf();
$conf->set('group.id', 'inventory-service-group'); // Несколько инстансов используют один ID
$conf->set('metadata.broker.list', 'localhost:9092');
$conf->set('auto.offset.reset', 'earliest');

$consumer = new KafkaConsumer($conf);
$consumer->subscribe(['orders-topic']);

while (true) {
    $message = $consumer->consume(120 * 1000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // Обработка сообщения только один раз внутри всей группы
            echo "Обработка заказа: " . $message->payload . PHP_EOL;
            $consumer->commitAsync($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
    }
}

3. Предотвращение дубликатов и идемпотентность​

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

Продюсер (Идемпотентность):

$conf->set('enable.idempotence', 'true');
$conf->set('acks', 'all');

Потребитель (Стратегия повтора с идемпотентностью): Если потребитель упал после обработки, но до фиксации смещения, он получит сообщение снова. Приложение должно быть идемпотентным.

// Пример на PHP 8.4: Идемпотентная обработка с проверкой в БД
public function processMessage(object $message): void
{
    $data = json_decode($message->payload, true);
    $messageId = $data['id'] ?? null;

    // Проверяем, обрабатывали ли мы это сообщение ранее
    if ($this->db->exists('processed_messages', ['msg_id' => $messageId])) {
        return; // Уже обработано, пропускаем
    }

    $this->db->transactional(function() use ($data, $messageId) {
        $this->doWork($data);
        $this->db->insert('processed_messages', ['msg_id' => $messageId]);
    });
}

4. Обработка состояний гонки (Race Conditions)​

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

// Использование колонки version для оптимистичной блокировки в PHP 8.4
try {
    $this->db->execute(
        "UPDATE accounts SET balance = balance - :amount, version = version + 1 
         WHERE id = :id AND version = :current_version",
        ['amount' => 100, 'id' => $accountId, 'current_version' => $data['version']]
    );
} catch (LockException $e) {
    // Стратегия повтора: получить актуальное состояние и попробовать снова,
    // или отправить сообщение в ретрай-топик.
    $this->retryTopic->produce(RD_KAFKA_PARTITION_UA, 0, $message->payload);
}