PHP操作Kafka：phprdkafka扩展的安装

【相关文章】 Linux下搭建Kafka服务

1、安装librdkafka：
librdkafka 是 Apache Kafka 协议的 C 库实现，提供生产者、消费者和管理客户端。它的设计考虑到了消息传递的可靠性和高性能，当前的数据对于生产者超过100 万 msgs/秒，对于消费者超过300万 msgs/秒。

#进入安装包存放目录
cd /usr/local/src

#克隆安装包
git clone https://github.com/edenhill/librdkafka.git

#进入安装目录
cd librdkafka/

#配置
./configure

#编译 安装
make && make install

2、安装php-rdkafka扩展：

#进入安装包存放目录
cd /usr/local/src
 
#克隆安装包
git clone https://github.com/arnaud-lb/php-rdkafka.git
 
#移到安装目录
cd php-rdkafka/
 
#安装
phpize
./configure --with-php-config=/www/server/php/73/bin/php-config
make && make install
 
#php.ini中添加：
extension = rdkafka.so

查看phpinfo()，如图安装成功：

     

3、测试：

      1）创建一个消费者 Consumer.php：（tp5）

<?php
namespace app\index\controller;
class Consumer
{
    public function index()
    {
        $conf = new \RdKafka\Conf();
        $conf->set('group.id', 'myConsumerGroup');
 
        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers("127.0.0.1:9092");
 
        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('auto.commit.interval.ms', 100);
        $topicConf->set('offset.store.method', 'file');
        $topicConf->set('offset.store.path', sys_get_temp_dir());
        $topicConf->set('auto.offset.reset', 'smallest');
 
        $topic = $rk->newTopic("topic1", $topicConf);
 
        // Start consuming partition 0
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
 
        while (true) {
            $message = $topic->consume(0, 120*10000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                //没有错误打印信息
                    var_dump($message);
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    echo "等待接收信息\n";
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    echo "超时\n";
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
                    break;
            }
        }
    }
}

2）创建一个生产者Producer.php ：

<?php
namespace app\index\controller;
class Producer
{
    public function index()
    {
        $rk = new \RdKafka\Producer();
        $rk->addBrokers("127.0.0.1"); //kafka服务器地址
        // $rk->addBrokers("10.0.0.1,10.0.0.2");  //多服务器地址
 
        $topic = $rk->newTopic("topic1"); //topic名称
        for ($i = 0; $i < 10; $i++) {
            $topic->produce(RD_KAFKA_PARTITION_UA, 0, "发送信息： $i");
            $rk->poll(0);
        }
        while ($rk->getOutQLen() > 0) {
            $rk->poll(50);
        }
    }
}

3）执行消费者监听消息：php index.php /index/consumer/index，然后执行生产者：php index.php /index/producer/index，可见消费者收到了消息：（tp5）