Скачать kafka php - скачать исходный код kafka php

kafka php

Другие категории

V0.2.0.8

Скачать

Кафка-php

中文文档

Kafka-php — это клиент Kafka на чистом PHP, который в настоящее время поддерживает версию Kafka выше 0.8.x. Этот проект v0.2.x и v0.1.x несовместимы при использовании исходной версии v0.1.x. Вы можете обратиться к document Kafka PHP v0.1.x Document, но рекомендуется переключиться на v0.2.x. v0.2.x использует асинхронную реализацию PHP и взаимодействие брокера Kafka, более стабильное, чем v0.1.x, поскольку использование языка PHP, поэтому не компилирует никаких расширений, может использоваться для снижения затрат на доступ и обслуживание.

Требования

Минимальная версия PHP: 7.1.
Версия Кафки выше 0,8
Потребительскому модулю требуется версия брокера Kafka выше 0.9.0.

Установка

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

Установка композитора

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

$ composer require nmred/kafka-php

Вот минимальный пример файла композитора.json:

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

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

Свойства конфигурации описаны в разделе Конфигурация.

Продюсер

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

 <?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 ( ' 1.0.0 ' );
$ config -> setRequiredAck ( 1 );
$ config -> setIsAsyn ( false );
$ config -> setProduceInterval ( 500 );
$ producer = new  Kafka  Producer (
    function () {
        return [
            [
                ' 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 ( ' 1.0.0 ' );
$ config -> setRequiredAck ( 1 );
$ config -> setIsAsyn ( false );
$ config -> setProduceInterval ( 500 );
$ producer = new  Kafka  Producer ();
$ producer -> setLogger ( $ logger );

for ( $ i = 0 ; $ i < 100 ; $ i ++) {
    $ producer -> send ([
        [
            ' topic ' => ' test1 ' ,
            ' value ' => ' test1....message. ' ,
            ' key ' => '' ,
        ],
    ]);
}

Потребитель

 <?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 ConsumerConfig:: getInstance ();
$ config -> setMetadataRefreshIntervalMs ( 10000 );
$ config -> setMetadataBrokerList ( ' 10.13.4.159:9192 ' );
$ config -> setGroupId ( ' test ' );
$ config -> setBrokerVersion ( ' 1.0.0 ' );
$ config -> setTopics ([ ' test ' ]);
// $ config - > setOffsetReset ( 'earliest' );
$ consumer = new  Kafka  Consumer ();
$ consumer -> setLogger ( $ logger );
$ consumer -> start ( function ( $ topic , $ part , $ message ) {
	var_dump ( $ message );
});