Работа с Kafka в PHP: библиотеки и примеры

Работа с Kafka в PHP: библиотеки, подключение и примеры кода

Apache Kafka — это распределённая платформа для потоковой передачи данных, которая позволяет публиковать, хранить и обрабатывать потоки сообщений в реальном времени. В экосистеме PHP для работы с Kafka используется расширение php-rdkafka — обёртка над библиотекой librdkafka, написанной на C. В этой статье мы разберём, как установить инструмент, написать продюсера и консюмера, а также рассмотрим типичные сценарии использования.

Что такое php-rdkafka?

php-rdkafka — это официальное PHP-расширение для работы с Apache Kafka. Оно предоставляет высокопроизводительный интерфейс для отправки и получения сообщений. Расширение поддерживает все ключевые возможности Kafka: топики, партиции, группы консюмеров, ручные коммиты, сжатие и многое другое.

Основные возможности:

Создание продюсеров (producers) для отправки сообщений в топики.
Создание консюмеров (consumers) для чтения сообщений из топиков.
Поддержка групп консюмеров и автоматического перебалансирования.
Работа с партициями и ручное управление offset'ами.
Сжатие сообщений (gzip, snappy, lz4, zstd).
Асинхронная отправка и пакетная обработка.

Установка php-rdkafka

Установка расширения возможна через PECL или сборку из исходников. Рекомендуемый способ — через PECL:

# Установка через PECL
pecl install rdkafka

# Для Ubuntu/Debian предварительно установите librdkafka-dev
sudo apt-get install librdkafka-dev
pecl install rdkafka

# Добавьте расширение в php.ini
extension=rdkafka.so

После установки проверьте, что расширение загружено:

php -m | grep rdkafka

Если вы используете Composer, дополнительно можно установить пакет arnaud-lb/php-rdkafka для автозагрузки:

composer require arnaud-lb/php-rdkafka

Пример продюсера (Producer)

Продюсер отправляет сообщения в указанный топик Kafka. Рассмотрим базовый пример с синхронной отправкой:

<?php
require_once 'vendor/autoload.php';

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

$producer = new RdKafka\Producer($conf);
$topic = $producer->newTopic('test_topic');

// Отправка сообщения
$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello, Kafka from PHP!');

// Ожидание отправки всех сообщений
$producer->flush(1000);
echo "Сообщение отправлено\n";

Пояснения к коду:

metadata.broker.list — адрес брокера Kafka (по умолчанию localhost:9092).
RD_KAFKA_PARTITION_UA — автоматический выбор партиции.
flush(1000) — блокирует выполнение до отправки всех сообщений (таймаут 1 секунда).

Пример консюмера (Consumer)

Консюмер читает сообщения из топика. Важно указать группу консюмера и стратегию коммита offset'ов:

<?php
require_once 'vendor/autoload.php';

$conf = new RdKafka\Conf();
$conf->set('metadata.broker.list', 'localhost:9092');
$conf->set('group.id', 'my_consumer_group');
$conf->set('auto.offset.reset', 'earliest');

$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe(['test_topic']);

echo "Ожидание сообщений...\n";

while (true) {
    $message = $consumer->consume(120 * 1000); // таймаут 2 минуты
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            echo "Получено сообщение: " . $message->payload . "\n";
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "Достигнут конец партиции\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Таймаут\n";
            break;
        default:
            echo "Ошибка: " . $message->errstr() . "\n";
            break;
    }
}

Важные параметры конфигурации консюмера: