RabbitMQ---延迟队列，整合springboot

转x33g5p2x  于2021-10-20 转载在 RabbitMQ  
字(23.4k)|赞(0)|评价(0)|浏览(697)

延迟队列

概念

延时队列,队列内部是有序的，最重要的特性就体现在它的延时属性上，延时队列中的元素是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

使用场景

1.订单在十分钟之内未支付则自动取消

2.新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。

3.用户注册成功后，如果三天内没有登陆则进行短信提醒。

4.用户发起退款，如果三天内没有得到处理则通知相关运营人员。

5.预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

这些场景都有一个特点，需要在某个事件发生之后或者之前的指定时间点完成某一项任务，

如：

发生订单生成事件，在十分钟之后检查该订单支付状态，然后将未支付的订单进行关闭；

看起来似乎使用定时任务，一直轮询数据，每秒查一次，取出需要被处理的数据，然后处理不就完事了吗？

如果数据量比较少，确实可以这样做，

比如：对于“如果账单一周内未支付则进行自动结算”这样的需求，如果对于时间不是严格限制，而是宽松意义上的一周，那么每天晚上跑个定时任务检查一下所有未支付的账单，确实也是一个可行的方案。

但对于数据量比较大，并且时效性较强的场景，如：“订单十分钟内未支付则关闭“，短期内未支付的订单数据可能会有很多，活动期间甚至会达到百万甚至千万级别，对这么庞大的数据量仍旧使用轮询的方式显然是不可取的，很可能在一秒内无法完成所有订单
的检查，同时会给数据库带来很大压力，无法满足业务要求而且性能低下。

RabbitMQ 中的 TTL

TTL 是什么呢？TTL 是 RabbitMQ 中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间。

单位是毫秒。换句话说，如果一条消息设置了 TTL 属性或者进入了设置 TTL 属性的队列，那么这条消息如果在 TTL 设置的时间内没有被消费，则会成为"死信"。如果同时配置了队列的 TTL 和消息的TTL，那么较小的那个值将会被使用，有两种方式设置 TTL。

队列设置 TTL

第一种是在创建队列的时候设置队列的“x-message-ttl”属性

消息设置 TTL

另一种方式便是针对每条消息设置 TTL

两者的区别

如果设置了队列的 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)，

而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；

另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，

如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

整合 springboot

pom文件

  1. <dependencies>
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter</artifactId>
  5.         </dependency>
  7.         <dependency>
  8.             <groupId>org.springframework.boot</groupId>
  9.             <artifactId>spring-boot-starter-test</artifactId>
  10.             <scope>test</scope>
  11.         </dependency>
  13.         <dependency>
  14.             <groupId>org.springframework.boot</groupId>
  15.             <artifactId>spring-boot-starter-amqp</artifactId>
  16.         </dependency>
  18.         <dependency>
  19.             <groupId>org.springframework.boot</groupId>
  20.             <artifactId>spring-boot-starter-web</artifactId>
  21.         </dependency>
  23.         <dependency>
  24.             <groupId>com.alibaba</groupId>
  25.             <artifactId>fastjson</artifactId>
  26.             <version>1.2.76</version>
  27.         </dependency>
  29.         <dependency>
  30.             <groupId>org.projectlombok</groupId>
  31.             <artifactId>lombok</artifactId>
  32.         </dependency>
  34.         <dependency>
  35.             <groupId>io.springfox</groupId>
  36.             <artifactId>springfox-swagger2</artifactId>
  37.             <version>3.0.0</version>
  38.         </dependency>
  40.         <dependency>
  41.             <groupId>io.springfox</groupId>
  42.             <artifactId>springfox-swagger-ui</artifactId>
  43.             <version>3.0.0</version>
  44.         </dependency>
  46.         <dependency>
  47.             <groupId>org.springframework.amqp</groupId>
  48.             <artifactId>spring-rabbit-test</artifactId>
  49.             <scope>test</scope>
  50.         </dependency>
  51.     </dependencies>

配置文件

  1. spring.rabbitmq.host=192.168.112.128
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=admin
  4. spring.rabbitmq.password=123

添加Swagger配置类

  1. @Configuration
  2.  @EnableSwagger2
  3.  public class SwaggerConfig {
  4.      @Bean
  5.      public Docket webApiConfig(){
  6.          return new Docket(DocumentationType.SWAGGER_2)
  7.                  .groupName("webApi")
  8.                  .apiInfo(webApiInfo())
  9.                  .select()
  10.                  .build();
  11.      }
  13.  private ApiInfo webApiInfo(){
  14.      return new ApiInfoBuilder()
  15.              .title("rabbitmq接口文档")
  16.              .description("本文档描述了rabbitmq微服务接口定义")
  17.              .version("1.0")
  18.              .contact(new Contact("dhy","www.baidu.com","123456@qq.com"))
  19.              .build();
  20.  }
  21. }

队列 TTL

代码架构

创建两个队列QA和QB，两者队列TTL分别设置为10S和40S，然后在创建一个交换机X和死信交换机Y，它们的类型都是direct，创建一个死信队列QD，它们的绑定关系如下：

代码实现

配置类代码

  1. /*
  2. * TTL队列 配置文件类代码
  3. *
  4. * */
  5. @Configuration
  6. public class TtlQueueConfig {
  8.     //普通交换机的名称
  9.     public static final String  X_EXCHANGE = "X";
  10.     //死信交换机的名称
  11.     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  12.     //普通队列的名称
  13.     public static final String QUEUE_A = "QA";
  14.     public static final String QUEUE_B = "QB";
  15.     //死信队列的名称
  16.     public static final String DEAD_LATTER_QUEUE = "QD";
  18.     //声明xExchange---普通交换机
  19.     @Bean("xExchange")
  20.     public DirectExchange xExchange()
  21.     {
  22.         return new DirectExchange(X_EXCHANGE);
  23.     }
  25.     //声明yExchange---死信交换机
  26.     @Bean("yExchange")
  27.     public DirectExchange yExchange()
  28.     {
  29.         return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
  30.     }
  32.     //声明队列---创建并设置普通队列相关属性
  33.     @Bean("queueA")
  34.     public Queue queueA(){
  35.         Map<String, Object> arguments = new HashMap<>(3);
  36.         //设置死信交换机
  37.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  38.         //设置死信Routing-key
  39.         arguments.put("x-dead-letter-routing-key","YD");
  40.         //设置TTL 单位是ms ----该队列里面消息过期时间为10s
  41.         arguments.put("x-message-ttl",10000);
  42.         //队列持久化
  43.         return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
  44.     }
  46.     //声明普通队列 TTL为40s
  47.     @Bean("queueB")
  48.     public Queue queueB(){
  49.         Map<String, Object> arguments = new HashMap<>(3);
  50.         //设置死信交换机
  51.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  52.         //设置死信Routing-key
  53.         arguments.put("x-dead-letter-routing-key","YD");
  54.         //设置TTL 单位是ms
  55.         arguments.put("x-message-ttl",40000);
  56.         return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
  57.     }
  59.     //死信队列
  60.     @Bean("queueD")
  61.     public Queue queueD(){
  62.         return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
  63.     }
  65.     //绑定
  66.     @Bean
  67.     public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
  68.                                   @Qualifier("xExchange") DirectExchange xExchange){
  69.         //队列A和X交换机绑定
  70.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  71.     }
  73.     //绑定
  74.     @Bean
  75.     public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
  76.                                   @Qualifier("xExchange") DirectExchange xExchange){
  77.         //队列B和x交换机绑定
  78.         return BindingBuilder.bind(queueB).to(xExchange).with("XB");
  79.     }
  81.     //绑定
  82.     @Bean
  83.     public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
  84.                                   @Qualifier("yExchange") DirectExchange yExchange){
  85.         //死信队列D和死信交换机绑定
  86.         return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  87.     }
  88. }

消费者代码：定义rabbitMq的消费者可以相当方便，只需要在消息处理类或者类方法加上@RabbitListener注解，指定队列名称即可。

  1. /*
  2.  * 队列TTL 消费者
  3.  * */
  4.  @Slf4j
  5.  @Component
  6.  public class DeadLetterQueueConsumer {
  7.      //接收消息
  8.      @RabbitListener(queues = "QD")
  9.      public void receiveD(Message message, Channel channel) throws Exception {
  10.          String msg = new String(message.getBody());
  11.          log.info("当前时间：{}，收到死信队列的消息：{}",new Date().toString(),msg);
  12.      }
  13.  }

生产者代码

  1. /*
  2.  * 发送延迟消息
  3.  * */
  4.  @Slf4j
  5.  @RestController
  6.  @RequestMapping("/ttl")
  7.  public class SendMsgController {
  8.  
  9.      @Autowired
  10.      private RabbitTemplate rabbitTemplate;
  11.  
  12.      //开始发消息
  13.      @GetMapping("/sendMsg/{message}")
  14.      public void sendMsg(@PathVariable String message){
  15.        log.info("当前时间：{}，发送一条信息给两个TTL队列：{}",new Date().toString(),message);
  16.  
  17.        //String exchange, String routingKey, Object object
  18.        rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
  19.        rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);
  20.      }
  21.  }

效果

第一条消息在 10S 后变成了死信消息，然后被消费者消费掉，第二条消息在 40S 之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了

不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只有 10S 和 40S两个时间选项，如果需要一个小时后处理，那么就需要增加 TTL 为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？

延时队列优化

代码架构图

在这里新增了一个队列 QC,绑定关系如下,该队列不设置 TTL 时间

因为我们知道，ttl可以在生产者这边设置，也可以在消费者这边设置，如果是在消费者这边设置，这针对的是当前队列里面的所有消息，如果是在生产者这边设置，针对的是发出去的每一条消息

因此这里我们新增一个队列，在消费者这边不设置ttl,而是在生产者发出消息的时候，动态调整ttl时间

实现

配置文件类，新增队列QC

  1. /*
  2. * TTL队列 配置文件类代码
  3. *
  4. * */
  5. @Configuration
  6. public class TtlQueueConfig {
  8.     //普通交换机的名称
  9.     public static final String  X_EXCHANGE = "X";
  10.     //死信交换机的名称
  11.     public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
  12.     //普通队列的名称
  13.     public static final String QUEUE_A = "QA";
  14.     public static final String QUEUE_B = "QB";
  15.     //优化队列QC
  16.     public static final String QUEUE_C = "QC";
  17.     //死信队列的名称
  18.     public static final String DEAD_LATTER_QUEUE = "QD";
  20.     //声明xExchange---普通交换机
  21.     @Bean("xExchange")
  22.     public DirectExchange xExchange()
  23.     {
  24.         return new DirectExchange(X_EXCHANGE);
  25.     }
  27.     //声明yExchange---死信交换机
  28.     @Bean("yExchange")
  29.     public DirectExchange yExchange()
  30.     {
  31.         return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
  32.     }
  34.     //声明队列---创建并设置普通队列相关属性
  35.     @Bean("queueA")
  36.     public Queue queueA(){
  37.         Map<String, Object> arguments = new HashMap<>(3);
  38.         //设置死信交换机
  39.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  40.         //设置死信Routing-key
  41.         arguments.put("x-dead-letter-routing-key","YD");
  42.         //设置TTL 单位是ms ----该队列里面消息过期时间为10s
  43.         arguments.put("x-message-ttl",10000);
  44.         //队列持久化
  45.         return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
  46.     }
  48.     //声明普通队列 TTL为40s
  49.     @Bean("queueB")
  50.     public Queue queueB(){
  51.         Map<String, Object> arguments = new HashMap<>(3);
  52.         //设置死信交换机
  53.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  54.         //设置死信Routing-key
  55.         arguments.put("x-dead-letter-routing-key","YD");
  56.         //设置TTL 单位是ms
  57.         arguments.put("x-message-ttl",40000);
  58.         return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
  59.     }
  61.     //声明QC队列
  62.     @Bean("queueC")
  63.     public Queue queueC(){
  64.         Map<String, Object> arguments = new HashMap<>();
  65.         //设置死信交换机
  66.         arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
  67.         //设置死信RoutingKey
  68.         arguments.put("x-dead-letter-routing-key","YD");
  69.         return QueueBuilder.durable().withArguments(arguments).build();
  70.     }
  72.     //死信队列
  73.     @Bean("queueD")
  74.     public Queue queueD(){
  75.         return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
  76.     }
  78.     //绑定
  79.     @Bean
  80.     public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
  81.                                   @Qualifier("xExchange") DirectExchange xExchange){
  82.         //队列A和X交换机绑定
  83.         return BindingBuilder.bind(queueA).to(xExchange).with("XA");
  84.     }
  86.     //绑定
  87.     @Bean
  88.     public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
  89.                                   @Qualifier("xExchange") DirectExchange xExchange){
  90.         //队列B和x交换机绑定
  91.         return BindingBuilder.bind(queueB).to(xExchange).with("XB");
  92.     }
  94.     //绑定
  95.     @Bean
  96.     public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){
  97.         return BindingBuilder.bind(queueC).to(xExchange).with("XC");
  98.     }
  99.     
  100.     //绑定
  101.     @Bean
  102.     public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
  103.                                   @Qualifier("yExchange") DirectExchange yExchange){
  104.         //死信队列D和死信交换机绑定
  105.         return BindingBuilder.bind(queueD).to(yExchange).with("YD");
  106.     }
  107. }

生产者

  1. /*
  2.  * 发送延迟消息
  3.  * */
  4.  @Slf4j
  5.  @RestController
  6.  @RequestMapping("/ttl")
  7.  public class SendMsgController {
  8.  
  9.      @Autowired
  10.      private RabbitTemplate rabbitTemplate;
  11.  
  12.      //开始发消息
  13.      @GetMapping("/sendMsg/{message}")
  14.      public void sendMsg(@PathVariable String message){
  15.        log.info("当前时间：{}，发送一条信息给两个TTL队列：{}",new Date().toString(),message);
  17.        //String exchange, String routingKey, Object object
  18.        rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
  19.        rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);
  20.      }
  22.     //开始发消息
  23.     @GetMapping("sendExpirationMsg/{message}/{ttlTime}")
  24.     public void sendMsg(@PathVariable String message,@PathVariable String ttlTime){
  25.         log.info("当前时间：{}，发送一条时长{}毫秒TTL信息给队列QC：{}",
  26.                 new Date().toString(),ttlTime,message);
  27.         //通过x交换机加上XC路由key，将消息放到指定队列中，并同时设置当前发送消息的ttl
  28.         rabbitTemplate.convertAndSend("X","XC",message,msg->{
  29.             //发送消息的时候 延迟时长
  30.             msg.getMessageProperties().setExpiration(ttlTime);
  31.             return msg;
  32.         });
  33.     }
  34.  }

消费者代码无需改变，死信队列代码无需改变

测试

http://localhost:8080/ttl/sendExpirationMsg/你好 1/20000

http://localhost:8080/ttl/sendExpirationMsg/你好 2/2000

看起来似乎没什么问题，但是在最开始的时候，就介绍过如果使用在消息属性上设置 TTL 的方式，消息可能并不会按时“死亡“，因为 RabbitMQ 只会检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行。

Rabbitmq 插件实现延迟队列

上文中提到的问题，确实是一个问题，如果不能实现在消息粒度上的 TTL，并使其在设置的 TTL 时间及时死亡，就无法设计成一个通用的延时队列。那如何解决呢，接下来我们就去解决该问题。

安装延时队列插件

在官网上下载 https://www.rabbitmq.com/community-plugins.html，下载rabbitmq_delayed_message_exchange 插件，然后解压放置到 RabbitMQ 的插件目录。

进入 RabbitMQ 的安装目录下的 plgins 目录，执行下面命令让该插件生效，然后重启 RabbitMQ

  1. 将插件拷贝到下面这个目录下面
  2. /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins

  1. 安装延迟插件
  2. rabbitmq-plugins enable rabbitmq_delayed_message_exchange

  1. 重启RabbitMQ
  2.  systemctl  restart rabbitmq-server

以前:

现在:

代码架构图

在这里新增了一个队列 delayed.queue,一个自定义交换机 delayed.exchange，绑定关系如下:

配置文件类代码

  1. @Configuration
  2.  public class DelayedQueueConfig {
  4.  //队列
  5.  public static final String DELAYED_QUEUE_NAME = "delayed.queue";
  6.  //交换机
  7.  public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
  8.  //routingKey
  9.  public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
  11.  //声明队列
  12.  @Bean
  13.  public Queue delayedQueue(){
  14.    return new Queue(DELAYED_QUEUE_NAME);
  15.  };
  18.     //自定义交换机 我们在这里定义的是一个延迟交换机
  19.  @Bean
  20.  public CustomExchange delayedExchange(){
  22.      Map<String, Object> arguments = new HashMap<>();
  23.      //自定义交换机的类型
  24.      arguments.put("x-delayed-type","direct");
  25.           /*
  26.           * 1.交换机的名称
  27.           * 2.交换机的类型
  28.           * 3.是否需要持久化
  29.           * 4.是否需要自动删除
  30.           * 5.其他的参数
  31.           * */
  33.      return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",
  34.              true,false,arguments);
  35.  }
  36.  //绑定
  37.  @Bean
  38.  public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
  39.                                                    @Qualifier("delayedExchange") CustomExchange delayedExchange){
  40.      //noargs()构建
  41.      return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
  42.  }
  43. }

消费者

  1. // 消费者代码 基于插件的延迟消息
  2.  @Slf4j
  3.  @Component
  4.  public class DelayQueueConsumer {
  6.  //监听消息---监听哪一个队列
  7.  @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
  8.  public void recieveDelayQueue(Message message){
  9.      String msg = new String(message.getBody());
  10.      log.info("当前时间：{}，收到延迟队列的消息：{}",new Date().toString(),msg);
  11.  }
  12. }

生产者

  1. /*
  2.  * 发送延迟消息
  3.  * */
  4.  @Slf4j
  5.  @RestController
  6.  @RequestMapping("/ttl")
  7.  public class SendMsgController {
  8.  
  9.      @Autowired
  10.      private RabbitTemplate rabbitTemplate;
  11.  
  12.      //开始发消息 基于插件的 消息 及 延迟的时间
  13.      @GetMapping("/sendDelayMsg/{message}/{delayTime}")
  14.      public void sendMsg(@PathVariable String message, @PathVariable Integer delayTime){
  15.          log.info("当前时间：{}，发送一条时长{}毫秒信息给延迟队列delayed.queue：{}",
  16.                  new Date().toString(),delayTime,message);
  17.          rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME
  18.                  ,DelayedQueueConfig.DELAYED_ROUTING_KEY,message,msg -> {
  19.              // 发送消息的时候 延迟时长 单位ms
  20.              msg.getMessageProperties().setDelay(delayTime);
  21.              return msg;
  22.                  });
  23.      }
  24.  }

发起请求：

http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000

http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000

第二个消息被先消费掉了，符合预期

总结

延时队列在需要延时处理的场景下非常有用，使用 RabbitMQ 来实现延时队列可以很好的利用RabbitMQ 的特性，如：消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。

当然，延时队列还有很多其它选择，比如利用 Java 的 DelayQueue，利用 Redis 的 zset，利用 Quartz或者利用 kafka 的时间轮，这些方式各有特点,看需要适用的场景

发布确认高级

介绍

在生产环境中由于一些不明原因，导致 rabbitmq重启，在RabbitMQ重启期间生产者消息投递失败,导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行RabbitMQ的消息可靠投递呢?特别是在这样比较极端的情况，RabbitMQ集群不可用的时候，无法投递的消息该如何处理呢：

代码架构

代码

配置类，声明交换机和队列，并进行绑定

  1. // 配置类：发布确认（高级）
  2. @Configuration
  3. public class ConfirmConfig {
  5.     //交换机
  6.     public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
  7.     //队列
  8.     public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
  9.     //RoutingKey
  10.     public static final String CONFIRM_routing_key = "key1";
  12.     //声明交换机
  13.     @Bean//默认bean的名字是类名首字母小写
  14.     public DirectExchange confirmExchange()
  15.     {
  16.         return new DirectExchange(CONFIRM_EXCHANGE_NAME);
  17.     }
  19.     //声明队列
  20.     @Bean//默认bean的名字是类名首字母小写
  21.     public Queue confirmQueue()
  22.     {
  23.         return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
  24.     }
  26.     //绑定
  27.     @Bean
  28.     public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
  29.                                         @Qualifier("confirmExchange")DirectExchange confirmExchange)
  30.     {
  31.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_routing_key);
  32.     }
  33. }

消息生产者

  1. // 开始发消息 测试确认
  2.  @RestController
  3.  @Slf4j
  4.  @RequestMapping("/confirm")
  5.  public class ProducerController {
  7.  @Autowired
  8.  private RabbitTemplate rabbitTemplate;
  10.  //发消息
  11.  @GetMapping("/sendMessage/{message}")
  12.  public void sendMessage(@PathVariable String message){
  13.      rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
  14.              ,ConfirmConfig.CONFIRM_routing_key
  15.              ,message);
  16.      log.info("发送消息内容：{}",message);
  17.  }
  18. }

消费者

  1. // 接收消息
  2.  @Slf4j
  3.  @Component
  4.  public class Consumer {
  6.      //监听的队列
  7.  @RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)
  8.  public void receiveConfirmMessage(Message message){
  9.      String msg = new String(message.getBody());
  10.      log.info("接受到的队列confirm.queue消息：{}",msg);
  11.  }
  12. }

回调接口

  1. //回调接口
  2. @Component
  3. @Slf4j
  4. //实现回调接口
  5. public class MyCallBack implements RabbitTemplate.ConfirmCallback {
  7.     @Autowired
  8.     private RabbitTemplate rabbitTemplate;
  10.     //Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
  11.     @PostConstruct
  12.     public void init(){
  13.         //将当前实现类，注入到rabbitTemplate中的接口上
  14.         //不然到时候rabbitTemplate调用接口时，找不到我们的实现类
  15.         rabbitTemplate.setConfirmCallback(this);
  16.     }
  18.     /*
  19.      * 交换机确认回调方法,发消息后，交换机接收到了就回调
  20.      *   1.1 correlationData：保存回调消息的ID及相关信息
  21.      *   1.2 b:交换机收到消息，为ack=true
  22.      *   1.3 s:失败原因，成功为null
  23.      *
  24.      * 发消息，交换机接受失败，也回调
  25.      *   2.1 correlationData：保存回调消息的ID及相关信息
  26.      *   2.2 b:交换机没收到消息，为ack=false
  27.      *   2.3 s:失败的原因
  28.      *
  29.      * */
  31.     @Override
  32.     public void confirm(CorrelationData correlationData, boolean b, String s) {
  33.         String id = correlationData!=null ? correlationData.getId():"";
  34.         if (b){
  35.             log.info("交换机已经收到ID为：{}的信息",id);
  36.         }else {
  37.             log.info("交换机还未收到ID为：{}的消息，由于原因：{}",id,s);
  38.         }
  39.     }
  40. }

增加配置文件相关属性设置和消息发送方

  1. @RestController
  2. @Slf4j
  3. @RequestMapping("/confirm")
  4. public class ProducerController {
  6.  @Autowired
  7.  private RabbitTemplate rabbitTemplate;
  9.  //发消息
  10.  @GetMapping("/sendMessage/{message}")
  11.  public void sendMessage(@PathVariable String message)
  12.  {
  13.   //消息唯一标识
  14.   CorrelationData correlationData = new CorrelationData("1");
  15.   //交换机的名称，交换机的路由，消息和消息的唯一标识
  16.   rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
  17.           ,ConfirmConfig.CONFIRM_routing_key
  18.           ,message,correlationData);
  19.   log.info("发送消息内容：{}",message);
  20.  }
  21. }

在配置文件当中需要添加

  1. spring.rabbitmq.publisher-confirm-type=correlated

⚫ NONE 禁用发布确认模式，是默认值
⚫ CORRELATED 发布消息成功到交换器后会触发回调方法
⚫ SIMPLE

经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法

其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker

最后一个SIMPLE就是单个确认模式，即发布一条消息，确认一条消息

  1. spring.rabbitmq.host=192.168.112.128
  2. spring.rabbitmq.port=5672
  3. spring.rabbitmq.username=admin
  4. spring.rabbitmq.password=123
  5. spring.rabbitmq.publisher-confirm-type=correlated

结果

可以看到，发送了两条消息，第一条消息的 RoutingKey 为 “key1”，第二条消息的 RoutingKey 为"key2"，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。

回退消息

当前问题: 当交换机收到消息后，会调用回调函数，同样当交换机没收到消息时，也会调用回调函数，但是如果交换机收到了消息，队列没有收到消息，我们无法得知

Mandatory 参数

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。

那么如何让无法被路由的消息帮我想办法处理一下？

最起码通知我一声，我好自己处理啊。通过设置 mandatory 参数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

配置文件中开启回退消息功能

  1. spring.rabbitmq.publisher-returns=true

设置回退接口

  1. //回调接口
  2. @Component
  3. @Slf4j
  4. //实现回调接口
  5. public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
  7.     @Autowired
  8.     private RabbitTemplate rabbitTemplate;
  10.     //Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
  11.     @PostConstruct
  12.     public void init(){
  13.         //将当前实现类，注入到rabbitTemplate中的接口上
  14.         //不然到时候rabbitTemplate调用接口时，找不到我们的实现类
  15.         rabbitTemplate.setConfirmCallback(this);
  16.         rabbitTemplate.setReturnsCallback(this);
  17.     }
  19.     /*
  20.      * 交换机确认回调方法,发消息后，交换机接收到了就回调
  21.      *   1.1 correlationData：保存回调消息的ID及相关信息
  22.      *   1.2 b:交换机收到消息，为ack=true
  23.      *   1.3 s:失败原因，成功为null
  24.      *
  25.      * 发消息，交换机接受失败，也回调
  26.      *   2.1 correlationData：保存回调消息的ID及相关信息
  27.      *   2.2 b:交换机没收到消息，为ack=false
  28.      *   2.3 s:失败的原因
  29.      *
  30.      * */
  32.     @Override
  33.     public void confirm(CorrelationData correlationData, boolean b, String s) {
  34.         String id = correlationData!=null ? correlationData.getId():"";
  35.         if (b){
  36.             log.info("交换机已经收到ID为：{}的信息",id);
  37.         }else {
  38.             log.info("交换机还未收到ID为：{}的消息，由于原因：{}",id,s);
  39.         }
  40.     }
  42.     //可以在当消息传递过程中不可达目的的时将消息返回给生产者
  43.     //只有不可达目的地的时候才可回退
  44.     @Override
  45.     public void returnedMessage(ReturnedMessage returnedMessage) {
  46.         log.error("消息{}，被交换机{}退回，退回的原因：{},路由Key：{}",
  47.                 new String(returnedMessage.getMessage().getBody())
  48.                 , returnedMessage.getExchange()
  49.                 , returnedMessage.getReplyText()
  50.                 , returnedMessage.getRoutingKey());
  51.     }
  53. }

发布测试

  1. @RestController
  2. @Slf4j
  3. @RequestMapping("/confirm")
  4. public class ProducerController {
  6.  @Autowired
  7.  private RabbitTemplate rabbitTemplate;
  9.     //发消息
  10.     @GetMapping("/sendMessage/{message}")
  11.     public void sendMessage(@PathVariable String message){
  13.         //交换机可达，队列可达
  14.         CorrelationData correlationData = new CorrelationData("1");
  15.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
  16.                 ,ConfirmConfig.CONFIRM_routing_key
  17.                 ,message+"key1",correlationData);
  18.         log.info("发送消息内容：{}",message+"key1");
  20.         //下面这个交换机可达，队列不存在，不可达
  21.         CorrelationData correlationData2 = new CorrelationData("2");
  22.         rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME
  23.                 ,ConfirmConfig.CONFIRM_routing_key+"2"
  24.                 ,message+"key12",correlationData2);
  25.         log.info("发送消息内容：{}",message+"key12");
  26.     }
  27. }

小总结

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（包含消息的唯一ID）,这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

  1. 对于ConfirmCallback来说：
  2. 如果消息没有到exchange,则confirm回调,ack=false
  3. 如果消息到达exchange,则confirm回调,ack=true
  5. 对于ReturnCallback来说：
  6. exchange到queue成功,则不回调return
  7. exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
  8. 比如路由不到队列时触发回调

可以使用rabbitTemplate.setMandatory(true);替代配置文件中的spring.rabbitmq.publisher-returns=true

  1. //Constructor(构造方法) -> @Autowired(依赖注入) -> @PostConstruct(注释的方法)
  2.     @PostConstruct
  3.     public void init(){
  4.         //将当前实现类，注入到rabbitTemplate中的接口上
  5.         //不然到时候rabbitTemplate调用接口时，找不到我们的实现类
  6.         rabbitTemplate.setConfirmCallback(this);
  7.         rabbitTemplate.setReturnsCallback(this);
  8.         /**
  9.          * true：
  10.          * 交换机无法将消息进行路由时，会将该消息返回给生产者
  11.          * false：
  12.          * 如果发现消息无法进行路由，则直接丢弃
  13.          */
  14.         rabbitTemplate.setMandatory(true);
  15.     }

RabbitMQ自学之路（七）—— RabbitMQ消息发送确认与消息接收确认机制

RabbitMQ消息确认机制之Confirm模式总结

备份交换机

有了mandatory 参数和回退消息，我们获得了对无法投递消息的感知能力，有机会在生产者的消息无法被投递时发现并处理。

但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。

而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。

而且设置mandatory参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。

如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢?

前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。

在RabbitMQ.中，有一种备份交换机的机制存在，可以很好的应对这个问题。

什么是备份交换机呢?

备份交换机可以理解为 RabbitMQ中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为Fanout，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。

当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。

代码实现

配置类
  1. // 配置类：发布确认（高级）
  2. @Configuration
  3. public class ConfirmConfig {
  5.     //交换机
  6.     public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
  7.     //队列
  8.     public static final String CONFIRM_QUEUE_NAME = "confirm_queue";
  9.     //RoutingKey
  10.     public static final String CONFIRM_routing_key = "key1";
  11.     //备份交换机
  12.     public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
  13.     //备份队列
  14.     public static final String BACKUP_QUEUE_NAME = "backup_queue";
  15.     //报警队列
  16.     public static final String WARNING_QUEUE_NAME = "warning_queue";
  18.     //声明交换机
  19.     @Bean//默认bean的名字是类名首字母小写
  20.     public DirectExchange confirmExchange()
  21.     {
  22.         return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
  23.             //指定备份交换机
  24.                 .withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
  25.     }
  27.     //声明队列
  28.     @Bean//默认bean的名字是类名首字母小写
  29.     public Queue confirmQueue()
  30.     {
  31.         return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
  32.     }
  34.     //备份交换机---一般是扇出类型
  35.     @Bean
  36.     public FanoutExchange backupExchange(){
  37.         return new FanoutExchange(BACKUP_EXCHANGE_NAME);
  38.     }
  40.     //备份队列
  41.     @Bean
  42.     public Queue backupQueue(){
  43.         return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
  44.     }
  46.     //报警队列
  47.     @Bean
  48.     public Queue warningQueue(){
  49.         return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
  50.     }
  52.     //绑定
  53.     @Bean
  54.     public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,
  55.                                         @Qualifier("confirmExchange")DirectExchange confirmExchange)
  56.     {
  57.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_routing_key);
  58.     }
  60.     @Bean
  61.     public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
  62.                                                     @Qualifier("backupExchange") FanoutExchange backupExchange){
  63.         return BindingBuilder.bind(backupQueue).to(backupExchange);
  64.     }
  66.     @Bean
  67.     public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue backupQueue,
  68.                                                      @Qualifier("backupExchange") FanoutExchange backupExchange){
  69.         return BindingBuilder.bind(backupQueue).to(backupExchange);
  70.     }
  71. }

消费者

  1. // 报警消费者
  2.  @Component
  3.  @Slf4j
  4.  public class WarningConsumer {
  6.  //接受报警消息
  7.  @RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)
  8.  public void receiveWarningMsg(Message message){
  9.      String msg = new String(message.getBody());
  10.      log.error("报警发现不可路由消息：{}",msg);
  11.  }
  12. }

重新启动项目的时候需要把原来的 confirm.exchange 删除因为我们修改了其绑定属性，不然报以下错:

结果:

注意:mandatory 参数与备份交换机可以一起使用的时候，如果两者同时开启，消息究竟何去何从？谁优先级高，经过上面结果显示答案是备份交换机优先级高

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/m0_53157173/article/details/120835622

内容来源于网络，如有侵权，请联系作者删除！

Javaspring bootspringcloudrabbitmq

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com