MQ系列：
kafka系列一: kafka简介
kafka系列二: kafka部署
kafka系列三: Spring kafka
kafka系列四：动态添加监听器
kafka系列五：失败后重试机制
前言
在 Spring Kafka 中，失败重试与死信队列的处理是关键功能，可以确保消息处理的可靠性和健壮性。当消费者处理消息失败时，可以配置重试机制，在重试多次后仍然失败时，将消息发送到死信队列进行处理。
重试机制的用法
springboot 中使用kafka消息失败重试机制非常便捷，关注 @RetryableTopic 和 @DltHandler 两个注解即可。
以下模拟处理失败的例子
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.retrytopic.DltHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.stereotype.Service;
import org.springframework.retry.annotation.Backoff;
@Service
public class KafkaConsumerService {
@RetryableTopic(
attempts = "3",
backoff = @Backoff(delay = 2000, multiplier = 2, maxDelay = 60000),
dltStrategy = RetryableTopic.DltStrategy.FAIL_ON_ERROR,
autoCreateTopics = "true"
)
@KafkaListener(topics = "my-topic", groupId = "${spring.kafka.consumer.group-id}")
public void listen(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
try {
System.out.println("Received message: " + record.value());
// 模拟异常
if (shouldFail()) {
throw new RuntimeException("Simulated failure");
}
acknowledgment.acknowledge();
} catch (Exception e) {
throw e;
}
}
private boolean shouldFail() {
// 模拟处理失败的条件
return true;
}
@DltHandler
public void dltListen(ConsumerRecord<String, String> record) {
System.out.println("Received message in DLT: " + record.value());
// 可以在这里添加对死信消息的处理逻辑
}
}
@RetryableTopic 注解：
这个注解会将失败后重试的监听注册为被标注的方法，如例子中，my-topic的监听器和my-topic处理失败后重试的监听都是方法 listen()
- attempts：指定重试次数。
- backoff：配置重试的间隔和倍数。
- dltStrategy：配置死信队列策略（DltStrategy.FAIL_ON_ERROR 表示处理失败时将消息发送到死信队列）。
- autoCreateTopics：配置是否自动创建重试和死信队列的主题。
@DltHandler 注解：
用于标记处理死信队列消息的方法。
死信队列的topic name根据原Topic(my-topic)在超过指定失败次数后自动生成新的Topic(my-topic.dlt)，并且被 @DltHandler 标注的方法监听。
自定义死信队列监听器
通过注解@RetryableTopic 和 @DltHandler，可以非常便捷地整合失败重试机制到你的app中，但是，不够灵活。
正如上篇文章提到的，动态增加新的Topic监听器时，如何引入对应的失败重试机制呢。
在上次分析的基础上，解析@RetryableTopic注解时，通过ConcurrentMessageListenerContainer的setCommonErrorHandler的方法设置异常重试配置的
/**
* Set the {@link CommonErrorHandler} which can handle errors for both record
* and batch listeners.
* @param commonErrorHandler the handler.
* @since 2.8
*/
public void setCommonErrorHandler(@Nullable CommonErrorHandler commonErrorHandler) {
this.commonErrorHandler = commonErrorHandler;
}
因此，我们需要定义一个返回实现CommonErrorHandler（DefaultErrorHandler ） 的方法
/***
*
* @return
*/
public DefaultErrorHandler errorHandler() {
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
backOff.setInitialInterval(2000);
backOff.setMultiplier(2);
backOff.setMaxInterval(10000);
return new DefaultErrorHandler(recoverer, backOff);
}
然后在动态添加新Topic监听器的方法中处理setCommonErrorHandler(errorHandler())
/**
* 添加 Topeic
*
* @param topic
* @param groupId
* @param isDeadLetter
*/
public void addKafkaListener(String topic, String groupId, boolean isDeadLetter) {
// kafka 消费者
DefaultKafkaConsumerFactory<String, String> consumerFactory = consumerFactory(groupId);
// 相关属性
ContainerProperties props = new ContainerProperties(topic);
// 设置监听器(区分死信队列与非死信队列)
if (!isDeadLetter) {
props.setMessageListener(new CustomerMsgErrorHandler());
} else {
props.setMessageListener(new DeadLetterHandler());
}
props.setGroupId(groupId);
props.setAckMode(ContainerProperties.AckMode.MANUAL);
ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(consumerFactory, props);
container.setCommonErrorHandler(errorHandler());
// 非死信队列，则添加一个死信队列监听器
if (!isDeadLetter) {
addKafkaListener(topic + ".DLT", groupId, true);
}
container.start();
}
如果直接用 @DltHandler标注方法的方式，添加死信队列监听器，监听器无效，故，直接增加了一个监听"topic.dlt"的监听器，来处理死信队列。
另外定义一个专门模拟异常的Hander和死信队列的Handkler
@Slf4j
public class CustomerMsgErrorHandler implements AcknowledgingMessageListener<String, String> {
@Override
public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
// doSomething
System.out.println("------------ onMessage topic " + data.topic() + ": " + data.value());
// 模拟异常
if (shouldFail()) {
throw new RuntimeException("Simulated failure");
}
// 因为前面设置了手动提交ack的方式，这里需要在消息处理完成后提交ack
acknowledgment.acknowledge();
}
private boolean shouldFail() {
// 模拟处理失败的条件
return true;
}
}
@Slf4j
public class DeadLetterHandler implements AcknowledgingMessageListener<String, String> {
@Override
public void onMessage(ConsumerRecord<String, String> data, Acknowledgment acknowledgment) {
// doSomething
System.out.println("------------ DLT onMessage topic " + data.topic() + ": " + data.value());
// 因为前面设置了手动提交ack的方式，这里需要在消息处理完成后提交ack
acknowledgment.acknowledge();
}
}
OK，这样就暂时实现了动态增加新Topic监听器的功能了，并且也用到了重试机制。
总结
上述配置展示了如何在 Spring Kafka 中实现失败重试与死信队列处理。通过配置 DefaultErrorHandler，可以设置重试机制和死信队列处理策略。在消费者监听器中处理消息时，如果出现异常，消息会根据配置的重试策略进行重试，多次重试失败后会被发送到死信队列。死信队列的监听器可以处理这些失败的消息，从而实现对异常消息的特殊处理。
在实现动态增加新Topic监听器的功能时，虽然，已经按照配置去执行失败重试了，但是，并没有如意料中的那样，回调@DltHandler标注的死信队列监听器。估计是没有把相关对象托管到springboot容器中的原因，下次再仔细瞧瞧，解决这个问题。
网友评论