PHP操作Kafka：php-rdkafka擴展的安裝

【相關文章】 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）