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

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

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

1. Библиотека php-rdkafka

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

Установка

Установка требует наличия librdkafka и PECL:

# Установка librdkafka (Ubuntu/Debian)
sudo apt-get install librdkafka-dev

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

# Включение в php.ini
extension=rdkafka.so

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

Продюсер отправляет сообщения в топик Kafka. Важно правильно настроить асинхронную отправку и обработку ошибок:

<?php
use RdKafka\Producer;
use RdKafka\Conf;

$conf = new Conf();
$conf->set('bootstrap.servers', 'localhost:9092');
$conf->set('acks', 'all'); // ожидание подтверждения от всех реплик

$producer = new Producer($conf);
$topic = $producer->newTopic('test-topic');

// Отправка сообщения
$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello Kafka!');
$producer->flush(1000); // ожидание доставки (1 сек)

echo "Сообщение отправлено\n";

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

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

<?php
use RdKafka\KafkaConsumer;
use RdKafka\Conf;

$conf = new Conf();
$conf->set('bootstrap.servers', 'localhost:9092');
$conf->set('group.id', 'my-consumer-group');
$conf->set('auto.offset.reset', 'earliest');

$consumer = new 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;
    }
}

2. Библиотека longlang/phpkafka

longlang/phpkafka — это реализация протокола Kafka на чистом PHP без внешних зависимостей. Она проще в установке (не требует компиляции) и подходит для разработки, тестирования или проектов с невысокими требованиями к производительности.

Установка

Установка через Composer:

composer require longlang/phpkafka

Пример: продюсер

<?php
use longlang\phpkafka\Producer\Producer;
use longlang\phpkafka\Producer\ProducerConfig;

$config = new ProducerConfig();
$config->setBootstrapServer('localhost:9092');
$config->setUpdateBrokers(true);

$producer = new Producer($config);
$producer->send('test-topic', 'Hello Kafka from pure PHP!', 'key-1');
echo "Сообщение отправлено\n";

Пример: консьюмер

<?php
use longlang\phpkafka\Consumer\Consumer;
use longlang\phpkafka\Consumer\ConsumerConfig;

$config = new ConsumerConfig();
$config->setBootstrapServer('localhost:9092');
$config->setGroupId('my-group');
$config->setTopic('test-topic');
$config->setAutoCommit(true);

$consumer = new Consumer($config);
$consumer->subscribe(function ($message) {
    echo "Получено: " . $message->getValue() . "\n";
    return true; // подтверждение обработки
});

3. Сравнение библиотек

Критерий	php-rdkafka	longlang/phpkafka
Также в библиотеке Работа с S3 и облачными хранилищами в Java: библиотека AWS SDK Генерация QR-кодов на Java: библиотека ZXing (подробное руководство) Валидация данных в Go: библиотеки, примеры и лучшие практики Планировщик задач Kotlin: kotlinx-coroutines-core и библиотека korGE CSVImporter: работа с CSV в Swift Рекомендаций Работа с датами в C++: библиотеки chrono, date.h и Boost.DateTime Кэширование в Java: Redis, Ehcache, Caffeine, Hazelcast — обзор и примеры Отправка Email на Java: библиотеки JavaMail и Simple Java Mail InversifyJS: IoC-контейнер для TypeScript-приложений 🍪 Сначала примите условия Чтобы войти на сайт, необходимо принять соглашение об использовании файлов cookie. Мы используем их для авторизации и улучшения работы сайта. Политика конфиденциальности Принять и продолжить Закрыть 🍪 Мы используем файлы cookie для авторизации, аналитики и улучшения работы сайта. Продолжая пользоваться сайтом, вы соглашаетесь с нашей политикой конфиденциальности. Принять Только необходимые