使用 spring-kafka 2.1.x 在 Spring Boot 2.0 应用程序中实现死信队列 (DLQ)概念的最佳方法是什么，以便将某些 bean 的@KafkaListener方法未能处理的所有消息发送到某个预定义的 Kafka DLQ 主题并且不会丢失单个消息？
所以消耗的 Kafka 记录是：
- 成功处理，
- 处理失败，发送到DLQ主题，
- 处理失败，未发送到DLQ主题（由于意外问题），因此将再次被侦听器消费。
我尝试使用 KafkaTemplate 将ErrorHandler发送记录的自定义实现创建为无法处理到 DLQ 主题的侦听器容器。使用禁用的自动提交和RECORD AckMode。
spring.kafka.enable-auto-ack=false
spring.kafka.listener.ack-mode=RECORD
@Configuration
public class KafkaConfig {
@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory = ...
...
factory.getContainerProperties().setErrorHandler(dlqErrorHandler);
return factory;
}
}
@Component
public class DlqErrorHandler implements ErrorHandler {
@Autowired
private KafkaTemplate<Object, Object> kafkaTemplate;
@Value("${dlqTopic}")
private String dlqTopic;
@Override
public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
log.error("Error, sending to DLQ...");
kafkaTemplate.send(dlqTopic, record.key(), record.value());
}
}
似乎这个实现并不能保证 item #3。如果在 DlqErrorHandler 中抛出异常，则监听器不会再次消费该记录。
使用事务性侦听器容器会有帮助吗？
factory.getContainerProperties().setTransactionManager(kafkaTransactionManager);
有没有使用 Spring Kafka 实现 DLQ 概念的便捷方法？
更新 2018/03/28
感谢 Gary Russell 的回答，我能够通过如下实现 DlqErrorHandler 来实现所需的行为
@Configuration
public class KafkaConfig {
@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory = ...
...
factory.getContainerProperties().setAckOnError(false);
factory.getContainerProperties().setErrorHandler(dlqErrorHandler);
return factory;
}
}
@Component
public class DlqErrorHandler implements ContainerAwareErrorHandler {
...
@Override
public void handle(Exception thrownException, list<ConsumerRecord<?, ?> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
Consumerrecord<?, ? record = records.get(0);
try {
kafkaTemplate.send("dlqTopic", record.key, record.value());
consumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
// Other records may be from other partitions, so seek to current offset for other partitions too
// ...
} catch (Exception e) {
consumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset());
// Other records may be from other partitions, so seek to current offset for other partitions too
// ...
throw new KafkaException("Seek to current after exception", thrownException);
}
}
}
这样，如果消费者轮询返回 3 条记录（1、2、3）并且无法处理第二条记录：
- 1 将被处理
- 2 将无法处理并发送到DLQ
- 3 感谢消费者 seek to record.offset() + 1，它将被传递给监听器
如果发送到 DLQ 失败，消费者会寻找 record.offset() 并且记录将被重新传递给侦听器（并且发送到 DLQ 可能会被停用）。
更新 2021/04/30
由于 Spring Kafka 2.7.0原生支持非阻塞重试和死信主题。
查看示例：https ://github.com/evgeniy-khist/spring-kafka-non-blocking-retries-and-dlt
重试通常应该是非阻塞的（在单独的主题中完成）并延迟：
- 不扰乱实时流量；
- 不增加呼叫次数，本质上是垃圾请求；
- 可观察性（获取重试次数和其他元数据）。使用 Kafka 实现非阻塞重试和 DLT 功能通常需要设置额外的主题并创建和配置相应的侦听器。
