如何获取Kafka Topic最新插入消息？该方案是否高效？

Hey there! Let's break down your Kafka question and solution step by step.

你的思路：手动获取偏移量再拉取 #

你的核心思路是完全正确的——先定位到分区的最新偏移量，再从该位置拉取消息。不过有几个关键细节需要调整，才能确保方案正常工作：

修正后的代码示例 #

val config = KafkaConfig()
val props = new Properties()
// ConsumerConfig配置
props.put("bootstrap.servers", config.bootstrapServers)
props.put("group.id", "stream-latest-consumer")
props.put(
  "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"
)
props.put(
  "value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"
)
// 手动控制偏移量时必须关闭自动提交
props.put("enable.auto.commit", "false")

val kafkaConsumer = new KafkaConsumer[String, String](props)
val targetPartition = new TopicPartition(config.topic, 0)
val partitions: util.Collection[TopicPartition] = List(targetPartition).asJava

// 第一步：给消费者分配目标分区（必须步骤，否则无法获取偏移量）
kafkaConsumer.assign(partitions)

// 第二步：获取分区的"下一条待写入消息"的偏移量
val offsetsMap: java.util.Map[TopicPartition, java.lang.Long] = kafkaConsumer.endOffsets(partitions)
val endOffset = offsetsMap.get(targetPartition)

// 第三步：定位到最新已存在的消息偏移量（endOffset是下一条待写入位置，所以要减1）
if (endOffset > 0) {
  kafkaConsumer.seek(targetPartition, endOffset - 1)
}

// 第四步：拉取最新消息
val records = kafkaConsumer.poll(Duration.ofMillis(1000))
for (record <- records.asScala) {
  println(s"最新消息：key=${record.key()}, value=${record.value()}, offset=${record.offset()}")
}

// 关闭消费者资源
kafkaConsumer.close()

更简化的实现：使用seekToEnd() #

Kafka Consumer提供了seekToEnd()方法，可以跳过手动调用endOffsets()的步骤，直接将消费者定位到分区的最新位置：

val config = KafkaConfig()
val props = new Properties()
props.put("bootstrap.servers", config.bootstrapServers)
props.put("group.id", "stream-latest-consumer")
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("enable.auto.commit", "false")

val kafkaConsumer = new KafkaConsumer[String, String](props)
val targetPartition = new TopicPartition(config.topic, 0)
val partitions: util.Collection[TopicPartition] = List(targetPartition).asJava

kafkaConsumer.assign(partitions)
// 直接定位到分区的最新待写入位置
kafkaConsumer.seekToEnd(partitions)
// 回退一个偏移量，获取最新已存在的消息
val currentOffset = kafkaConsumer.position(targetPartition)
if (currentOffset > 0) {
  kafkaConsumer.seek(targetPartition, currentOffset - 1)
}

val records = kafkaConsumer.poll(Duration.ofMillis(1000))
for (record <- records.asScala) {
  println(s"最新消息：key=${record.key()}, value=${record.value()}, offset=${record.offset()}")
}

kafkaConsumer.close()

你的方案是否高效？ #

你的核心方案（获取最新偏移量再拉取）是高效且合理的，原因如下：

• endOffsets()的性能开销极小：Kafka Broker会缓存每个分区的最新偏移量，这个请求的响应速度非常快，不会成为性能瓶颈。
• 逻辑直接：不需要额外的中间步骤，精准定位到目标位置拉取消息，避免了无效的消息遍历。

如果要进一步优化，可以注意两点：

1. 如果Topic有多个分区，需要遍历所有分区执行相同操作，才能获取全Topic的最新消息。
2. 如果是需要持续监听最新消息，可以在拉取完一次后保持消费者活跃，持续调用poll()获取新写入的消息，不需要每次都重新定位偏移量。

额外注意事项 #

• 消费者组隔离：使用独立的group.id（比如你用的stream-latest-consumer），避免和其他业务消费者的偏移量互相干扰。
• 空分区处理：如果分区还没有任何消息，endOffset会是0，此时不要执行seek操作，否则会抛出偏移量越界的异常。
• 资源释放：使用完消费者后一定要调用close()，避免占用Kafka集群的连接资源。

内容的提问来源于stack exchange，提问作者Rajeev