0

您好，我在 php 中使用 Kafka。特别是这个库：\RdKafka。这是我的脚本，它工作正常，但我一个接一个地接收消息，但我希望得到 10 个 10 个。需要一些帮助来实现这一点，因为我是使用 Kafka 的新手。

消费者连接到 Kafka 并正确获取消息。

<?php
include_once("config.php");
const REQUEST_SLEEP_TIME = 12*1000; //12 seg
$conf = new RdKafka\Conf();
$conf->set("bootstrap.servers", $KAFKA_SOCKET);
$conf->set("group.id", "test-consumer-group");
$rk = new RdKafka\Consumer($conf);

$topicConf = new RdKafka\TopicConf();
$topicConf->set("request.required.acks", 1);
$topicConf->set("auto.commit.enable", 0);
$topicConf->set("auto.commit.interval.ms", 100);
 $topicConf->set("offset.store.method", "broker");
$topic = $rk->newTopic(KAFKA_TOPIC, $topicConf);
$topic->consumeStart(0, RD_KAFKA_OFFSET_END);

$i = 0;
while (true) {
    echo "start $i \n";
    $message = $topic->consume(0, REQUEST_SLEEP_TIME);

    if (is_null($message)) {
        sleep(1);
        echo "No more messages: ".date("H:i:s")."\n";
        continue;
    }

    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            echo "RD_KAFKA_RESP_ERR_NO_ERROR\n";
            print_r($message->payload."\n");
        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;
    }
    $i++;
    echo "end $i\n";
}
4

0 回答 0