RabbitMQ-Java-08-备份交换机

说明

• RabbitMQ-Java-08-备份交换机
• 本案例是一个Maven+SpringBoot项目
• 假设你已经实现了上一节发布确认高级
• 官方文档已包含绝大多数本案例内容。请移步：https://docs.spring.io/spring-amqp/docs/current/reference/html/

核心概念

》备份交换机说明

• 备份交换机是为了解决什么问题呢？
  • 跟发布确认高级差不多，也是为了解决防止消息丢失问题
  • 无法投递的消息将转发给备份交换机
• 主要操作有哪些呢？
  • 新建配置类
    • 新建一个普通交换机（direct），通过附加参数声明备份交换机

      return ExchangeBuilder.directExchange(EXCHANGE_NORMAL).withArgument(
          "alternate-exchange", EXCHANGE_BACKUP // 指定备份交换机
      ).build();

    • 新建一个普通队列
    • 绑定普通队列、交换机、路由key
    • 新建一个备份交换机（fanout）
    • 新建一个备份队列
    • 绑定备份队列、交换机
  • 新建消费者组件
    • 正常队列消费者
    • 备份队列消费者
  • 控制器
    • 生产者

操作步骤

》完整代码

• application.properties

  spring.rabbitmq.host=192.168.3.202
  spring.rabbitmq.port=5672
  spring.rabbitmq.username=admin
  spring.rabbitmq.password=123456
  spring.rabbitmq.publisher-confirm-type=correlated
  spring.rabbitmq.publisher-returns=true

• MyBackupConfig

  package cn.cnyasin.rabbit.config;
  
  import org.springframework.amqp.core.*;
  import org.springframework.beans.factory.annotation.Qualifier;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  
  @Configuration
  public class MyBackupConfig {
      // 交换机
      public static final String EXCHANGE_NORMAL = "exchange_normal";
      public static final String EXCHANGE_BACKUP = "exchange_backup";
  
      // 队列
      public static final String QUEUE_NORMAL = "queue_normal";
      public static final String QUEUE_BACKUP = "queue_backup";
  
      // 路由key
      public static final String ROUTING_NORMAL = "routing_normal";
  
      // 声明交换机
      @Bean
      public DirectExchange exchangeNormal() {
          return ExchangeBuilder.directExchange(EXCHANGE_NORMAL).withArgument(
                  "alternate-exchange", EXCHANGE_BACKUP // 指定备份交换机
          ).build();
      }
  
      @Bean
      public FanoutExchange exchangeBackup() {
          return new FanoutExchange(EXCHANGE_BACKUP);
      }
  
      // 声明队列
      @Bean
      public Queue queueNormal() {
          return QueueBuilder.durable(QUEUE_NORMAL).build();
      }
  
      @Bean
      public Queue queueBackup() {
          return QueueBuilder.durable(QUEUE_BACKUP).build();
      }
  
      // 绑定队列、交换机、路由key
      @Bean
      public Binding queueNormalBindExchangeNormal(
              @Qualifier("queueNormal") Queue queue,
              @Qualifier("exchangeNormal") Exchange exchange
      ) {
          return BindingBuilder.bind(queue).to(exchange).with(ROUTING_NORMAL).noargs();
      }
  
      @Bean
      public Binding queueBackupBindExchangeBackup(
              @Qualifier("queueBackup") Queue queue,
              @Qualifier("exchangeBackup") Exchange exchange
      ) {
          return BindingBuilder.bind(queue).to(exchange).with("").noargs();
      }
  
  }
  

• MyBackupConsumer

  package cn.cnyasin.rabbit.consumer;
  
  import cn.cnyasin.rabbit.config.MyBackupConfig;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.amqp.rabbit.annotation.RabbitListener;
  import org.springframework.stereotype.Component;
  
  import java.util.Date;
  
  @Slf4j
  @Component
  public class MyBackupConsumer {
      /**
       * 正常队列消费者
       *
       * @param message
       */
      @RabbitListener(queues = MyBackupConfig.QUEUE_NORMAL)
      public void queueNormalConsumer(String message) {
          log.info("[*] [{}] 正常队列收到消息：{}", new Date().toString(), message);
      }
  
      /**
       * 备份队列消费者
       *
       * @param message
       */
      @RabbitListener(queues = MyBackupConfig.QUEUE_BACKUP)
      public void queueBackupConsumer(String message) {
          log.info("[*] [{}] 备份队列收到消息：{}", new Date().toString(), message);
      }
  
  }
  

• MyBackupController

  package cn.cnyasin.rabbit.controller;
  
  import cn.cnyasin.rabbit.config.MyBackupConfig;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.amqp.rabbit.core.RabbitTemplate;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.web.bind.annotation.PathVariable;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RestController;
  
  import java.util.Date;
  
  @Slf4j
  @RestController
  @RequestMapping("/backup")
  public class MyBackupController {
  
      @Autowired
      private RabbitTemplate rabbitTemplate;
  
      /**
       * 生产者
       *
       * @param msg
       * @return
       */
      @RequestMapping("/send/{msg}")
      public String send(@PathVariable String msg) {
          log.info("[*] [{}] 准备发送消息：{}", new Date().toString(), msg);
  
          // 发送到存在的路由key
          rabbitTemplate.convertAndSend(MyBackupConfig.EXCHANGE_NORMAL, MyBackupConfig.ROUTING_NORMAL, msg);
          // 发送到不存在的路由key
          rabbitTemplate.convertAndSend(MyBackupConfig.EXCHANGE_NORMAL, "qwe", msg);
  
          return "ok";
      }
  
  }
  

备注

• 该教程部分内容收集自网络，感谢原作者。

附录

• 无