背景
Spring Boot对于Rabbit有了AutoConfig的功能，但是延迟队列及失败重试却没有很好的实现
- 延迟队列
延迟队列存储的对象肯定是对应的延时消息，所谓”延时消息”是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。 - 失败自动重试
对于有些消息，有可能种种原因业务方消费失败，而又想重新让该消息自动进行重试的功能，实现类似RocketMq重试队列的功能
实现原理
- 延迟队列可以基于RabbitMq的DeadLetterExchange来实现，而DeadLetterExchange顾名思义，就是死信邮箱，类似于RocketMq的死信队列的味道存在，将消息发送到死信邮箱，通过设置以下三个参数来讲消息重新路由到真正的队列上
x-message-ttl: 1000 //消息延迟时间 x-dead-letter-exchange: tradeExchange //失败后重新将消息路由到具体exchange上 x-dead-letter-routing-key: tradeRouteKey //失败后重新将消息路由到具体routeKey上
具体原理可以参考 <a href="http://blog.csdn.net/u014308482/article/details/53036770">延迟队列原理</a>
- 失败自动重试有两种做法
1）原生的spring rabbit是基于spring retry机制来做，重复调用invokeListener来实现失败重试的功能
2）还是基于DeadLetterExchange来做失败重试的功能
两者的优缺点有： - spring-retry来说优点是spring rabbit已经帮忙实现好了，配置即可使用，但是存在一个问题，使用该方式会导致消费线程堵塞，以及如果在失败重试的过程中宕机了，该重试将彻底不起作用
- 基于DeadLetterExchange的话，没有实现，需要自己写代码实现，不会产生消费线程堵塞的问题，消息不会肯定不会丢失
如何基于spring boot来实现
基于上面两者需求，在spring boot下如何实现呢？
spring boot rabbit的自动化配置的问题
@Configuration
@ConditionalOnClass({ RabbitTemplate.class, Channel.class })
@EnableConfigurationProperties(RabbitProperties.class)
@Import(RabbitAnnotationDrivenConfiguration.class)
public class RabbitAutoConfiguration {
RabbitAnnotationDrivenConfiguration这个Configuration
作用是消费监听的主要自动化配置，构建SimpleRabbitListenerContainer，而我们要实现失败重试的功能的话，必须要有一个将消息重新发送到死信邮箱的功能
RabbitAnnotationDrivenConfiguration是在RabbitAutoConfiguration之前加载，在构建的时候RabbitTemplate还没开始初始化，所以RabbitAnnotationDrivenConfiguration这种方式是无法注入RabbitTemplate的
总结：基于这种情况Spring boot的rabbit 的自动化配置我们只能自己重新定义，而需要将原生的spring boot rabbit的自动化配置给屏蔽掉
如何屏蔽spring boot的自动化配置
大家可以看看这篇文章 <a href="http://www.jianshu.com/p/aa27507df448">Spring Boot自动化配置的利弊及解决之道</a>
但是这种做法对于两个框架层面上存在问题
- 无法保证用户去配置@EnableAutoConfiguration(exclude={RabbitAutoConfiguration.class})
- 担心是否会覆盖用户配置的spring.autoconfigure.exclude的值
总结就是：期望就是引入jar就能自动给我解决这些问题，我不想多加任何配置
spring boot的黑科技
我们看看spring将autoconfig给exclude的源码，看看这个类AutoConfigurationImportSelector
private List<String> getExcludeAutoConfigurationsProperty() {
if (getEnvironment() instanceof ConfigurableEnvironment) {
RelaxedPropertyResolver resolver = new RelaxedPropertyResolver(
this.environment, "spring.autoconfigure.");
Map<String, Object> properties = resolver.getSubProperties("exclude");
if (properties.isEmpty()) {
return Collections.emptyList();
}
List<String> excludes = new ArrayList<String>();
for (Map.Entry<String, Object> entry : properties.entrySet()) {
String name = entry.getKey();
Object value = entry.getValue();
if (name.isEmpty() || name.startsWith("[") && value != null) { //黑科技出现
excludes.addAll(new HashSet<String>(Arrays.asList(StringUtils
.tokenizeToStringArray(String.valueOf(value), ","))));
}
}
return excludes;
}
RelaxedPropertyResolver resolver = new RelaxedPropertyResolver(getEnvironment(),
"spring.autoconfigure.");
String[] exclude = resolver.getProperty("exclude", String[].class);
return (Arrays.asList(exclude == null ? new String[0] : exclude));
}
这里吐槽一下，spring boot的这个代码写的真心不咋样，黑科技一下子就能体现出来了，我们发现spring boot对于Property是基于两层方式的，如果是基于PropertiesPropertySource的name含有[的话，他就会累加，而不是覆盖，所以最终我们可以在代码中这样实现屏蔽自动化配置
public class RabbitEnviromentPostProcessor implements EnvironmentPostProcessor, Ordered {
private static final String EXCLUDE_AUTOCONFIGURATION =
"spring.autoconfigure.exclude[rabbitSource]"; //这里一定要加[，否则将会用户在Application.yml的配置给覆盖掉了
private static final String RABBIT_AUTOCONFIGURATION =
"org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration";
@Override
public void postProcessEnvironment(ConfigurableEnvironment environment,
SpringApplication application) {
try {
MutablePropertySources mutablePropertySources = environment.getPropertySources();
Properties propertySource = new Properties();
propertySource.setProperty(EXCLUDE_AUTOCONFIGURATION, RABBIT_AUTOCONFIGURATION);
EnumerablePropertySource<?> enumerablePropertySource =
new PropertiesPropertySource("rabbitSource", propertySource);
mutablePropertySources.addFirst(enumerablePropertySource);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
@Override
public int getOrder() {
return 0;
}
}
最终的话，我们在spring.factories上配置上这个EnvironmentPostProcessor就可以了
org.springframework.boot.env.EnvironmentPostProcessor=com.dianrong.platform.amqp.RabbitEnviromentPostProcessor
基于这种方式，好处是在代码中实现了将spring boot的autoconfig功能给屏蔽掉，不会增加配置工作量
延迟队列和重试队列的具体实现
以上聊了这么多就是为了实现延迟队列及重试队列的功能做的铺垫
- 延迟队列实现
拦截所有的调用发送消息的方法，如果开启了延迟队列的功能，将他的Exchange自动改为DeadLetterExchange，如：
在发送端的方法体上加上 @Delay的注解
@Delay
public void send() {
this.rabbitTemplate.convertAndSend("testexchange", "testroute", "hello");
}
扩展RabbitTemplate
@Override
protected void doSend(Channel channel, String exchange, String routingKey, Message message,
boolean mandatory, CorrelationData correlationData) throws Exception {
try {
String exchangeCopy = exchange;
if (DELAY_QUEUE_CONTENT.get()) { //如果当前线程上下文开启了延迟队列，将自动exchange改为RabbitTemplate
exchangeCopy = "DeadLetterExchange";
}
super.doSend(channel, exchangeCopy, routingKey, message, mandatory, correlationData);
} finally {
setDelayQueue(Boolean.FALSE);
}
}
- 重试队列的实现
扩展MessageRecoverer的恢复，如果是消费失败了，重新发送到DeadLetterExchange上
@Override
public void recover(Message message, Throwable cause) {
MessageProperties messageProperties = message.getMessageProperties();
Map<String, Object> headers = message.getMessageProperties().getHeaders();
Integer republishTimes = (Integer) headers.get(X_REPUBLISH_TIMES);
if (republishTimes != null) { //如果超过了重试次数，直接返回
if (republishTimes >= recoverTimes) {
log.warn(String.format("this message [ %s] republish times >= %d times, and will discard",
message.toString(), RabbitConstant.DEFAULT_REPUBLISH_TIMES));
return;
} else {
republishTimes = republishTimes + 1; //重试次数+1
}
} else {
republishTimes = 1;
}
headers.put(RepublishDeadLetterRecoverer.X_REPUBLISH_TIMES, republishTimes);
messageProperties.setRedelivered(true);
headers.put(X_EXCEPTION_STACKTRACE, getStackTraceAsString(cause));
headers.put(X_EXCEPTION_MESSAGE,
cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
headers.put(X_ORIGINAL_EXCHANGE, message.getMessageProperties().getReceivedExchange());
headers.put(X_ORIGINAL_ROUTING_KEY, message.getMessageProperties().getReceivedRoutingKey());
String routingKey = genRouteKey(message);
this.errorTemplate.send("DeadLetterExchange", routingKey, message);
log.info("The #" + republishTimes + " republish message ["
+ message.getMessageProperties().getMessageId() + "] to exchange [" + this.errorExchangeName
+ "] and routingKey[" + routingKey + "]");
}
以上就是如何在spring boot的框架下如何比较优雅的实现延迟及重试队列的一些做法，具体代码的话，改天上传到Github上，也欢迎关注我的 <a href="https://github.com/linking12/">GitHub</a>
实现源码 https://github.com/linking12/spring-boot-starter-rabbit
