README

php-pubsub 软件包的 Kafka 适配器。

安装

1. 安装 librdkafka C 库

   $ cd /tmp
   $ mkdir librdkafka
   $ cd librdkafka
   $ git clone https://github.com/edenhill/librdkafka.git .
   $ ./configure
   $ make
   $ make install

2. 安装 php-rdkafka PECL 扩展

   $ pecl install rdkafka

3. 将以下内容添加到您的 php.ini 文件中，以启用 php-rdkafka 扩展：extension=rdkafka.so

4. composer require superbalist/php-pubsub-kafka

用法

// 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);

示例

该库包含适配器的 示例 和运行示例脚本的 Dockerfile。

运行 make up。

您将进入 bash 提示符，位于 /opt/php-pubsub 目录。

如果您需要另一个 shell 来向阻塞消费者发布消息，可以运行 make shell

运行示例

$ php examples/KafkaConsumerExample.php
$ php examples/KafkaPublishExample.php (in a separate shell)