php-rdkafka 扩展安装_qq_20376949的博客-程序员秘密

php有两种方式调用kafka

一、php-rdkafka

文档地址：https://arnaud-lb.github.io/php-rdkafka/phpdoc/book.rdkafka.html
rdkafka安装需要依赖librdkafka所以我们需要先安装librdkafka
下载地址http://pecl.php.net/package/rdkafka

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

安装php-rdkafka扩展

git clone https://github.com/arnaud-lb/php-rdkafka.git
cd php-rdkafka
phpize
./configure --with-php-config=/usr/local/php7.0/bin/php-config
make && make install

然后在php.ini写入

extension = redis.so

image.png

二、kafka-php 扩展包

文档地址：https://github.com/weiboad/kafka-php

三、简单示例

生成者

<?php

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

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

for ($i = 0; $i < 10; $i++) {
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "发送信息： $i");
    $rk->poll(0);
}

while ($rk->getOutQLen() > 0) {
    $rk->poll(50);
}

?>

消费者

<?php

$conf = new RdKafka\Conf();

$conf->set('group.id', 'myConsumerGroup');

$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("192.168.2.150:9092");

$topicConf = new RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 100);
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', sys_get_temp_dir());
$topicConf->set('auto.offset.reset', 'smallest');

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

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

while (true) {
    $message = $topic->consume(0, 120*10000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
        //没有错误打印信息
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "等待接收信息\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "超时\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

?>

image.png

本文链接：https://blog.csdn.net/qq_20376949/article/details/84868097

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

linux命令执行技巧，转后台执行，顺序执行_weixin_34159110的博客-程序员秘密

来源：http://os.51cto.com/art/200908/143126_all.htm快捷键：Linux命令行编辑快捷键：history 显示命令历史列表↑(Ctrl+p) 显示上一条命令↓(Ctrl+n) 显示下一条命令!num 执行命令历史列表的第num条命令!! 执行上一条命令!?string? 执行含有string字符串...

Qt半透明对话框_sunxiaopengsun的博客-程序员秘密

// 设置整个面板为透明QDialog dailog;dailog.setWindowFlags(Qt::FramelessWindowHint); // 去除页面边框dailog.setFixedSize(150, 100);QPalette pl = dailog.palette();pl.setColor(QPalette::Window, QColor(0, 0, 0, 200)); // 设置背景颜色为黑色，如果不设置默认为白色dailog.setPalette(.

vue 项目安装_vuecli-agan_aganliang的博客-程序员秘密

1.搭建前端服务器node.js安装参考：https://blog.csdn.net/aganliang/article/details/103228597cnpm install -g vue-cli把vue命令添加到bin路径ln -s /root/node-v12.14.0-linux-x64/lib/node_modules/vue-cli/bin/vue /u...

mysql提供了()种整型_mysql unsigned 用法及相减出现补数溢出解决方法_小仙女CiCi的博客-程序员秘密

unsigned 既为非负数，用此类型可以增加数据长度!例如如果 tinyint最大是127，那 tinyint unsigned 最大就可以到 127 * 2unsigned 属性只针对整型，而binary属性只用于char 和varchar。类型说明tinyint非常小的整数smallint较小整数mediumint中等大小整数int标准整数bigint较大整数float单精度浮点数doub...

VMware Workstation 15 Player虚拟机CentOS-7-x86_64-DVD-2009 系统卸载openjdk 安装 jdk-8u321-linux-x64.tar.gz_创无意的博客-程序员秘密

1.1 查看现有版本查看系统中是否安装jdk：java -version看到当前系统包含 openjdk 1.8 .1.2 查询所有与java相关的安装的rpm包rpm -qa|grep java1.3 删除除了noarch 结尾的所有文件rpm -e --nodeps java-1.8.0-openjdk-headless-1.8.0.312.b07-1.el7_9.x86_64rpm -e --nodeps java-1.8.0-openjdk-1.8.0.3

FFmpeg完美入门【6】-FFmpeg架构之I\O模块分析_ffplay typedef int (*player_init) (long window_han_matthew0618band的博客-程序员秘密

1概述ffmpeg项目的数据IO部分主要是在libavformat库中实现，某些对于内存的操作部分在libavutil库中。数据IO是基于文件格式（Format）以及文件传输协议(Protocol)的，与具体的编解码标准无关。 ffmpeg工程转码时数据IO层次关系如图所示：对于上面的数据IO流程，具体可以用下面的例子来说明，我们从一个http服务器获取音视频数据，格式是flv的，需要

随便推点

ubuntu 2.6.32.21内核树搭建_ab198604的博客-程序员秘密

　　什么是内核树？刚开始我也没弄明白，通过这几天的学习，有所感悟，就说说我的理解吧！从形式上看，内核树与内核源码的目录结构形式是相同的，都是由各个层次的文件目录结构组成，但是其中的具体内容肯定是不同的。从功能上看，内核树中含有编写或编译Kernel程序或驱动时所需要的一些内核函数库以及定义好的一些数据结构，数据类型等，说白了，当你在进行kernel编程或者是编写linux设备驱动时都需要用到这里面的一些资源，并且我们写好的驱动程序时，也需要用到内核树来编译，因此从这个角度上来看，内核树也可以看作是一个编译器