NATS Consumer仅读取500条消息：如何获取所有Subject最后一条消息

你当前的实现仅执行了一次fetch操作，而NATS JetStream的fetch默认批量上限是500条，因此无法获取超过500条的消息。要实现无限制读取所有Subject的最后一条消息，需要循环执行fetch操作，直到没有新消息返回为止。

修改后的代码 #

val options = Options.builder()
    .server("nats://url:port")
    .authHandler(Nats.credentials("/Users/check/key.txt"))
    .build()
val connection = Nats.connect(options)
val js = connection.jetStream()
val consumerConfig = ConsumerConfiguration
    .builder()
    .filterSubject("subject.>")
    .deliverPolicy(DeliverPolicy.LastPerSubject)
    .build()

// 获取流上下文并创建/更新消费者
val streamContext: StreamContext = js.getStreamContext("STREAM_NAME")
val consumerContext = streamContext.createOrUpdateConsumer(consumerConfig)

val fetchConsumeOptions = FetchConsumeOptions.builder().noWait().build()
var totalCount = 0

try {
    while (true) {
        var batchCount = 0
        consumerContext.fetch(fetchConsumeOptions).use { fetchConsumer ->
            while (true) {
                val msg = fetchConsumer.nextMessage() ?: break
                batchCount++
                println("Message Received:")
                println(msg.subject)
                println(String(msg.data, StandardCharsets.UTF_8))
                msg.ack()
            }
        }
        // 如果当前批次没有获取到消息，说明所有消息已读取完成
        if (batchCount == 0) {
            break
        }
        totalCount += batchCount
    }
    println("Total messages received: $totalCount")
} catch (e: Exception) {
    e.printStackTrace()
} finally {
    // 关闭连接释放资源
    connection.close()
}

关键修改说明 #

• 循环执行fetch：在外层添加循环，每次fetch处理完当前批次的消息后，再次调用fetch获取下一批，直到某次fetch没有返回任何消息（batchCount == 0）。
• 保留LastPerSubject策略：该配置确保消费者只会获取每个匹配Subject的最后一条消息，不会重复读取同一Subject的历史消息。
• 添加连接关闭逻辑：在finally块中关闭NATS连接，避免资源泄漏。

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