详解RabbitMQ中延迟队列结合业务场景的使用

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

@Component
public class RabbitMQConfig {
    public static final String EXCHANGE_NAME = "order_exchange_name";
    public static final String QUEUE_NAME = "order_queue_name";
    public static final String DELAY_EXCHANGE_NAME = "delay_exchange_name";
    public static final String DELAY_QUEUE_NAME = "delay_queue_name";
    @Bean("orderExchange")
    public Exchange testExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }
    @Bean("delayExchange")
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange(DELAY_EXCHANGE_NAME).durable(true).build();
    }
    //订单队列绑定延迟交换机并且带上路由键
    @Bean("orderQueue")
    public Queue testQueue(){
        return QueueBuilder.durable(QUEUE_NAME).deadLetterExchange(DELAY_EXCHANGE_NAME)
                .deadLetterRoutingKey("order.delay.user").build();
    }
    @Bean("delayQueue")
    public Queue deadQueue(){
        return QueueBuilder.durable(DELAY_QUEUE_NAME).build();
    }
    @Bean
    public Binding link(@Qualifier("orderExchange") Exchange exchange,
                        @Qualifier("orderQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs();
    }
    @Bean
    public Binding deadLink(@Qualifier("delayExchange") Exchange exchange,
                            @Qualifier("delayQueue") Queue queue){
        return BindingBuilder.bind(queue).to(exchange).with("order.delay.#").noargs();
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
class RabbitmqProducerApplicationTests {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testProducer() throws InterruptedException {
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b) System.out.println("交换机成功接受到了消息");
                else System.out.println("消息失败原因" + s);
            }
        });
        // 设置交换机处理失败消息的模式
        // true:消息到达不了队列时 会将消息重新返回给生产者 false:消息到达不了队列直接丢弃
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("队列接受不到交换机的消息进行了失败回调");
            }
        });
        // 以上代码均是为了保证消息的可靠性传递
        // 对消息进行后置处理 设置其过期时间为10s
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("10000");
                return message;
            }
        };
        // 下单成功发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.delay.user","ikun书籍", messagePostProcessor);
    }
}

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.5.0</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
    </dependency>
</dependencies>

@Slf4j
@Component
public class OrderListener implements ChannelAwareMessageListener {
    @RabbitListener(queues = "delay_queue_name") // 监听的是死信队列
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 这里要注意在监听这条消息前 肯定会有接口点击支付去更改支付状态的业务逻辑 此处我是做了10s的订单业务而已
            log.info("您在时间为：{},时有一条订单为：{}", LocalDateTime.now().minusSeconds(10), new String(message.getBody()));
            // 下面开始接受订单消息逻辑
            log.info("将订单id传入数据库查询订单支付字段");
            log.info("字段为支付成功状态就手动确认签收");
            log.info("字段为未支付状态就取消订单并且回滚事务");
            channel.basicAck(deliveryTag,false);// 仅确认本次消息
        } catch (Exception e){
            log.info("出现异常 拒绝签收消息 并且不重回队列");
            channel.basicNack(deliveryTag,false,false);
        }
    }
}