SpringBoot2.3整合RabbitMQ实现延迟消费消息

@Configuration
  public class RabbitMQConfig {

          /**
           * 测试发送消息到MQ
           * @return
           */
          @Bean
          public Queue testHello() {
                  return new Queue(SysConstant.QUEUE_TEST_HELLO);
          }


          /**
           * 死信交换机
           * @return
           */
          @Bean
          public DirectExchange sysOrderDelayExchange() {
                  return new DirectExchange(SysConstant.SYS_ORDER_DELAY_EXCHANGE);
          }

          /**
           * 死信队列
           * @return
           */
          @Bean
          public Queue sysOrderDelayQueue() {
                  Map<String, Object> map = new HashMap<String, Object>(16);
                  map.put("x-dead-letter-exchange",SysConstant.SYS_ORDER_RECEIVE_EXCHANGE); //指定死信送往的交换机
                  map.put("x-dead-letter-routing-key", SysConstant.SYS_ORDER_RECEIVE_KEY); //指定死信的routingkey
                  return new Queue(SysConstant.SYS_ORDER_DELAY_QUEUE, true, false, false, map);
          }

          /**
           * 给死信队列绑定死信交换机
           * @return
           */
          @Bean
          public Binding sysOrderDelayBinding() {
                  return BindingBuilder.bind(sysOrderDelayQueue()).to(sysOrderDelayExchange()).with(SysConstant.SYS_ORDER_DELAY_KEY);
          }

          /**
           * 死信接收交换机,用于接收死信队列的消息
           * @return
           */
          @Bean
          public DirectExchange sysOrderReceiveExchange() {
                  return new DirectExchange(SysConstant.SYS_ORDER_RECEIVE_EXCHANGE);
          }

          /**
           * 死信接收队列
           * @return
           */
          @Bean
          public Queue sysOrderReceiveQueue() {
                  return new Queue(SysConstant.SYS_ORDER_RECEIVE_QUEUE);
          }

          /**
           * 死信接收交换机绑定接收死信队列消费队列
           * @return
           */
          @Bean
          public Binding sysOrdeReceiveBinding() {
                  return BindingBuilder.bind(sysOrderReceiveQueue()).to(sysOrderReceiveExchange()).with(SysConstant.SYS_ORDER_RECEIVE_KEY);
          }
  }

 发送延时消息到死信交换器方法

          @Service
          public class MsgService {

                  @Autowired
                  private RabbitTemplate rabbitTemplate;
                  /**
                   * 发送延时消息到mq
                   * @param exchange 死信交换机
                   * @param routeKey 路由key
                   * @param data 发送数据
                   * @param delayTime 过期时间，单位毫秒
                   */
                  public void sendDelayMsgToMQ(String exchange, String routeKey, String data,int delayTime) {
                          rabbitTemplate.convertAndSend(exchange, routeKey, data, message -> {
                                  message.getMessageProperties().setExpiration(delayTime + "");
                                  return message;
                          });
                  }
          }
 

   /**
           * 获取到的延时消息
           * 这里接收到消息进行对应的业务处理(例如:取消订单，关闭支付，回滚库存等 ...)
           * @param msg
           */
          @RabbitListener(queues = SysConstant.SYS_ORDER_RECEIVE_QUEUE)
          @RabbitHandle
          public void getdelayMsg(String msg) {
                  log.info("MQ接收消息时间:{},消息内容:{}", DateUtil.formatDateTime(DateUtil.date()),msg);
                  log.info("------->这里实现订单关闭、支付关闭、回滚库存业务逻辑...");
          }

          创建Controller向队列发送消息，设置过期时间10秒

                  @RestControlle
                  @RequestMapping("mq")
                  @Slf4j
                  public class MQController {

                          @Autowired
                          private MsgService msgService;

                          @GetMapping("sendMsg")
                          public String sendMsg() {
                                  log.info("发送延时消息时间:" + DateUtil.formatDateTime(DateUtil.date()));

                                  OrderInfo orderInfo = new OrderInfo();
                                  orderInfo.setOrderId(IdUtil.fastSimpleUUID());
                                  orderInfo.setOrderState("待支付");
                                  orderInfo.setPayMoney(999.88);
                                  msgService.sendDelayMsgToMQ(SysConstant.SYS_ORDER_DELAY_EXCHANGE,SysConstant.SYS_ORDER_DELAY_KEY, JSONUtil.toJsonStr(orderInfo),10*1000);//1分钟
                                  return JSONUtil.toJsonStr("发送延时消息成功");
                          }
                  }