PHP rdKafka 安装和基本使用

安装，通过安装扩展即可，不需要额外新增源码Installation (le-blanc.net)生产者demo：<?php$conf = new RdKafka\Conf();$conf->set('metadata.broker.list', 'localhost:9092');$producer = new RdKafka\Producer($conf);$topic = $pro

潘广宇

1566人浏览 · 2022-01-21 11:27:05

潘广宇 · 2022-01-21 11:27:05 发布

安装，通过安装扩展即可，不需要额外新增源码

Installation (le-blanc.net)

生产者demo：

<?php

$conf = new RdKafka\Conf();
$conf->set('metadata.broker.list', 'localhost:9092');

$producer = new RdKafka\Producer($conf);

$topic = $producer->newTopic("test");

for ($i = 0; $i < 10; $i++) {
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message $i");
    $producer->poll(0);
}

for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
    $result = $producer->flush(10000);
    if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
        break;
    }
}

if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
    throw new \RuntimeException('Was unable to flush, messages might be lost!');
}

?>

消费者demo：

<?php

$conf = new RdKafka\Conf();
$conf->set('group.id', 'myConsumerGroup');

$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', 'earliest');

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

?>

更好的方案：封装一个 KafkaClient，详见 Thinkphp RdKafka 公共类封装_技术博主-CSDN博客