doukuo9116
2017-10-13 13:20
采纳率: 100%
浏览 2.1k
已采纳

如何使用php-rdkafka在kafka中确认消费消息?

我使用php-rdkafka作为php Kafka客户端,并使用测试组成功地生成了测试消息。我使用下面的代码来使用该消息:

$kafkaConsumer = new RdKafka\Consumer();
$kafkaConsumer->addBrokers("127.0.0.1:9292");
$topic = $kafkaConsumer->newTopic("test");
$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);

while (true) {
    $msg = $topic->consume(0, 1000);
    if($msg){
    if ($msg->err) {
        echo $msg->errstr(), "
";
        break;
    } else {
        echo $msg->payload, "
";
    }
  }
}

但是,当我再次尝试在测试组中设置消息并尝试使用测试组的消息时,我得到了旧消息和新消息。所以我只想知道如何才能确认旧的信息,这样我才能得到新的信息,而不是旧的信息?有人能给个建议吗??

我的kafka版本是0.11.0.1。

图片转代码服务由CSDN问答提供 功能建议

我正在使用 php-rdkafka 作为php kafka客户端。 我使用 test group成功生成我的测试消息。并使用下面的代码消费该消息,

  $ kafkaConsumer = new RdKafka \ Consumer()  ; 
 $ kafkaConsumer-> addBrokers(“127.0.0.1:9292”); 
 $ topic = $ kafkaConsumer-> newTopic(“test”); 
 $ topic-> consumeStart(0,RD_KAFKA_OFFSET_BEGINNING);  
 
而(true){
 $ msg = $ topic-> consume(0,1000); 
 if($ msg){
 if($ msg-> err){
 echo $ msg  - > errstr(),“
”; 
 break; 
} else {
 echo $ msg-> payload,“
”; 
} 
} 
} 
   
 
 

但是当我尝试再次在 test 组中设置消息并尝试使用 test 组的消息时,我就变老了 消息以及新消息。 所以我只想知道如何确认旧消息,这样我才能得到新消息而不是旧消息? 有人可以对此有所了解吗?

我的kafka版本是 0.11.0.1

  • 写回答
  • 好问题 提建议
  • 追加酬金
  • 关注问题
  • 收藏
  • 邀请回答

1条回答 默认 最新

  • douwuli4512 2017-10-22 14:15
    已采纳

    The method to acknowledge consumed messages in Kafka is to commit its offset. That way when restarting your consumer it can retrieve the last committed offset and restart where it left off.

    As suggested in the comments, you need to use RD_KAFKA_OFFSET_STORED to instruct the consumer to retrieve the stored offset.

    But you also need to provide a group name by setting the group.id config:

    <?php
    
    $conf = new RdKafka\Conf();
    
    // Set the group id. This is required when storing offsets on the broker
    $conf->set('group.id', 'myConsumerGroup');
    
    $rk = new RdKafka\Consumer($conf);
    $rk->addBrokers("127.0.0.1:9292");
    
    $topicConf = new RdKafka\TopicConf();
    $topicConf->set('auto.commit.interval.ms', 100);
    
    // Set where to start consuming messages when there is no initial offset in
    // offset store or the desired offset is out of range.
    // 'smallest': start from the beginning
    $topicConf->set('auto.offset.reset', 'smallest');
    
    $topic = $rk->newTopic("test", $topicConf);
    
    // Start consuming partition 0
    $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
    ";
                break;
            case RD_KAFKA_RESP_ERR__TIMED_OUT:
                echo "Timed out
    ";
                break;
            default:
                throw new \Exception($message->errstr(), $message->err);
                break;
        }
    }
    ?>
    
    评论
    解决 无用
    打赏 举报

相关推荐 更多相似问题