Kafka在php中的使用----生产者与消费者

安装教程 kafka和php的rdkafka扩展教程网上有很多,大家可以自行查询,例如:Kafka-php－使用 PHP 编写的 Kafka 客户端

不清楚里面的api的可以在文档中查询 kafka中文文档

创建 composer.json填写内容

{
  "require": {
        "nmred/kafka-php": "v0.2.0.8"
  }
}

<?php
require_once __DIR__ . '/Vendor/autoload.php';

use KafkaProducerConfig;
use KafkaProducer;

//异步方式调用
config = ProducerConfig::getInstance();
//设置数据刷新时间(毫秒)config->setMetadataRefreshIntervalMs(10);
//地址
config->setMetadataBrokerList('localhost:9092');
//设置代理版本config->setBrokerVersion('0.9.0.1');
//开启消息确认
config->setRequiredAck(1);config->setIsAsyn(false);
//设置生产间隔
config->setProduceInterval(500);
//生产者producer = new Producer(function () {
    return array([
        'topic' => 'test',//主题
        'value' => 'test message',
        'key' => 'testKey',//key
    ]);
});
producer->success(function (result) {
    echo '投递成功' . json_encode(result, 256) . PHP_EOL;
});producer->error(function (result) {
    echo '投递失败' . json_encode(result, 256) . PHP_EOL;
});
$producer->send(true);

<?php
require_once __DIR__ . '/Vendor/autoload.php';

use KafkaProducerConfig;
use KafkaProducer;

config = ProducerConfig::getInstance();
//这是元组数据刷新间隔毫秒config->setMetadataRefreshIntervalMs(10);
//代理地址
config->setMetadataBrokerList('localhost:9092');
//设置代理版本config->setBrokerVersion('0.9.0.1');
//开启消息确认
config->setRequiredAck(1);config->setIsAsyn(false);
//生产间隔
config->setProduceInterval(10);producer = new Producer();
for (i = 0;i < 100; i++) {result = producer->send(array(
        [
            'topic' => 'test',//主题
            'value' => 'test message',
            'key' => '',//key
        ]
    ));
    echo '投递成功' . json_encode(result, 256) . PHP_EOL;
}

<?php
require_once __DIR__ . '/Vendor/autoload.php';
use KafkaConsumerConfig;
use KafkaConsumer;
config = ConsumerConfig::getInstance();config->setMetadataRefreshIntervalMs(10);
config->setMetadataBrokerList('localhost:9092');
//设置分组分区config->setGroupId('test');
config->setBrokerVersion('0.9.0.1');config->setTopics(['test']);
//设置偏移量
config->setOffsetReset('earliest');consumer = new Consumer();
consumer->start(function (topic, part,message) {
    var_dump($message);
});