文档章节

Kafka之PHP扩展开发

hming
 hming
发布于 2017/03/27 17:11
字数 318
阅读 402
收藏 0

PHP扩展安装

要使用PHP作为kafka的producer,consumer者,必须使用php的扩展rdkafka,而rdkafka是依据librdkafka作为底层实现开发
出来的php扩展,那么在使用前我们需要先安装 
1.rdkafka的服务底层依赖: https://github.com/edenhill/librdkafka
2.rdkafka PHP拓展地址: https://github.com/arnaud-lb/php-rdkafka

librdkafka安装

git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure
make
make install

rdkafka安装

git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
/usr/local/php/bin/phpize
./configure --with-php-config=/usr/local/php/bin/php-config
make all -j 5
make install
# 在php.ini加入如下信息
vim /usr/local/php/etc/php.ini
extension=rdkafka.so
/usr/local/php/bin/php -m
查看使用有 rdkafka

php-kafka代码

produce编写

$rk = new RdKafka\Producer();
$rk->setLogLevel(LOG_DEBUG);
$rk->addBrokers("192.168.201.73");

$topic = $rk->newTopic("boyaa");

$topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message payload");

consumer编写

$conf = new RdKafka\Conf();

$conf->set('group.id', 'elk_consumer'); #定义消费组
$conf->set('broker.version.fallback', '0.8.2.2');  #重要 设置版本代理
$rk = new RdKafka\Consumer($conf); 

$rk->addBrokers("192.168.201.73");

$topicConf = new RdKafka\TopicConf();

$topicConf->set('auto.offset.reset', 'smallest');
$topicConf->set('auto.commit.enable', 'true');
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', '/data/wwwroot/test/data');
$topicConf->set('offset.store.sync.interval.ms', 0);

$topic = $rk->newTopic("boyaa", $topicConf);

// Start consuming partition 0
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

while (true) {
    $message = $topic->consume(0, 2000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            $message->payload;  #消息体 可处理相应业务
            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 "Timed out\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

© 著作权归作者所有

上一篇: UDP丢包问题分析
下一篇: ELK 环境部署详解
hming
粉丝 3
博文 88
码字总数 86648
作品 0
深圳
程序员
私信 提问
[转]使用PHP处理Kafka消息

Kafka 是一种高吞吐的分布式消息系统,能够替代传统的消息队列用于解耦合数据处理,缓存未处理消息等,同时具有更高的吞吐率,支持分区、多副本、冗余,因此被广泛用于大规模消息数据处理应用...

daos
2018/01/03
0
0
weiboad/kafka-php

Kafka-php English Document Kafka-php 使用纯粹的PHP 编写的 kafka 客户端,目前支持 0.8.x 以上版本的 Kafka,该项目 v0.2.x 和 v0.1.x 不兼容,如果使用原有的 v0.1.x 的可以参照文档 Ka...

weiboad
2017/04/28
0
0
PHP 编写的 kafka 客户端--kafka-php

Kafka-php 使用纯粹的 PHP 编写的 kafka 客户端,目前支持 0.8.x 以上版本的 Kafka,该项目 v0.2.x 和 v0.1.x 不兼容,如果使用原有的 v0.1.x 的可以参照文档 Kafka PHP v0.1.x Document, 不...

nmred_2008
2017/04/28
465
0
Kafka、RabbitMQ、RocketMQ消息中间件的对比

引言 分布式系统中,我们广泛运用消息中间件进行系统间的数据交换,便于异步解耦。现在开源的消息中间件有很多,目前对Kafka、RabbitMQ、RocketMQ这三个消息中间件做下对比分析。 - - kafka Ro...

黄文博
2018/08/22
0
0
kafka和flume的区别和对比使用

(1)kafka和flume都是日志系统。kafka是分布式消息中间件,自带存储,提供push和pull存取数据功能。flume分为agent(数据采集器),collector(数据简单处理和写入),storage(存储器)三部分...

u010963948
2017/12/13
0
0

没有更多内容

加载失败,请刷新页面

加载更多

c++队列(转自本人csdn)

#include using namespace std;int main(){char input, alphabet = ‘A’;cout << "输入最后一个大写字母: ";cin >> input;for(int i = 1; i <= (input-'A'+1); ++i)......

WinkJie
28分钟前
1
0
外观模式(Facade)

//这篇写的可以 https://www.jianshu.com/p/f71051475e32

南桥北木
47分钟前
2
0
锤子科技确认卖身,官网微博均变更所属公司

年前曾有过消息,罗永浩创办的锤子科技将被今日头条和抖音的母公司字节跳动收购。其中的部分工作人员将随专利一同成为字节跳动所属,而罗永浩本人将离开锤子科技。 今天我们终于看到了锤子科...

linux-tao
53分钟前
2
0
[json]阿里fastjson1.2.24字符串转为json对象/数组

java.lang.ClassCastException: com.alibaba.fastjson.JSONArray cannot be cast to com.alibaba.fastjson.JSONObject at com.alibaba.fastjson.JSON.parseObject(JSON.java:206) 多行要用J......

Danni3
今天
1
0
关于jenkins 构建打包的技巧

关于jenkins 构建打包的技巧 jenkins是一个专业构建平台,在各大公司内部比较受欢迎,可以自定义构建方式,以下整理一些技巧,防止以后忘记。 邮件通知 jenkins 自带邮件通知功能,但是比较单...

shzwork
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部