OSCHINA-MIRROR/weiboad-kafka-php

Kafka-php

Kafka-php — это клиент Kafka на чистом PHP, который в настоящее время поддерживает версии Kafka больше 0.8.x. Версии проекта v0.2.x и v0.1.x несовместимы при использовании оригинальной версии v0.1.x. Можно обратиться к документу «Kafka PHP v0.1.x Document» (https://github.com/weiboad/kafka-php/blob/v0.1.6/README.md), но рекомендуется перейти на версию v0.2.x. Версия v0.2.x использует асинхронную реализацию PHP и взаимодействие с брокером Kafka, более стабильна, чем версия v0.1.x, и эффективна, поскольку используется язык PHP, поэтому не нужно компилировать какие-либо расширения, что позволяет снизить затраты на доступ и обслуживание.

Требования

Минимальная версия PHP: 7.1
Версия Kafka больше 0.8
Для модуля потребителя требуется версия брокера Kafka больше 0.9.0

Установка

Добавьте каталог lib в PHP include_path и используйте автозагрузчик, подобный тому, что находится в каталоге примеров (код следует соглашению PEAR/Zend один класс на файл).

Composer Install

Просто добавьте зависимость от nmred/kafka-php в файл composer.json вашего проекта, если вы используете Composer для управления зависимостями вашего проекта. Вот минимальный пример файла composer.json:

{
    "require": {
        "nmred/kafka-php": "0.2.*"
    }
}

Конфигурация

Свойства конфигурации описаны в Configuration.

Producer

Асинхронный режим

<?php
require '../vendor/autoload.php';
date_default_timezone_set('PRC');
use Monolog\Logger;
use Monolog\Handler\StdoutHandler;
// Create the logger
$logger = new Logger('my_logger');
// Now add some handlers
$logger->pushHandler(new StdoutHandler());

$config = \Kafka\ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('10.13.4.159:9192');
$config->setBrokerVersion('0.9.0.1');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer(function() {
    return array(
        array(
            'topic' => 'test',
            'value' => 'test....message.',
            'key' => 'testkey',
            ),
    );
});
$producer->setLogger($logger);
$producer->success(function($result) {
    var_dump($result);
});
$producer->error(function($errorCode) {
        var_dump($errorCode);
});
$producer->send(true);

Синхронный режим

<?php
require '../vendor/autoload.php';
date_default_timezone_set('PRC');
use Monolog\Logger;
use Monolog\Handler\StdoutHandler;
// Create the logger
$logger = new Logger('my_logger');
// Now add some handlers
$logger->pushHandler(new StdoutHandler());

$config = \Kafka\ProducerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('127.0.0.1:9192');
$config->setBrokerVersion('0.9.0.1');
$config->setRequiredAck(1);
$config->setIsAsyn(false);
$config->setProduceInterval(500);
$producer = new \Kafka\Producer();
$producer->setLogger($logger);

for($i = 0; $i < 100; $i++) {
        $result = $producer->send(array(
                array(

...

*Перевод выполнен автоматически. В тексте могут быть неточности.* ## Потребитель

```php
<?php
require '../vendor/autoload.php';
date_default_timezone_set('PRC');
use Monolog\Logger;
use Monolog\Handler\StdoutHandler;
// Создаем логгер
$logger = new Logger('my_logger');
// Теперь добавляем обработчики
$logger->pushHandler(new StdoutHandler());

$config = \Kafka\ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(10000);
$config->setMetadataBrokerList('10.13.4.159:9192');
$config->setGroupId('test');
$config->setBrokerVersion('0.9.0.1');
$config->setTopics(array('test'));
//$config->setOffsetReset('earliest');
$consumer = new \Kafka\Consumer();
$consumer->setLogger($logger);
$consumer->start(function($topic, $part, $message) {
    var_dump($message);
});

Низкоуровневый API

См. Пример.