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

2017-06-28  本文已影响2796人  linking12

背景

Spring Boot对于Rabbit有了AutoConfig的功能，但是延迟队列及失败重试却没有很好的实现

实现原理

具体原理可以参考 <a href="http://blog.csdn.net/u014308482/article/details/53036770">延迟队列原理</a>

如何基于spring boot来实现

基于上面两者需求，在spring boot下如何实现呢？

spring boot rabbit的自动化配置的问题
@Configuration
@ConditionalOnClass({ RabbitTemplate.class, Channel.class })
@EnableConfigurationProperties(RabbitProperties.class)
@Import(RabbitAnnotationDrivenConfiguration.class)
public class RabbitAutoConfiguration {

RabbitAnnotationDrivenConfiguration这个Configuration

总结：基于这种情况Spring boot的rabbit 的自动化配置我们只能自己重新定义，而需要将原生的spring boot rabbit的自动化配置给屏蔽掉

如何屏蔽spring boot的自动化配置

大家可以看看这篇文章 <a href="http://www.jianshu.com/p/aa27507df448">Spring Boot自动化配置的利弊及解决之道</a>
但是这种做法对于两个框架层面上存在问题

总结就是：期望就是引入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功能给屏蔽掉，不会增加配置工作量

延迟队列和重试队列的具体实现

以上聊了这么多就是为了实现延迟队列及重试队列的功能做的铺垫

在发送端的方法体上加上 @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);
    }

  }

@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

上一篇下一篇

猜你喜欢

热点阅读