设为首页 加入收藏

TOP

php5.X连接kafka
2019-02-12 14:27:37 】 浏览:107
Tags:php5.X 连接 kafka
版权声明:本博客文章均有时效性,参考前请注意发表时间,以免耽误各位时间。欢迎转载。 https://blog.csdn.net/q1177660557/article/details/81219664

kafka扩展:https://github.com/arnaud-lb/php-rdkafka

测试版本:php-5.5.25/30

安装:

1、先安装librdkafka:

git clone https://github.com/edenhill/librdkafka/

cd librdkafka

./configure && make && make install

2、安装php-rdkafka扩展:

git clone https://github.com/arnaud-lb/php-rdkafka

cd php-rdkafka

phpize && ./configure --with-php-config=/usr/local/php/bin/php-config && make all -j 5 && make install

3、测试代码,用的是他的代码改了改,加了些sasl鉴权的东西:

<php

$conf = new RdKafka\Conf();

// Set the group id. This is required when storing offsets on the broker
$conf->set('group.id', 'vcs');
$conf->set('security.protocol','SASL_PLAINTEXT');
$conf->set('sasl.mechanism','PLAIN');
$conf->set('sasl.usern
		    

ame','hello'); $conf->set('sasl.password','world'); $rk = new RdKafka\Consumer($conf); $rk->addBrokers("mq.cloud.com:9012"); var_dump($rk); //$conf->set('group.id', 'vcs'); $topicConf = new RdKafka\TopicConf(); $topicConf->set('auto.commit.interval.ms', 100); // Set the offset store method to 'file' $topicConf->set('offset.store.method', 'file'); $topicConf->set('offset.store.path', sys_get_temp_dir()); // Alternatively, set the offset store method to 'broker' // $topicConf->set('offset.store.method', 'broker'); // Set where to start consuming messages when there is no initial offset in // offset store or the desired offset is out of range. // 'smallest': start from the beginning $topicConf->set('auto.offset.reset', 'smallest'); $topic = $rk->newTopic("tv.tpvcs.setvcs", $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 "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; } }

执行就好


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇《Apache Kafka 实战》笔记 - 10... 下一篇Kafka 监控

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(217) }