Упрощённое расширение PHP-Kafka See-KafKa

# Установка librdkafka
git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure
make
make install

# Установка php-rdkafka
git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
phpize
./configure
make all -j 5
make install
# Добавьте следующую информацию в конфигурационный файл php.ini
vim /usr/local/php/etc/php.ini
extension=rdkafka.so  

<?php
/**
 * Пример использования производителя See-Kafka
 * Цикл записи 10 000 сообщений за 15 миллисекунд
 */
``````markdown
## 2.1 Настройка конфигурации Kafka (по умолчанию порт 9092) через запятую
```php
$Kafka_Lite = new Kafka_Lite("127.0.0.1,localhost");
// Установка темы
$Kafka_Lite->setTopic("test");
// Одноразовая запись эффективна, запись 10 000 сообщений за 15 миллисекунд
$Producer = $Kafka_Lite->newProducer();
// Параметры: разделение, содержание сообщения, ключ сообщения (необязательно)
// Разделение: может быть установлено как KAFKA_PARTITION_UA для автоматического распределения, если есть 6 разделов, то при записи будет случайным образом выбран раздел
$Producer->setMessage(0, "Привет");
?>

<?php
/**
 * Пример использования потребителя See-kafka
 */

$Kafka_Lite = new Kafka_Lite("127.0.0.1,localhost");
// Установка Topic
$Kafka_Lite->setTopic("test");
// Установка группы потребителя (не обязательно, если не используется автоматическое получение offset'а)
$Kafka_Lite->setGroup("test");
// Получение экземпляра потребителя
$consumer = $Kafka_Lite->newConsumer();// Получение группы сообщений параметры: разделение, максимальное количество сообщений, offset (необязательно) по умолчанию KAFKA_OFFSET_STORED
$rs = $consumer->getMessages(0, 100);
// Возвращаемое значение - массив, каждый элемент типа Kafka_Message
?>

<?php
/**
 * Пример использования потребителя See-kafka
 * 889 миллисекунд для получения 10 000 сообщений
 */
```// Настройка конфигурации Kafka (по умолчанию порт 9092) через запятую
$Kafka_Lite = new Kafka_Lite("127.0.0.1,localhost");
// Установка Topic
$Kafka_Lite->setTopic("test");
// Установка группы потребителя (не обязательно, если не используется автоматическое получение offset'а)
$Kafka_Lite->setGroup("test");
?>
```// Это настройка определяет, с какого оффсета начинается новый group — с минимального или максимального. По умолчанию используется максимальный (или последний).
$Kafka_Lite->setTopicConf('auto.offset.reset', 'smallest');
// Эта конфигурация определяет, будет ли автоматическое подтверждение сообщений считаться успешным после получения данных, без необходимости вызова метода stop. Однако есть ограничения.
// Чем меньше значение времени, тем быстрее происходит подтверждение. Чем больше время, тем реже происходит подтверждение. При получении одного сообщения и возникновении ошибки, время между получением и выбросом исключения используется для вычисления завершённости обработки.
// Если исключение выбрасывается раньше указанного времени, оффсет не обновляется. Если же время больше этого значения, оффсет обновляется сразу. Рекомендуется установить это значение в диапазоне от 100 до 1000 миллисекунд.
$Kafka_Lite->setTopicConf('auto.commit.interval.ms', 1000);## 3. Конфигурационные файлыSee-kafka предоставляет два типа конфигурационных файлов, передаваемых ключами и значениями. Для получения информации о конкретных конфигурациях и их действиях обратитесь к следующему адресу:

https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

Описание конфигурационных файлов: [https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md "Описание конфигурационных файлов")

```php
$KafKa_Lite->setTopicConf();
$KafKa_Lite->setKafkaConf();

// Этот параметр определяет поведение при использовании новой группы — начинать с минимального смещения или максимального (по умолчанию максимального или последнего).
$KafKa_Lite->setTopicConf('auto.offset.reset', 'smallest');

При инициализации объекта `KafKa_Lite` производится проверка доступности портов кластера. Если ни один из портов недоступен, будет выброшено исключение **Не могу использовать KafKa**. Также можно активировать операцию ping для проверки наличия доступных машин в кластере. При получении исключения от Consumer будет выброшено исключение **KafKa_Exception_Base**, которое имеет код ошибки, который можно использовать для справки в файле Exception/err.php. Рекомендуется использовать try-catch для обработки.

## 5. Заключение

Цель расширения See-KafKa — сделать более удобной интеграцию KafKa с PHP и обеспечить простоту использования. Если вас заинтересует этот проект, попробуйте его использовать, и если возникнут вопросы, вы можете обратиться с отзывами. Автор этого расширения будет продолжать его поддерживать!

Официальная группа поддержки: 438882880