PHP擴(kuò)展之kafka安裝應(yīng)用案例詳解

第PHP擴(kuò)展之kafka安裝應(yīng)用案例詳解話說用了就要有點產(chǎn)出，要不然過段時間又忘了，所以在這里就記錄一下試用Kafka的安裝過程和php擴(kuò)展的試用。

實話說，如果用于隊列的話，跟PHP比較配的，還是Redis。用的順手，呵呵，只是Redis不能有多個consumer。但Kafka官方對PHP不支持，PHP擴(kuò)展是愛好者或使用者寫的。下面就開始講Kafka的安裝吧。我以CentOS6.4為例,64位。

一.首先確認(rèn)下jdk有沒有安裝

使用命令

[root@localhost~]#java-version

javaversion"1.8.0_73"

Java(TM)SERuntimeEnvironment(build1.8.0_73-b02)

JavaHotSpot(TM)64-BitServerVM(build25.73-b02,mixedmode)

如果有以上信息的話，就往下安裝吧，有些可能是jdk對不上，那就裝到對的上的。如果沒有安裝，就看一下下面的jdk安裝方法：

/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

到這個地址下載jdk8版本，我下載的是jdk-8u73-linux-x64.tar.gz，然后解壓到/usr/local/jdk/下。

然后打開/etc/profile文件

[root@localhost~]#vim/etc/profile

把下面這段代碼寫到文件里

exportJAVA_HOME=/usr/local/jdk/jdk1.8.0_73

exportCLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar

exportPATH=$JAVA_HOME/bin:$PATH

最后

[root@localhost~]#source/etc/profile

這時jdk就生效了，可以使用java-version驗證下。

二.接下來安裝Kafka

1.下載Kafka

到/downloads.html下載相應(yīng)的版本，我使用的是kafka_2.9.1-.tgz。

2.下載完解壓到你喜歡的目錄

我是解壓到/usr/local/kafka/kafka_2.9.1-

3.運行默認(rèn)的Kafka

啟動Zookeeperserver

[root@localhostkafka_2.9.1-]#shbin/zookeeper-server-start.shconfig/perties

啟動Kafkaserver

[root@localhostkafka_2.9.1-]#shbin/kafka-server-start.shconfig/perties

運行生產(chǎn)者producer

[root@localhostkafka_2.9.1-]#shbin/kafka-console-producer.sh--broker-listlocalhost:9092--topictest

運行消費者consumer

[root@localhostkafka_2.9.1-]#shbin/kafka-console-consumer.sh--zookeeperlocalhost:2181--topictest--from-beginning

這樣，在producer那邊輸入內(nèi)容，consumer馬上就能接收到。

4.當(dāng)有跨機(jī)的producer或consumer連接時

需要配置config/perties的，要不然跨機(jī)的連不上。

三.Kafka-PHP擴(kuò)展

使用了一圈，就/nmred/kafka-php可以用。

我是使用composer安裝的，以下是示例：

producer.php

require'vendor/autoload.php';

while(1){

$part=mt_rand(0,1);

$produce=\Kafka\Produce::getInstance('kafka0:2181',3000);

//getavailablepartitions

$partitions=$produce-getAvailablePartitions('topic_name');

var_dump($partitions);

//sendmessage

$produce-setRequireAck(-1);

$produce-setMessages('topic_name',0,array(date('Y-m-dH: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($resultas$topicName=$partition){

foreach($partitionas$partId=$messageSet){

var_dump($partition-ge