如何在Spring Boot控制器中获取Kafka最新消息？

针对你需要从消费者端（或直接从Kafka集群）获取最新消息、无需关注offset的需求，下面是几个实用的实现方案，支持Kotlin/Java：

方案一：临时消费者直接拉取最新消息 #

每次调用接口时，创建一个临时Kafka消费者，配置为从最新位置开始拉取消息，获取后就关闭（或复用实例）。这种方式不需要维护offset，完全按需获取。

Kotlin代码示例：

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer
import java.time.Duration
import java.util.Properties

fun getLatestMessage(topic: String): String? {
    val props = Properties().apply {
        put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092")
        put(ConsumerConfig.GROUP_ID_CONFIG, "temp-latest-fetcher") // 用独立临时组ID，不干扰现有消费组
        put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest") // 直接从分区最新位置开始拉取
        put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false") // 不提交offset，避免留下消费痕迹
    }

    // use块自动关闭消费者
    KafkaConsumer<String, String>(props).use { consumer ->
        consumer.subscribe(listOf(topic))
        // 短时间拉取一次，避免长时间阻塞
        val records = consumer.poll(Duration.ofMillis(1000))
        // 返回拉取到的最后一条消息（最新的）
        return records.lastOrNull()?.value()
    }
}

注意：频繁创建消费者会有一定性能开销，如果接口调用频繁，可以考虑把消费者做成单例复用，每次调用前调用consumer.seekToEnd(emptyList())重置到最新位置。

方案二：在现有消费者中维护最新消息缓存 #

如果你的消费者一直在实时消费消息，可以在消费逻辑里把最新收到的消息缓存到内存（比如用原子类保证线程安全），接口直接读取这个缓存。这种方式性能最高，接口直接读内存。

Kotlin代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer
import java.time.Duration
import java.util.Properties
import java.util.concurrent.atomic.AtomicReference

// 用原子类缓存最新消息，保证多线程安全
private val latestMessageCache = AtomicReference<String?>()

// 现有消费者启动逻辑
fun startExistingConsumer(topic: String) {
    val props = Properties().apply {
        put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092")
        put(ConsumerConfig.GROUP_ID_CONFIG, "your-existing-consumer-group")
        put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    }

    KafkaConsumer<String, String>(props).use { consumer ->
        consumer.subscribe(listOf(topic))
        while (true) {
            val records = consumer.poll(Duration.ofMillis(100))
            records.forEach { record ->
                // 原有的消息处理逻辑
                processBusinessLogic(record)
                // 更新最新消息缓存
                latestMessageCache.set(record.value())
            }
        }
    }
}

// 对外暴露的获取最新消息接口
fun getLatestMessageFromCache(): String? {
    return latestMessageCache.get()
}

private fun processBusinessLogic(record: ConsumerRecord<String, String>) {
    // 你的业务处理代码
}

注意：如果消费者挂掉或者暂停消费，缓存里的消息就不会更新，而且只能拿到消费者已经消费过的最新消息，不适合消费者落后于集群最新offset的场景。

方案三：通过Admin API定位最新消息 #

先借助Kafka Admin API获取每个分区的最新offset（即下一条要写入的位置），然后让消费者定位到这个offset的前一个位置（因为最新offset是未写入的位置，前一个才是最后一条已写入的消息），再拉取这条消息。这种方式能拿到集群中真正的最新消息，不受现有消费者状态影响。

Kotlin代码示例：

import org.apache.kafka.clients.admin.AdminClient
import org.apache.kafka.clients.admin.ListOffsetsResult
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import java.time.Duration
import java.util.Properties

fun getLatestClusterMessage(topic: String): String? {
    val adminProps = Properties().apply {
        put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092")
    }

    AdminClient.create(adminProps).use { admin ->
        // 获取目标主题的所有分区
        val topicMetadata = admin.describeTopics(listOf(topic)).all().get()[topic] ?: return null
        val partitions = topicMetadata.partitions().map { TopicPartition(topic, it.partition()) }

        // 获取每个分区的最新offset（下一条待写入的位置）
        val offsetMap = partitions.associateWith { ListOffsetsResult.ListOffsetsSpec.latest() }
        val latestOffsets = admin.listOffsets(offsetMap).all().get()

        val consumerProps = Properties().apply {
            put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092")
            put(ConsumerConfig.GROUP_ID_CONFIG, "temp-admin-fetcher")
            put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
            put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
            put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
        }

        KafkaConsumer<String, String>(consumerProps).use { consumer ->
            consumer.assign(partitions)
            var latestMsg: String? = null
            for ((tp, offsetMeta) in latestOffsets) {
                val targetOffset = offsetMeta.offset() - 1
                if (targetOffset >= 0) {
                    // 定位到最后一条已写入消息的位置
                    consumer.seek(tp, targetOffset)
                    // 拉取这条消息
                    val records = consumer.poll(Duration.ofMillis(500))
                    records.firstOrNull()?.let { latestMsg = it.value() }
                }
            }
            return latestMsg
        }
    }
}

注意：这个方案需要你的应用有Kafka集群的Admin权限，步骤相对繁琐，但能保证拿到绝对最新的消息。

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