話說用了就要有點(diǎn)產(chǎn)出，要不然過段時(shí)間又忘了，所以在這里就記錄一下試用Kafka的安裝過程和php擴(kuò)展的試用。
實(shí)話說，如果用于隊(duì)列的話，跟PHP比較配的，還是Redis。用的順手，呵呵，只是Redis不能有多個(gè)consumer。但Kafka官方對(duì)PHP不支持，PHP擴(kuò)展是愛好者或使用者寫的。下面就開始講Kafka的安裝吧。我以CentOS6.4為例,64位。
一. 首先確認(rèn)下jdk有沒有安裝
使用命令
[root@localhost ~]# java -version
java version "1.8.0_73"
Java(TM) SE Runtime Environment (build 1.8.0_73-b02)
Java HotSpot(TM) 64-Bit Server VM (build 25.73-b02, mixed mode)
如果有以上信息的話，就往下安裝吧，有些可能是jdk對(duì)不上，那就裝到對(duì)的上的。如果沒有安裝，就看一下下面的jdk安裝方法：
http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
到這個(gè)地址下載jdk8版本，我下載的是jdk-8u73-linux-x64.tar.gz，然后解壓到/usr/local/jdk/下。
然后打開/etc/profile文件
[root@localhost ~]# vim /etc/profile
把下面這段代碼寫到文件里
export JAVA_HOME=/usr/local/jdk/jdk1.8.0_73
export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
export PATH=$JAVA_HOME/bin:$PATH
最后
[root@localhost ~]# source /etc/profile
這時(shí)jdk就生效了，可以使用 java -version驗(yàn)證下。
二. 接下來安裝Kafka
1. 下載Kafka
到http://kafka.apache.org/downloads.html下載相應(yīng)的版本，我使用的是kafka_2.9.1-0.8.2.2.tgz。
2. 下載完解壓到你喜歡的目錄
我是解壓到 /usr/local/kafka/kafka_2.9.1-0.8.2.2
3. 運(yùn)行默認(rèn)的Kafka
啟動(dòng)Zookeeper server
[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/zookeeper-server-start.sh config/zookeeper.properties
啟動(dòng)Kafka server
[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-server-start.sh config/server.properties
運(yùn)行生產(chǎn)者producer
[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
運(yùn)行消費(fèi)者consumer
[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
這樣，在producer那邊輸入內(nèi)容，consumer馬上就能接收到。
4. 當(dāng)有跨機(jī)的producer或consumer連接時(shí)
需要配置config/server.properties的host.name，要不然跨機(jī)的連不上。
三. Kafka-PHP擴(kuò)展
使用了一圈，就https://github.com/nmred/kafka-php可以用。
我是使用composer安裝的，以下是示例：
producer.php
?php
require 'vendor/autoload.php';
while (1) {
$part = mt_rand(0, 1);
$produce = \Kafka\Produce::getInstance('kafka0:2181', 3000);
// get available partitions
$partitions = $produce->getAvailablePartitions('topic_name');
var_dump($partitions);
// send message
$produce->setRequireAck(-1);
$produce->setMessages('topic_name', 0, array(date('Y-m-d H:i:s'));
sleep(3);
}
consumer.php
require 'vendor/autoload.php';
$consumer = \Kafka\Consumer::getInstance('kafka0:2181');
$group = 'topic_name';
$consumer->setGroup($group);
$consumer->setFromOffset(true);
$consumer->setTopic('topic_name', 0);
$consumer->setMaxBytes(102400);
$result = $consumer->fetch();
print_r($result);
foreach ($result as $topicName => $partition) {
foreach ($partition as $partId => $messageSet) {
var_dump($partition->getHighOffset());
foreach ($messageSet as $message) {
var_dump((string)$message);
}
var_dump($partition->getMessageOffset());
}
}
到此這篇關(guān)于PHP擴(kuò)展之kafka安裝應(yīng)用案例詳解的文章就介紹到這了,更多相關(guān)PHP擴(kuò)展之kafka安裝應(yīng)用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- php測(cè)試kafka項(xiàng)目示例
- 使用 PHP Masked Package 屏蔽敏感數(shù)據(jù)的實(shí)現(xiàn)方法
- 完美解決phpdoc導(dǎo)出文檔中@package的warning及Error的錯(cuò)誤
- 利用ThinkPHP內(nèi)置的ThinkAjax實(shí)現(xiàn)異步傳輸技術(shù)的實(shí)現(xiàn)方法
