使用RabbitMQ实现订单超时取消（延迟队列）

使用RabbitMQ实现订单超时取消，大致流程：

 

生产者生产一条设置了TTL的延迟取消订单消息=>延迟队列交换机（通过绑定路由键）=>消息投递至延迟队列=>消息延迟队列时间到期=>经过死信队列交换机（通过绑定路由键）=>投递至死信队列=>消费者监听死信队列消息即时消费（做取消订单逻辑）。

 

下面来看代码：

一、先声明交换机、队列以及他们的绑定关系：

@Configuration
public class RabbitMQConfig {

    // 声明延时队列交换机
    public static final String DELAY_EXCHANGE_NAME = "delay.queue.demo.business.exchange";


    //延时队列c
    public static final String DELAY_QUEUEC_NAME = "delay.queue.demo.business.queuec";


    //延时队列c路由key
    public static final String DELAY_QUEUEC_ROUTING_KEY = "delay.queue.demo.business.queuec.routingkey";

    //声明死信队列交换机
    public static final String DEAD_LETTER_EXCHANGE = "delay.queue.demo.deadletter.exchange";


    // 死信队列c
    public static final String DEAD_LETTER_QUEUEC_NAME = "delay.queue.demo.deadletter.queuec";


    //死信交换机 的 不设时间路由key
    public static final String DEAD_LETTER_QUEUEC_ROUTING_KEY = "delay.queue.demo.deadletter.delay_anytime.routingkey";

    // 声明延时Exchange
    @Bean("delayExchange")
    public DirectExchange delayExchange(){
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }

    // 声明死信Exchange
    @Bean("deadLetterExchange")
    public DirectExchange deadLetterExchange(){
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }



    // 声明延时队列C 不设置TTL
    // 并绑定到对应的死信交换机
    @Bean("delayQueueC")
    public Queue delayQueueC(){
        Map<String, Object> args = new HashMap<>(3);
        // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        // x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEC_ROUTING_KEY);
        return QueueBuilder.durable(DELAY_QUEUEC_NAME).withArguments(args).build();
    }


    // 声明死信队列C 用于接收延时任意时长处理的消息
    @Bean("deadLetterQueueC")
    public Queue deadLetterQueueC(){
        return new Queue(DEAD_LETTER_QUEUEC_NAME);
    }



    // 声明延时队列C绑定关系
    @Bean
    public Binding delayBindingC(@Qualifier("delayQueueC") Queue queue,
                                 @Qualifier("delayExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUEC_ROUTING_KEY);
    }


    // 声明死信队列C绑定关系
    @Bean
    public Binding deadLetterBindingC(@Qualifier("deadLetterQueueC") Queue queue,
                                      @Qualifier("deadLetterExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEC_ROUTING_KEY);
    }
}

二、设置延迟队列配置绑定关系

@Configuration
public class DelayedRabbitMQConfig {
    public static final String DELAYED_QUEUE_NAME = "delay.queue.demo.delay.queue";
    public static final String DELAYED_EXCHANGE_NAME = "delay.queue.demo.delay.exchange";
    public static final String DELAYED_ROUTING_KEY = "delay.queue.demo.delay.routingkey";

    @Bean
    public Queue immediateQueue() {
        return new Queue(DELAYED_QUEUE_NAME);
    }

    @Bean
    public CustomExchange customExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAYED_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding bindingNotify(@Qualifier("immediateQueue") Queue queue,
                                 @Qualifier("customExchange") CustomExchange customExchange) {
        return BindingBuilder.bind(queue).to(customExchange).with(DELAYED_ROUTING_KEY).noargs();
    }
}

三、生产者发送消息

@Component
public class DelayMessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendDelayMsg(String msg, Integer delayTime) {
        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, msg, a ->{
            a.getMessageProperties().setDelay(delayTime);
            return a;
        });
    }
}

 

 

四、设置消费者监听

@Slf4j
@Component
public class DeadLetterQueueConsumer {

    @RabbitListener(queues = DELAYED_QUEUE_NAME)
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("当前时间：{},取消订单,msg:{}",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()), msg); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); 
    } 
}

五、调用接口

@Slf4j
@RequestMapping("rabbitmq")
@RestController
public class RabbitMQMsgController {

    @Autowired
    private DelayMessageSender sender;


    /**
     * 发送延迟取消订单消息
     * @param msg 消息体
     * @param delayTime 自定义延迟取消订单时间（毫秒）
     */
    @RequestMapping("delayMsg")
    public void delayMsg2(String msg, Integer delayTime) {
         msg = msg +"=>"+ (int)(Math.random() * 90000.0D + 10000.0D);
        log.info("当前时间：{},生成订单,msg:{},delayTime:{}", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()), msg, delayTime);
        sender.sendDelayMsg(msg, delayTime);
    }
}

 

输出结果：我设置的是5秒后取消订单

 

rmq配置：