php版本為：7.2

# cd /var/www/html/

生產者，producer.php

$rk = new RdKafka\Producer();

$rk->setLogLevel(LOG_DEBUG); // 設置日志級別

$rk->addBrokers('127.0.0.1'); // 添加經紀人，就是ip地址

$topic = $rk->newTopic("test2"); // 新建主題

// 第一個參數：是分區。RD_KAFKA_PARTITION_UA代表未分配，并讓librdkafka選擇分區

// 第二個參數：是消息標志，必須為0

// 第三個參數：消息，如果不為NULL，它將被傳遞給主題分區程序

$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Message'); // 生成并發送單個消息

消費者，consumer.php

$rk = new RdKafka\Consumer();

$rk->setLogLevel(LOG_DEBUG); // 設置日志級別

$rk->addBrokers("127.0.0.1"); // 添加經紀人，就是ip地址

$topic = $rk->newTopic("test2"); // 這里的$rk和生產者是不同的類哦

// 第一個參數分區ID

// 第二個參數是開始消費的偏移量，有效值

$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);

while (true) {

// 第一個參數要消耗的分區

// 第二個參數是等待收到消息的最長時間，1000是一秒

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

if (@$msg->err) {

echo $msg->errstr(), "\n"; // 輸出錯誤

break;

} else {

echo @$msg->payload, "\n"; // 輸出消息

}

}

測試

# cd /var/www/html/

# php consumer.php

新開一個終端，運行生產者，發送消息

# php producer.php

每當生產者運行一次就會發送一條消息，消費者那邊就會接收到這條消息并進行相應的處理。

總結

以上是生活随笔為你收集整理的php rdkafka实例,kafka 消息队列 php-rdkafka扩展示例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。