av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術文章
文章詳情頁

利用擴展的方式在PHP中使用Kafka的教程分享

瀏覽:4日期:2022-06-14 10:34:01
目錄前言安裝1. 下載2. 目錄使用1. 生產2. 消費(從指定的 partition 消費)其他解決方法前言

由于之前在 PHP 中使用 Kafka 是通過 composer 包的方式,由于 nmred/kafka-php 很久沒有維護,并且網上相關問題的文章也比較少。所以我這次換成 PHP 擴展 RdKafka 繼續使用,主要介紹擴展安裝和這種方式的基本操作。

安裝1. 下載

地址(找到與自己環境匹配的就可以)

2. 目錄

由于 php-rdkafka 依賴 librdkafka,linux 就需要先安裝 librdkafka 后安裝 php-rdkafka,而 windows 版本是如下幾個文件,安裝方法如下:

(1). 將 librdkafka.dll 和 librdkafka.pdb 放入 PHP 安裝的根目錄下,而 php_rdkafka.dll 和 php_rdkafka.pdb 放入 PHP 安裝目錄的 ext 下。

(2). php.ini 配置文件添加 extension=php_rdkafka.dll,最后重啟 PHP。

(3). php-m 或這 phpinfo (); 就可以查看到擴展了。

通過 get_declared_classes() 也可以查看到擴展里預設的函數了。

使用1. 生產public function kafkaTest(){$rk = new \RdKafka\Producer();$rk->addBrokers('127.0.0.1:9092');$topic = $rk->newTopic('shop');$ret = [];for ($i = 0; $i < 5; $i++) {$content = '第' . $i . '次發送失敗';$message = ['mobile' => '15623652142', 'content' => $content];$payload = json_encode($message);// 指定向0號partition生產數據$ret[]['produce_res'] = $topic->produce(0, 0, $payload, 'sms_$i');// 隨機選擇partition//$topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload);if ($rk->getOutQLen() > 0) {$ret[]['produce_poll'] = $rk->poll(500);} else {$ret[]['produce_poll'] = $rk->poll(0);}}dump($ret);}

2. 消費(從指定的 partition 消費)protected function execute(Input $input, Output $output){$output->writeln('!!!hello kafka!!!');$conf = new \RdKafka\Conf();$conf->set('group.id', 'sms-consumer-group');$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('shop', $topicConf);$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);while(true) {// 設置消費時的時間間隔,單位毫秒,以下表示5秒消費一個$message = $topic->consume(0, 5000);if ($message) {echo '讀取到消息\n\r';// 消息對象,包括消息主題,消息創建時間戳,消息分區編號,消息主體,消息鍵名,消息長度等var_dump($message);switch ($message->err) {case RD_KAFKA_RESP_ERR_NO_ERROR:echo '讀取消息成功:\n\r';var_dump($message->payload);break;case RD_KAFKA_RESP_ERR__PARTITION_EOF:echo '讀取消息失敗\n\r';break;case RD_KAFKA_RESP_ERR__TIMED_OUT:echo '請求超時\n\r';break;default:throw new \Exception($message->errstr(), $message->err);break;}} else {echo '未讀取到消息\n\r';}}$output->writeln('!!!the end!!!');}

其他

在執行消費過程中,發現 kafka 停止服務,拋出的異常:ERROR Shutdown broker because all log dirs in /tmp/kafka-logs have failed。

解決方法

刪除 kafka-logs 下的所有日志,再重新啟動 Kafaka, kafka-server-start.bat ....\config\server.properties &

到此這篇關于利用擴展的方式在PHP中使用Kafka的教程分享的文章就介紹到這了,更多相關PHP使用Kafka內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網!

標簽: PHP
主站蜘蛛池模板: 国产精品久久久久久久久久尿 | 日韩欧美国产成人一区二区 | 九九久久精品 | 国产精品久久亚洲7777 | 亚洲成人网在线观看 | 亚洲成网站 | 成人欧美一区二区三区在线播放 | 日韩欧美在线免费观看视频 | 特一级黄色毛片 | 男人天堂视频在线观看 | 男女那个视频 | 日韩一二三区 | 一级美国黄色片 | 在线视频91 | 日韩一区二区三区在线视频 | 欧美视频免费在线观看 | 久久久久国产一区二区三区四区 | 久久国产精99精产国高潮 | 羞羞网站免费观看 | 最新免费黄色网址 | 精品国产乱码久久久久久果冻传媒 | 久久久久久久久毛片 | 欧美一区二区三区,视频 | 亚洲444kkkk在线观看最新 | 亚洲精品免费在线观看 | 久久久一| 97色在线观看免费视频 | 综合色在线| 中文字幕一区在线观看视频 | 欧美精品一区二区三 | 91精品国产91久久久久久吃药 | 成人1区 | 国产yw851.c免费观看网站 | 涩涩视频在线观看免费 | 精品国产一区二区三区久久影院 | 天天操天天舔 | 国产精品欧美一区二区三区不卡 | 国产精品久久久亚洲 | 亚洲免费在线播放 | 亚洲欧洲精品在线 | 国产精品久久久久一区二区 |