我尝试在 scala 中实现一个非常简单的 Kafka (0.9.0.1) 消费者(代码如下)。
据我了解,Kafka(或者更确切地说是 Zookeeper)为每个 groupId 存储给定主题的最后一条消费消息的偏移量。所以给定以下场景:
- Consumer with
groupId1
昨天消费了唯一的 5 主题中的消息。现在最后消费的消息有偏移量 4(考虑到 偏移量为 0) 的第一条消息 - 晚上有 2 条新消息到达该主题
- 今天重启消费者,使用相同的
groupId1
,会有 有两种选择:
选项 1:如果我将以下属性设置为 "latest"
,消费者将阅读夜间到达的最后 2 条新消息:
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
选项 2:如果我将以下属性设置为 "earliest"
,消费者将阅读主题中的所有 7 条消息:
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
问题:出于某种原因,如果我将消费者的 groupId 更改为 groupId2
,这是给定主题的新 groupId,因此它从未消费过任何消息之前,它的最新偏移量应该是 0。我期待通过设置
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
消费者将在第一次执行期间读取主题中存储的所有消息(相当于最早)。然后对于接下来的执行,它将只消耗新的。然而,事实并非如此。
如果我设置一个新的 groupId
并将 AUTO_OFFSET_RESET_CONFIG
保持为 latest
,消费者将无法阅读任何消息。然后我需要做的是在第一次运行时将 AUTO_OFFSET_RESET_CONFIG
设置为 earliest
,一旦 groupID 已经有一个不同于 0 的偏移量,我就可以移动到 最新的
。
我的消费者应该是这样的吗?有没有比在我第一次运行消费者后切换 AUTO_OFFSET_RESET_CONFIG
更好的解决方案?
下面是我作为一个简单的消费者使用的代码:
class KafkaTestings {
val brokers = "listOfBrokers"
val groupId = "anyGroupId"
val topic = "anyTopic"
val props = createConsumerConfig(brokers, groupId)
def createConsumerConfig(brokers: String, groupId: String): Properties = {
val props = new Properties()
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000")
props.put(ConsumerConfig.CLIENT_ID_CONFIG, "12321")
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer")
props
}
def run() = {
consumer.subscribe(Collections.singletonList(this.topic))
Executors.newSingleThreadExecutor.execute( new Runnable {
override def run(): Unit = {
while (true) {
val records = consumer.poll(1000)
for (record <- records) {
println("Record: "+record.value)
}
}
}
})
}
}
object ScalaConsumer extends App {
val testConsumer = new KafkaTestings()
testConsumer.run()
}
This被用作编写这个简单消费者的引用
最佳答案
这是按照记录工作的。
如果你开始一个新的消费者组(即 Kafka 中没有存储现有偏移量的消费者组),你必须选择消费者是否应该从最早的可能消息开始(主题中仍然可用的最旧消息)或来自最新的(仅从现在开始产生的消息)。
Is there a better solution than switching the AUTO_OFFSET_RESET_CONFIG after the first time I run the consumer?
您可以将它保持在 EARLIEST,因为当您第二次运行消费者时,它已经存储了偏移量,只需从那里获取。重置策略仅在创建新的消费者组时使用。
Today I restart the consumer, with the same groupId1, there will be two options:
不是真的。由于消费者组在前一天运行,它将找到其提交的偏移量并从中断的地方继续。因此,无论您将重置策略设置为什么,它都会收到这两条新消息。
虽然知道,Kafka 不会永远存储这些偏移量,但我相信默认值只是一周。因此,如果您关闭消费者的时间超过这个时间,偏移量可能会过时,并且您可能会意外重置为 EARLIEST(这对于大型主题来说可能代价高昂)。鉴于此,无论如何将其更改为 LATEST 可能是谨慎的做法。
关于scala - Kafka - 为什么在将 AUTO_OFFSET_RESET_CONFIG 设置为 "latest"时,新的 groupId 不返回主题中的所有消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57195132/