php连接kafka

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