问题描述
你好,我在 PHP 中使用 Kafka。特别是这个库:\RdKafka。这是我的脚本,它运行良好,但我正在一条一条地接收消息,但我希望收到 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";
}
解决方法
暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!
如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@)