php-pubsub-kafka
A Kafka adapter for the php-pubsub package.
Installation
-
Install librdkafka c library
$ cd /tmp $ mkdir librdkafka $ cd librdkafka $ git clone https://github.com/edenhill/librdkafka.git . $ ./configure $ make $ make install
-
Install the php-rdkafka PECL extension
$ pecl install rdkafka
-
Add the following to your php.ini file to enable the php-rdkafka extension
extension=rdkafka.so -
composer require superbalist/php-pubsub-kafka
Usage
// create consumer
$topicConf = new \RdKafka\TopicConf();
$topicConf->set('auto.offset.reset', 'largest');
$conf = new \RdKafka\Conf();
$conf->set('group.id', 'php-pubsub');
$conf->set('metadata.broker.list', '127.0.0.1');
$conf->set('enable.auto.commit', 'false');
$conf->set('offset.store.method', 'broker');
$conf->set('socket.blocking.max.ms', 50);
$conf->setDefaultTopicConf($topicConf);
$consumer = new \RdKafka\KafkaConsumer($conf);
// create producer
$conf = new \RdKafka\Conf();
$conf->set('socket.blocking.max.ms', 50);
$conf->set('queue.buffering.max.ms', 20);
$producer = new \RdKafka\Producer($conf);
$producer->addBrokers('127.0.0.1');
$adapter = new \Superbalist\PubSub\Kafka\KafkaPubSubAdapter($producer, $consumer);
// consume messages
// note: this is a blocking call
$adapter->subscribe('my_channel', function ($message) {
var_dump($message);
});
// publish messages
$adapter->publish('my_channel', 'HELLO WORLD');
$adapter->publish('my_channel', ['hello' => 'world']);
$adapter->publish('my_channel', 1);
$adapter->publish('my_channel', false);
// publish multiple messages
$messages = [
['hello' => 'world'],
'lorem ipsum',
];
$adapter->publishBatch('my_channel', $messages);Examples
The library comes with examples for the adapter and a Dockerfile for running the example scripts.
Run make up.
You will start at a bash prompt in the /opt/php-pubsub directory.
If you need another shell to publish a message to a blocking consumer, you can run make shell
To run the examples:
$ php examples/KafkaConsumerExample.php
$ php examples/KafkaPublishExample.php (in a separate shell)