1、首先安装kafka扩展#安装librdkafka:   版本： https://github.com/edenhill/librdkafka/releases/tag/v0.9.2

$   git clone https://github.com/edenhill/librdkafka.git

$  ./configure

$  make

$  sudo make install

#安装 rdkafka.so  版本：https://github.com/arnaud-lb/php-rdkafka/releases/tag/3.0.1

$ git clone https://github.com/arnaud-lb/php-rdkafka.git

$ cd php-rdkafka

$ phpize

$ ./configure

$ make all -j 5

$ sudo make install

2、生产者代码示例

$rcf = new RdKafka\Conf();

$rcf->set('group.id', 'test');  //topicname

$cf = new RdKafka\TopicConf();

$cf->set('offset.store.method', 'broker');

$cf->set('auto.offset.reset', 'smallest');

$rk = new RdKafka\Producer($rcf);

$rk->setLogLevel(LOG_DEBUG);

$rk->addBrokers("127.0.0.1"); //brokeraddr

$topic = $rk->newTopic("test", $cf);  //topicname

for($i = 0; $i < 10; $i++) {

$topic->produce(0,0,'test' . $i);

}

3、消费者代码示例

$rcf = new RdKafka\Conf();

$rcf->set('group.id', 'test');

$rcf->set('broker.version.fallback', '0.8.2');  //brokername,kafkaversion

$cf = new RdKafka\TopicConf();

$cf->set('auto.offset.reset', 'smallest');

$cf->set('auto.commit.enable', true);

$rk = new RdKafka\Consumer($rcf);

$rk->setLogLevel(LOG_DEBUG);

$rk->addBrokers("127.0.0.1"); //brokeraddr

$topic = $rk->newTopic("test", $cf);  //topicname,topicobject

$topic->consumeStart(0,10);  //partition,offset

$msg = $topic->consume(0, 1000);   //partition,timeout

var_dump($msg);