kafka安装及Kafka-PHP扩展的使用,kafkakafka-php扩展
2018-02-10 09:41
525 查看
http://www.bkjia.com/PHPjc/1100706.html
话说用了就要有点产出,要不然过段时间又忘了,所以在这里就记录一下试用Kafka的安装过程和php扩展的试用。实话说,如果用于队列的话,跟PHP比较配的,还是Redis。用的顺手,呵呵,只是Redis不能有多个consumer。但Kafka官方对PHP不支持,PHP扩展是爱好者或使用者写的。下面就开始讲Kafka的安装吧。我以CentOS6.4为例,64位。一. 首先确认下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对不上,那就装到对的上的。如果没有安装,就看一下下面的jdk安装方法:http://www.oracle.com/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把下面这段代码写到文件里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这时jdk就生效了,可以使用 java -version验证下。二. 接下来安装Kafka1. 下载Kafka到http://kafka.apache.org/downloads.html下载相应的版本,我使用的是kafka_2.9.1-0.8.2.2.tgz。2. 下载完解压到你喜欢的目录我是解压到 /usr/local/kafka/kafka_2.9.1-0.8.2.23. 运行默认的Kafka启动Zookeeper server[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/zookeeper-server-start.sh config/zookeeper.properties &启动Kafka server[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-server-start.sh config/server.properties &运行生产者producer[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test运行消费者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那边输入内容,consumer马上就能接收到。4. 当有跨机的producer或consumer连接时需要配置config/server.properties的host.name,要不然跨机的连不上。三. Kafka-PHP扩展使用了一圈,就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.phprequire '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());
}
}
话说用了就要有点产出,要不然过段时间又忘了,所以在这里就记录一下试用Kafka的安装过程和php扩展的试用。实话说,如果用于队列的话,跟PHP比较配的,还是Redis。用的顺手,呵呵,只是Redis不能有多个consumer。但Kafka官方对PHP不支持,PHP扩展是爱好者或使用者写的。下面就开始讲Kafka的安装吧。我以CentOS6.4为例,64位。一. 首先确认下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对不上,那就装到对的上的。如果没有安装,就看一下下面的jdk安装方法:http://www.oracle.com/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把下面这段代码写到文件里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这时jdk就生效了,可以使用 java -version验证下。二. 接下来安装Kafka1. 下载Kafka到http://kafka.apache.org/downloads.html下载相应的版本,我使用的是kafka_2.9.1-0.8.2.2.tgz。2. 下载完解压到你喜欢的目录我是解压到 /usr/local/kafka/kafka_2.9.1-0.8.2.23. 运行默认的Kafka启动Zookeeper server[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/zookeeper-server-start.sh config/zookeeper.properties &启动Kafka server[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-server-start.sh config/server.properties &运行生产者producer[root@localhost kafka_2.9.1-0.8.2.2]# sh bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test运行消费者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那边输入内容,consumer马上就能接收到。4. 当有跨机的producer或consumer连接时需要配置config/server.properties的host.name,要不然跨机的连不上。三. Kafka-PHP扩展使用了一圈,就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.phprequire '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());
}
}
相关文章推荐
- kafka安装及Kafka-PHP扩展的使用
- kafka安装及Kafka-PHP扩展的使用
- kafka安装及Kafka-PHP扩展的使用
- php安装使用http扩展类
- centos 使用phpize动态安装php扩展
- windows下redis的安装配置和php扩展使用phpredis
- PHP安装与使用VLD查看opcode代码【PHP安装第三方扩展的方法】
- 提高define性能的php扩展hidef的安装和使用
- 使用pecl安装php底层so扩展示例
- redis安装、配置、使用和redis php扩展安装教程
- php学习笔记(二十四)php使用memcache扩展库的使用和安装
- 使用phpize安装php扩展
- 安装php的php-protobu扩展及使用
- PHP使用phpize安装expect扩展
- Mac OS X 10.9自带php-fpm的配置使用和扩展安装
- PHP安装与使用VLD查看opcode代码【PHP安装第三方扩展的方法】
- 安装使用php zlib扩展实现页面gzip压缩输出实例
- 安装使用php zlib扩展实现页面gzip压缩输出实例
- php几个常见扩展的安装使用
- 安装和使用php的mcrypt扩展