@KafkaListener注解
@KafkaListener(topics = "${xxxx}", containerFactory = "xxx")
public void consume(String message) {
service.save(xx);
}
在 Spring Boot 中，使用 @KafkaListener 注解消费 Kafka 消息时，如果 service.save(cleanWikiSource) 抛出异常，消息不会被认为已经成功消费完毕。相应的行为取决于 Kafka 配置的 ack 模式以及错误处理机制：
默认情况（ack 模式为 BATCH 或 RECORD）：
如果 service.save(cleanWikiSource) 抛出异常，Kafka 会自动将消息标记为未成功消费，并会重新尝试交付此消息（即消息会被重新消费）。
如果配置了 max.poll.interval.ms 或 max.poll.records 等参数，可以控制消费者重试间隔和次数。
手动确认（ack 模式为 MANUAL 或 MANUAL_IMMEDIATE）：
如果您启用了手动提交确认（通过 KafkaListener 配置 ackMode = MANUAL），则在成功处理完消息后需手动调用 Acknowledgment.acknowledge() 方法标记消息已被消费。
如果 cleanWikiSourceService.save 方法失败且没有调用 acknowledge()，那么消息将会被重新投递。
错误处理机制：
可以配置 SeekToCurrentErrorHandler（适用于 KafkaListenerContainerFactory）来处理消费异常。此类错误处理器允许指定重试次数和延迟时间，消息在达到最大重试次数后将跳过或交给死信队列。
原文地址：https://blog.csdn.net/jifgnie/article/details/143017742
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！