文档章节

php连接kafka

 果树啊
发布于 2017/06/30 10:54
字数 759
阅读 539
收藏 0

安装kafka

yum search jdk
yum -y install java-1.8.0-openjdk.i686 java-1.8.0-openjdk-devel.i686
wget http://mirror.its.dal.ca/apache/kafka/1.0.0/kafka_2.11-1.0.0.tgz
tar -xzvf kafka_2.11-1.0.0.tgz
cd kafka_2.11-1.0.0/bin
#启动zookeeper
./zookeeper-server-start.sh  ../config/zookeeper.properties  &
#启动kafka
 ./kafka-server-start.sh ../config/server.properties &
ps -ef | grep kafka




创建一个叫"test1234"的topic,它只有一个分区,一个副本:
 ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1234

可以用list查看创建的topic,当前创建了4个topic
./kafka-topics.sh --list --zookeeper localhost:2181

发送消息。运行producer并在控制台中输一些消息,这些消息将被发送到服务端
./kafka-console-producer.sh --broker-list localhost:9092 --topic test1234

开启consumer,可以读取到刚才发出的消息并输出。
 ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1234 --from-beginning

安装可视化工具

wget http://ftp.cuhk.edu.hk/pub/packages/apache.org/maven/maven-3/3.5.4/binaries/apache-maven-3.5.4-bin.zip
unzip apache-maven-3.5.4-bin.zip
mv apache-maven-3.5.4 apache-maven
vim /etc/profile
export MAVEN_HOME=/root/dev/apache-maven-3.5.4
export PATH=$PATH:$MAVEN_HOME/bin
source /etc/profile
wget https://github.com/HomeAdvisor/Kafdrop/archive/kafdrop-2.0.0.zip
unzip kafdrop-2.0.0.zip
cd kafdrop
mvn clean package
java -jar ./target/kafdrop-2.0.0.jar --zookeeper.connect=127.0.0.1:2181 --server.port=9999

安装扩展

wget https://github.com/edenhill/librdkafka/archive/master.zip
unzip master.zip
cd librdkafka-master/
./configure
make
sudo make install
git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
phpize
./configure --with-php-config=/usr/local/php/bin/php-config 
make all -j 5
sudo make install
vi /usr/local/php/etc/php.ini
extension=rdkafka.so

生产者

$conf = new \RdKafka\Conf();
$this->conf->set('message.send.max.retries', 3);
        $conf->set('socket.timeout.ms', 50);//网络请求的超时时间,默认60s,改成50ms
        $conf->set('request.timeout.ms', 20000);//生产者请求超时时间,默认5s改成10s
        $conf->set('topic.metadata.refresh.sparse', true);//仅获取自己用的元数据 减少带宽
        $conf->set('topic.metadata.refresh.interval.ms', 600000);//设置刷新元数据时间间隔为600s 减少带宽
        $conf->set('log.connection.close', 'false');
        if (function_exists('pcntl_sigprocmask')) {
            pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
            $this->conf->set('internal.termination.signal', SIGIO);//设置kafka客户端线程在其完成后立即终止
        } else {
            $this->conf->set('queue.buffering.max.ms', 1);//确保消息尽快发送
        }

$conf->setDrMsgCb(function ($kafka, $message) {//每次都会调用
  if ($message->err) {
    // message permanently failed to be delivered
    $error_info = rd_kafka_err2str ($message->err);
  } else {
    // message successfully delivered
    print_r($message);
  }
});
        $conf->setErrorCb(function ($kafka, $err, $reason) {//发送失败后调用
            printf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason);
        });
        $rk = new \RdKafka\Producer($conf);
        $rk->setLogLevel(LOG_DEBUG);
        $rk->addBrokers('IP:PORT');
        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('message.timeout.ms', 100);//设置超时时间避免发送不成功长期堵住
        $topic = $rk->newTopic('TOPIC_ID',$topicConf);
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, 'message');
$rk->poll(0);//立即调用DrMsgCb

消费者

<?php
$config = require('config.php');
if(!extension_loaded('Rdkafka')){
    //如果扩展没有被加载
    file_put_contents("./extension.txt", "RdKafka extension is not load",FILE_APPEND);
}
$conf = new \RdKafka\Conf();
$conf->setDrMsgCb(function ($kafka, $message) {
    if ($message->err) {
        $error_info = rd_kafka_err2str ($message->err);
        echo $error_info;
    } else {
    }
});
$conf->setErrorCb(function ($kafka, $err, $reason) {
    printf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason);
});
$rk = new \RdKafka\Consumer($conf);
$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers($config['kafka']['consumer']['brokers']);
$topicConf = new \RdKafka\TopicConf();
$topicConf->set('auto.offset.reset', 'smallest');
$topicConf->set('auto.commit.enable', true);
$topicConf->set('offset.store.path', './kafka_offset');
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.sync.interval.ms', 0);
$topic = $rk->newTopic($config['kafka']['consumer']['topic'], $topicConf);
do{
    $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
    $msg = $topic->consume(0, 1000);
    if($msg != null && $msg->err == 0){
        $data = $msg->payload;
    }
    $topic->consumeStop(0);
}while(true);
return array(
    'kafka' => array(
        'produce' => array(
            'brokers' => '127.0.0.1:9092',
            'topic' => '1'
        ),
        'consumer' => array(
            'brokers' => '127.0.0.1:9092',
            'topic' => '1'
        )
    ),
);

可配置属性

https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

文档

https://arnaud-lb.github.io/php-rdkafka/phpdoc/book.rdkafka.html

可视化工具https://github.com/HomeAdvisor/Kafdrop

shell命令 http://www.cnblogs.com/xiaodf/p/6093261.html#3

 

© 著作权归作者所有

上一篇: swoole
粉丝 12
博文 247
码字总数 55975
作品 0
福州
高级程序员
私信 提问
简单舒适的 PHP-KafKa 拓展--See-KafKa

[喵咪KafKa(3)]PHP拓展See-KafKa 前言 (Simple 简单 easy 容易 expand 的拓展) KafKa是由Apache基金会维护的一个分布式订阅分发系统,KafKa它最初的目的是为了解决,统一,高效低延时,高通量(同...

喵了_个咪
2016/09/27
2.9K
1
喵了个咪/See-KafKa

#See-KafKa 简单舒适的PHP-KafKa拓展 ##前言 (Simple 简单 easy 容易 expand 的拓展) KafKa是由Apache基金会维护的一个分布式订阅分发系统,KafKa它最初的目的是为了解决,统一,高效低延时,高通...

喵了个咪
2016/09/27
0
0
[喵咪KafKa(3)]PHP拓展See-KafKa

[喵咪KafKa(3)]PHP拓展See-KafKa 前言 (Simple 简单 easy 容易 expand 的拓展) KafKa是由Apache基金会维护的一个分布式订阅分发系统,KafKa它最初的目的是为了解决,统一,高效低延时,高通量(同...

喵了_个咪
2016/09/27
888
1
【官方速报】360开源又一力作——KafkaBridge:让操作kafka更简单!

女主宣言 KafkaBridge 封装了对Kafka集群的读写操作,接口极少,简单易用,稳定可靠,支持c++/c、php、python、golang等多种语言,并特别针对php-fpm场景中作了长连接复用的优化,已在360公司...

ZVAyIVqt0UFji
2018/10/12
0
0
Apache Kafka 客户端 SDK - KafkaBridge

KafkaBridge 是奇虎 360 开源的 Kafka 客户端 SDK ,底层基于 librdkafka ,与之相比封装了大量的使用细节,简单易用,使用者无需了解过多的 Kafka 系统细节,只需调用极少量的接口,就可完成...

匿名
2018/10/16
1K
0

没有更多内容

加载失败,请刷新页面

加载更多

刚哥谈架构 (二) 我眼中的架构师

之前在公司,有小伙伴在向别人介绍我的时候,经常会有人这么说:“刚哥是我们的architcture”,如果来人是老外,心中一定是一惊,心中暗叹,“这位匪首看上去貌不惊人,难道已经做到了架构和...

naughty
39分钟前
3
0
OSChina 周日乱弹 —— 别问,问就是没空

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @tom_tdhzz :#今日歌曲推荐# 分享容祖儿/彭羚的单曲《心淡》: 《心淡》- 容祖儿/彭羚 手机党少年们想听歌,请使劲儿戳(这里) @wqp0010 :周...

小小编辑
今天
109
4
golang微服务框架go-micro 入门笔记2.1 micro工具之micro api

micro api micro 功能非常强大,本文将详细阐述micro api 命令行的功能 重要的事情说3次 本文全部代码https://idea.techidea8.com/open/idea.shtml?id=6 本文全部代码https://idea.techidea8....

非正式解决方案
今天
5
0
Spring Context 你真的懂了吗

今天介绍一下大家常见的一个单词 context 应该怎么去理解,正确的理解它有助于我们学习 spring 以及计算机系统中的其他知识。 1. context 是什么 我们经常在编程中见到 context 这个单词,当...

Java知其所以然
昨天
5
0
Spring Boot + Mybatis-Plus 集成与使用(二)

前言: 本章节介绍MyBatis-Puls的CRUD使用。在开始之前,先简单讲解下上章节关于Spring Boot是如何自动配置MyBatis-Plus。 一、自动配置 当Spring Boot应用从主方法main()启动后,首先加载S...

伴学编程
昨天
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部