php-rdkafka扩展的安装

安装 PHP 扩展

一、安装 librdkafka 扩展

rdkafka 安装需要依赖 librdkafka , 所以先安装 librdkafka

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

我这里的名字是 librdkafka

cd librdkafka
./configure
make && make install

二、安装 php-rdkafka 扩展

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

我这里的名字是 php-rdkafka-master

cd php-rdkafka
phpize
./configure --with-php-config=/usr/local/php/bin/php-config

#这里根据自己的情况填写路径

make && make install

成功会显示这个 这样的就可以直接添加

三、修改配置文件

在 php-ini 加上 在PHPinfo输出里面看你的配置文件地址

extension=rdkafka.so

如果提示找不到rdkafka.so 就直接全局搜 然后写绝对路径就可以了

重启，php-fpm，就应该可以看到该扩展。

执行phpinfo（）

应该就有了 这个时候就可以使用了

四、简单示例

生产者 创建一个生产者Producer.php

<?php
     $rk = new \RdKafka\Producer();
        $rk->addBrokers("localhost:9092"); //kafka服务器地址
         
        $topic = $rk->newTopic("qzcsbj"); //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);
        }
?>

消费者 创建一个消费者 Consumer.php

<?php
        $conf = new \RdKafka\Conf();
        $conf->set('group.id', 'myConsumerGroup');
 
        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers("localhost: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("qzcsbj", $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;
            }
        }
?>