Kafka之PHP擴展開發

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