Kafka不节省抵消如果消耗短时间

Question

具有特定组id的使用者连接到代理，监听主题不到1分钟，然后断开连接（根据业务逻辑）。当它监听主题时，它可以使用一些消息。当同一个使用者重复这个动作时，它会使用相同的消息！

我发现Kafka用间隔1分钟保存偏移。这意味着消费者必须听超过1分钟的主题。我怎样才能缩短这个间隔？

我发现了这样的属性：

null

log.flush.offset.checkpoint.interval.ms=6000
log.flush.start.offset.checkpoint.interval.ms=6000
offset.flush.interval.ms=6000

null

<?php
$topicConf = new \RdKafka\TopicConf();
$topicConf->set('auto.offset.reset', 'smallest');

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

$kafkaConsumer = new \RdKafka\Consumer($conf);
$kafkaConsumer->addBrokers('queue.a:9092');
$kafkaConsumer->setLogLevel(LOG_DEBUG);

$topicConf = new \RdKafka\TopicConf();
$topicConf->set('auto.offset.reset', 'smallest');

$queue = $kafkaConsumer->newQueue();
$topic = $kafkaConsumer->newTopic('topic_name', $topicConf);
$topic->consumeQueueStart(0, \RD_KAFKA_OFFSET_STORED, $queue);

while (true) {
    $msg = $queue->consume(2000);
    if ($msg !== null) {
        var_dump($msg);
    }
}

Answer 1

您应该尝试显式提交使用者中的偏移量：

在消费者中显式提交偏移量如果使用自动偏移量提交，则不需要担心显式提交偏移量。但是，如果您决定需要对偏移量提交的时间进行更多的控制，那么您确实需要考虑如何提交偏移量--或者是为了最小化重复，或者是因为您在主使用者轮询循环之外进行事件处理。

摘自Kafka权威指南，第127页。（这是一本免费的电子书，你可以下载）

while (true) {
    $msg = $queue->consume(2000);
    if ($msg !== null) {
        var_dump($msg);
        $kafkaConsumer->commit($msg);
    }
}