php5.X连接kafka

2019-02-12 14:27 

版权声明:本博客文章均有时效性,参考前请注意发表时间,以免耽误各位时间。欢迎转载。 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.username','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;
        }
}

执行就好

 

发表评论

您必须 登录 才能发表留言!