PHP 消息队列 Kafka 使用(2)

当前位置:
首页 > PHP >

PHP 消息队列 Kafka 使用(2)
创建一个消费类
<?php

class KafkaConsumer

{

    public static $brokerList = '127.0.0.1:9092';

      public static function consumer()

    {

        $conf = new \RdKafka\Conf();

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

        $rk = new \RdKafka\Consumer($conf);

   $rk->addBrokers("127.0.0.1");

        $topicConf = new \RdKafka\TopicConf();

        $topicConf->set('auto.commit.interval.ms', 100);

        $topicConf->set('offset.store.method', 'broker');

        $topicConf->set('auto.offset.reset', 'smallest');

        $topic = $rk->newTopic('test', $topicConf);

        $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;

            }

        }

    }

}
栏目列表