php搭建kafka队列_已解决_博问_博客园
首页
新闻
搜索
专区
学院
园子 · 关注 · 粉丝 ·
随便看看 · 消息
欢迎，请先 登录 或者 注册 。
登录 / 注册
闪存
博客
发言
小组
投递
新闻
提问
博问
添加
收藏
文库
问题反馈
问题列表
我关注的
我的博问
博问标签
php搭建kafka队列
悬赏园豆：180
[已解决问题]
解决于 2017-06-07 15:27
java 环境搭建 jdk :http://www.oracle.com/technetwork/java/javase/downloads/jdk7-downloads-1880260.htmljre:http://java.com/zh_CN/download/manual.jsp
下载kafka：http://apache.fayea.com/kafka/0.9.0.1/kafka_2.10-0.9.0.1.tgzkafka_2.10-0.9.0.1.tgz
1.启动zookeeper：（配置中端口默认为2181） bin/zookeeper-server-start.sh config/zookeeper.properties &
2.启动kafka：bin/kafka-server-start.sh config/server.properties
3.创建 topic创建一个叫做“test”的topic，它只有一个分区，一个副本。./bin/kafka-topics.sh -zookeeper localhost:2181 -topic test -replication-factor 1 -partitions 1 -create
4.可以通过list命令查看创建的topic:> bin/kafka-topics.sh --list --zookeeper localhost:2181
5.发送消息：bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
6.另开客户端获取消息bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginnin
bin/kafka-console-consumer.sh -zookeeper 172.16.1.197:2181 --from-beginning --topic rangeEvent38
7: 搭建一个多个broker的分布式集群（日志文件，端口，ip，连接zooker的ip+prot）cp config/server.properties config/server2.propertiescp config/server.properties config/server3.properties
修改：broker.id=2port=9292
8.php扩展安装安装基础库：librdkafka.tar.gz扩展：php-rdkafka.tar.gz （先make clean ，再删除configure，再phpize，因为此为别人安装过的文件，由于环境不同所以进行上述操作）安装提pcre错误，安装 yum -y install pcre pcre-devel
9.参数设置，如超出大小删除，指定保存几天的数据10.命令操作，删除topic
资源下载#libkafkalibrdkafka-master.zip1.01MBgit clone https://github.com/edenhill/librdkafka.git
#php-kafkaphp-rdkafka-master.zip63.65KBgit clone https://github.com/arnaud-lb/php-rdkafka.git
yum install cyrus-sasl-plain cyrus-sasl-devel cyrus-sasl-gssapiyum install krb5-workstation krb5-libs
修改/etc/krb5.conf文件
#yum -y install krb5*#yum install cyrus-sasl*
http://blog.csdn.net/dianyueneo/article/details/37527087https://cwiki.apache.org/confluence/display/KAFKA/Clients#Clients-PHPhttps://github.com/arnaud-lb/php-rdkafka
代码例子：读取：<?php$rk = new RdKafka\Consumer();$rk->setLogLevel(LOG_DEBUG);$rk->addBrokers('localhost:9092,localhost:9192,localhost:9292');$offset = 1;$topic = $rk->newTopic('test-x');$partition = 0;$topic -> consumeStart($partition,$offset);while (true) { $message = $topic->consume($partition, 100); if(!is_object($message)){ continue; } switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: print_r($message); break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: //echo "No more messages; will wait for more\n"; break; case RD_KAFKA_RESP_ERR__TIMED_OUT: //echo "Timedout\n"; break; default: throw new \Exception($message->errstr(), $message->err); break; }}
写入：<?php$rk = new RdKafka\Producer();$rk->setLogLevel(LOG_DEBUG);$rk->addBrokers('localhost:9092,localhost:9192,localhost:9292');$topic = $rk->newTopic('test-x');try{ $topic -> produce(0, 0, 'xxxxxxxxxx'.date('Y-m-d H:i:s',time()));}catch (Exception $e){ }
$rk = new RdKafka\Consumer();$rk->setLogLevel(LOG_DEBUG);$rk->addBrokers('172.16.2.95:9092');//brokerList ip 列表$dsp = new posteventv2();$topic = $rk->newTopic('classline_alarm_runline');//topic$queue = $rk->newQueue();for($partition = 0;$partition<4;$partition++){//分区个数，这里服务端开了4个 //开始订阅数据 $topic -> consumeQueueStart($partition,RD_KAFKA_OFFSET_END,$queue); //RD_KAFKA_OFFSET_END 永远都从最新的开始取数据}while (true) { $message = $queue->consume(100); if(!is_object($message)){ continue; } switch ($message->err) { case RD_KAFKA_RESP_ERR_NO_ERROR: tools::datalog(var_export($message,true),'classline_alarm_runline');//获取到的数据，订阅方根据自己逻辑自己处理 break; case RD_KAFKA_RESP_ERR__PARTITION_EOF: break; case RD_KAFKA_RESP_ERR__TIMED_OUT: break; default: throw new \Exception($message->errstr(), $message->err); break; }}
kafka
芒芒芒芒果
初学一级
园豆：24
提问于：2017-06-07 15:23
显示帮助
使用"Ctrl+Enter"可进行快捷提交，评论支持部分 Markdown 语法：[link](http://example.com) _italic_ **bold** `code`。
分享
分享您的问题
最佳答案
你这个是kafka，挺吊啊
收获园豆：180
风行天下12
老鸟四级
|园豆：3862
2017-06-07 15:26
编辑文本
预览
上传图片
Ctrl+Enter键快速提交
其他回答(1)
以上，做个记录
芒芒芒芒果
园豆：24
(初学一级)
2017-06-07 15:25
编辑文本
预览
上传图片
Ctrl+Enter键快速提交
清除回答草稿
您需要登录以后才能回答，未注册用户请先注册。
关于博客园联系我们广告合作沪ICP备09004260号-1©2004-2022博客园Powered by .NET on Kubernetes