RabbitMQ 整合spring AMQP相关组件

作者: 三不猴子 | 来源:发表于2020-07-14 23:42 被阅读0次

RabbitMQ 整合spring AMQP相关组件

RabbitAdmin 是对rabbitMQ的操作工作工具.比如声明交换机，声明队列，删除之类的操作。

核心配置写法为

 @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        //connectionFactory.setAddresses("127.0.0.1:5672")
        connectionFactory.setAddresses("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("admin");
        connectionFactory.setUsername("admin");
        connectionFactory.setVirtualHost("/");
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

然后就可以直接注入一个RabbitAdmin对象了。RabbitAdmin源码如下：

@ManagedResource(description = "Admin Tasks")
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, ApplicationEventPublisherAware,
      BeanNameAware, InitializingBean {
.......
}

这个类实现了一些接口，我先看一下InitializingBean，这个接口表明这个类会初始化后会调用afterPropertiesSet方法。

@Override
  public void afterPropertiesSet() {

      synchronized (this.lifecycleMonitor) {

          if (this.running || !this.autoStartup) {
              return;
          }

          if (this.retryTemplate == null && !this.retryDisabled) {
              this.retryTemplate = new RetryTemplate();
              this.retryTemplate.setRetryPolicy(new SimpleRetryPolicy(DECLARE_MAX_ATTEMPTS));
              ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
              backOffPolicy.setInitialInterval(DECLARE_INITIAL_RETRY_INTERVAL);
              backOffPolicy.setMultiplier(DECLARE_RETRY_MULTIPLIER);
              backOffPolicy.setMaxInterval(DECLARE_MAX_RETRY_INTERVAL);
              this.retryTemplate.setBackOffPolicy(backOffPolicy);
          }
          if (this.connectionFactory instanceof CachingConnectionFactory &&
                  ((CachingConnectionFactory) this.connectionFactory).getCacheMode() == CacheMode.CONNECTION) {
              this.logger.warn("RabbitAdmin auto declaration is not supported with CacheMode.CONNECTION");
              return;
          }

          // Prevent stack overflow...
          final AtomicBoolean initializing = new AtomicBoolean(false);

          this.connectionFactory.addConnectionListener(connection -> {

              if (!initializing.compareAndSet(false, true)) {
                  // If we are already initializing, we don't need to do it again...
                  return;
              }
              try {
                  /*
                   * ...but it is possible for this to happen twice in the same ConnectionFactory (if more than
                   * one concurrent Connection is allowed). It's idempotent, so no big deal (a bit of network
                   * chatter). In fact it might even be a good thing: exclusive queues only make sense if they are
                   * declared for every connection. If anyone has a problem with it: use auto-startup="false".
                   */
                  if (this.retryTemplate != null) {
                      this.retryTemplate.execute(c -> {
                          initialize();
                          return null;
                      });
                  }
                  else {
                      initialize();
                  }
              }
              finally {
                  initializing.compareAndSet(true, false);
              }

          });

          this.running = true;

      }
  }

核心内容就是

if (this.retryTemplate != null) {
                      this.retryTemplate.execute(c -> {
                          initialize();
                          return null;
                      });
                  }
                  else {
                      initialize();
                  }

我们再看看initialize()方法中的实现

@Override // NOSONAR complexity
  public void initialize() {

      if (this.applicationContext == null) {
          this.logger.debug("no ApplicationContext has been set, cannot auto-declare Exchanges, Queues, and Bindings");
          return;
      }

      this.logger.debug("Initializing declarations");
      Collection<Exchange> contextExchanges = new LinkedList<Exchange>(
              this.applicationContext.getBeansOfType(Exchange.class).values());
      Collection<Queue> contextQueues = new LinkedList<Queue>(
              this.applicationContext.getBeansOfType(Queue.class).values());
      Collection<Binding> contextBindings = new LinkedList<Binding>(
              this.applicationContext.getBeansOfType(Binding.class).values());
      Collection<DeclarableCustomizer> customizers =
              this.applicationContext.getBeansOfType(DeclarableCustomizer.class).values();

      processDeclarables(contextExchanges, contextQueues, contextBindings);

      final Collection<Exchange> exchanges = filterDeclarables(contextExchanges, customizers);
      final Collection<Queue> queues = filterDeclarables(contextQueues, customizers);
      final Collection<Binding> bindings = filterDeclarables(contextBindings, customizers);

      for (Exchange exchange : exchanges) {
          if ((!exchange.isDurable() || exchange.isAutoDelete())  && this.logger.isInfoEnabled()) {
              this.logger.info("Auto-declaring a non-durable or auto-delete Exchange ("
                      + exchange.getName()
                      + ") durable:" + exchange.isDurable() + ", auto-delete:" + exchange.isAutoDelete() + ". "
                      + "It will be deleted by the broker if it shuts down, and can be redeclared by closing and "
                      + "reopening the connection.");
          }
      }

      for (Queue queue : queues) {
          if ((!queue.isDurable() || queue.isAutoDelete() || queue.isExclusive()) && this.logger.isInfoEnabled()) {
              this.logger.info("Auto-declaring a non-durable, auto-delete, or exclusive Queue ("
                      + queue.getName()
                      + ") durable:" + queue.isDurable() + ", auto-delete:" + queue.isAutoDelete() + ", exclusive:"
                      + queue.isExclusive() + ". "
                      + "It will be redeclared if the broker stops and is restarted while the connection factory is "
                      + "alive, but all messages will be lost.");
          }
      }

      if (exchanges.size() == 0 && queues.size() == 0 && bindings.size() == 0) {
          this.logger.debug("Nothing to declare");
          return;
      }
      this.rabbitTemplate.execute(channel -> {
          declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
          declareQueues(channel, queues.toArray(new Queue[queues.size()]));
          declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
          return null;
      });
      this.logger.debug("Declarations finished");

  }

  private void processDeclarables(Collection<Exchange> contextExchanges, Collection<Queue> contextQueues,
          Collection<Binding> contextBindings) {

      Collection<Declarables> declarables = this.applicationContext.getBeansOfType(Declarables.class, false, true)
              .values();
      declarables.forEach(d -> {
          d.getDeclarables().forEach(declarable -> {
              if (declarable instanceof Exchange) {
                  contextExchanges.add((Exchange) declarable);
              }
              else if (declarable instanceof Queue) {
                  contextQueues.add((Queue) declarable);
              }
              else if (declarable instanceof Binding) {
                  contextBindings.add((Binding) declarable);
              }
          });
      });
  }

声明了一些链表存放Exchange、Queue、Binding等对象。然后判断一下各种条件，讲这些对象加入进去，转换RabbitMQ client对象，进行交互。使用InitializingBean是在bean初始化后进行的。

在RabbitMQ中API中声明一个Exchange、绑定和队列。

在原生RabbitMQ中使用的channel进行操作的。在使用AMQP之后可以注入一个Exchange的bean的方式来声明。

     @Bean  
    public TopicExchange exchange001() {  
        return new TopicExchange("topic001", true, false);  
    }  

    @Bean  
    public Queue queue001() {  
        return new Queue("queue001", true); //队列持久  
    }

RabbitTemplate：整合spring AMQP的发送消息的模板类。消息可靠性投递模板，ConfirmCalllBack和ReturnCallBack等接口。

  @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
     RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
     return rabbitTemplate;
    }

SimpleMessageListenerContainer简单消息监听容器，监听多个队列、自动启动自动声明。设置事务、事务管理器、事务属性、事务容量（并发）、是否开启事务、回滚消息等。设置消息的签收模式、捕获消息异常的handle。设置消费者标签生成策略、是否独占模式、消费者属性。设置具体的监听器转换器等操作。可以动态的设置。

     @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001());
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(5);
        container.setDefaultRequeueRejected(false);
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        container.setExposeListenerChannel(true);
        container.setConsumerTagStrategy(queue -> queue + "_" + UUID.randomUUID().toString());
        return container;
    }

MessageListenerAdapter 消息监听适配器,主要是做消息监听器的适配工作。比如指定转换器之类的操作。

Message listener adapter that delegates the handling of messages to target listener methods via reflection, with flexible message type conversion. Allows listener methods to operate on message content types, completely independent from the Rabbit API.
消息监听适配器（adapter），通过反射将消息处理委托给目标监听器的处理方法，并进行灵活的消息类型转换。允许监听器方法对消息内容类型进行操作，完全独立于Rabbit API。
- defaultListenerMethod默认监听方法，设置监听方法名。
- deletage对象：真实委托对象，用于处理消息的对象。
- queueOrTagToMethodName队列标识与方法名组成的集合。可以将队列与方法名进行匹配。队列名称与方法名称绑定，即指定的队列的消息会被绑定的方法接受处理。
```
         @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {
     
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
     MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
      adapter.setDefaultListenerMethod("consumeMessage");

      //全局的转换器:
      ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();

      TextMessageConverter textConvert = new TextMessageConverter();
      convert.addDelegate("text", textConvert);
      convert.addDelegate("html/text", textConvert);
      convert.addDelegate("xml/text", textConvert);
      convert.addDelegate("text/plain", textConvert);
      adapter.setMessageConverter(convert);
      container.setMessageListener(adapter);
     
     return container;
     }
```

MessageConverter消息转换器接口，自定义消息转换器实现这个接口重写toMessage和fromMessage方法。

public class TextMessageConverter implements MessageConverter {
    @Override
    public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
      return new Message(object.toString().getBytes(), messageProperties);
}
   
       @Override
       public Object fromMessage(Message message) throws MessageConversionException {
           String contentType = message.getMessageProperties().getContentType();
           if (null != contentType && contentType.contains("text")) {
               return new String(message.getBody());
           }
           return message.getBody();
       }
   }
   ```

RabbitMQ 整合spring AMQP相关组件

RabbitMQ 整合spring AMQP相关组件

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读