RabbitMq延迟、重试队列及Spring Boot的黑科技

背景

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

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容