异步 @KafkaListener 返回类型
从版本 3.2 开始，@KafkaListener（以及 @KafkaHandler）方法可以指定为异步返回类型，从而让回复被异步发送。返回类型包括 CompletableFuture<?>、Mono<?> 和 Kotlin suspend 函数。
@KafkaListener(id = "myListener", topics = "myTopic")
public CompletableFuture<String> listen(String data) {
...
CompletableFuture<String> future = new CompletableFuture<>();
future.complete("done");
return future;
}
@KafkaListener(id = "myListener", topics = "myTopic")
public Mono<Void> listen(String data) {
...
return Mono.empty();
}
当检测到异步返回类型时，AckMode 会被自动设置为 MANUAL 并启用乱序提交；取而代之的是，异步完成时会在异步操作完成后进行确认。当异步结果以错误完成时，消息是否可恢复取决于容器错误处理器。如果监听器方法内部发生了某些异常，导致无法创建异步结果对象，你必须捕获该异常并返回一个合适的返回对象，使消息被确认或恢复。
如果在一个具有异步返回类型（包括 Kotlin suspend 函数）的监听器上配置了 KafkaListenerErrorHandler，则在失败后会调用该错误处理器。有关此错误处理器及其用途的更多信息，请参阅处理异常。
对于像 DefaultErrorHandler 这样的“处理后查找”错误处理器，与阻塞监听器相比，异步监听器在同一分区上每个失败记录会多一次派发循环投递。派发循环会在任何异步失败回调触发之前，对每一条被 poll 的记录调用监听器，因此对于一次“毒丸”突发，第一次 poll 会对每条记录调用一次，然后才在分区上注册任何重试状态。此后，头部记录的后续重试周期会跳过排在它后面的记录，因此总调用次数与突发大小保持线性关系，而不是随积压深度增长。
对于单分区上 maxAttempts = n 的 N 条始终失败的记录，suspend 监听器总共收到 N * (n + 2) - 1 次调用，而阻塞监听器为 N * (n + 1) 次。iter-0 的派发调用无法被省略，否则就放弃了每次 poll 的并行性——而这正是异步监听器的意义所在。这不会影响阻塞监听器，因为同步抛出会在后续记录被调用之前中止批次迭代。
如果你的监听器在每次投递时执行非幂等工作（一次出站 HTTP 调用、一个不与偏移量提交构成事务的数据库写入），建议采用以下之一：
- 使用
@RetryableTopic（非阻塞重试）。失败记录被路由到单独的重试主题，因此主分区持续前进，没有记录会收到额外的派发循环投递。 - 使监听器幂等（Kafka 默认的至少一次投递本来就有此预期）。
这个选择可以按主题做出，而不是应用范围统一。那些正确性依赖严格按分区顺序的事件（例如以实体 id 为键的状态转换）可以继续使用基于查找的处理器，并依靠分区数加键分布来获得吞吐量。那些不要求顺序的事件（搜索索引更新、通知、下游富化）可以运行在 @RetryableTopic 监听器上，这样失败记录就不会收到额外的派发循环投递。
当同一分区上的记录结果混合时，异步返回类型也不会保持按分区的处理顺序：当一条失败记录处于其重试/退避周期时，同一分区上后续偏移量的成功记录可能会先完成其监听器主体。Kafka 仍然按顺序投递，已提交的偏移量保持一致，但监听器主体并发运行，因此副作用可能以非分区顺序落地。阻塞监听器避免了这一点，因为它严格一次一条记录地处理分区——较早偏移量的记录运行到完成（成功或恢复）之后，才调用分区上的下一条记录。如果你的消费者依赖按键的处理顺序（以实体 id 为键的幂等更新、有序状态转换），无论你使用哪种错误处理器，阻塞监听器都是正确的模型。