快捷導(dǎo)航

詳解RabbitMQ中延遲隊列結(jié)合業(yè)務(wù)場景的使用

更新時間：2023年05月19日 11:13:17 作者：小白的救贖

這篇文章主要介紹了詳解RabbitMQ中延遲隊列結(jié)合業(yè)務(wù)場景的使用，延遲隊列中的元素都是帶有時間屬性的，延遲隊列就是用來存放需要在指定時間被處理的元素的隊列，需要的朋友可以參考下

消息進入隊列后不會立即被消費，只有到達指定時間后才會被消費。業(yè)務(wù)場景就是支付時間內(nèi)未支付就清除訂單或者用戶注冊一段時間后發(fā)短信問候。在這里想說的是這只是一種思想，并不是真正的一種用法，這種思想所需要的用法就是用上消息TTL存活時間以及死信隊列來實現(xiàn)。

生產(chǎn)者端

目錄結(jié)構(gòu)

導(dǎo)入依賴

<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>

修改yml

spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
publisher-returns: true # 開啟退回回調(diào)
#三個類型：none默認不開啟確認回調(diào) correlated開啟確認回調(diào)
#simple也會確認回調(diào) 還會調(diào)用waitForConfirms()方法或waitForConfirmsOrDie()方法
publisher-confirm-type: correlated # 開啟確認回調(diào)

業(yè)務(wù)邏輯

@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();
    }
    //訂單隊列綁定延遲交換機并且?guī)下酚涉I
    @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);
            }
        });
        // 設(shè)置交換機處理失敗消息的模式
        // true:消息到達不了隊列時 會將消息重新返回給生產(chǎn)者 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("隊列接受不到交換機的消息進行了失敗回調(diào)");
            }
        });
        // 以上代碼均是為了保證消息的可靠性傳遞
        // 對消息進行后置處理 設(shè)置其過期時間為10s
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("10000");
                return message;
            }
        };
        // 下單成功發(fā)送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.delay.user","ikun書籍", messagePostProcessor);
    }
}

測試結(jié)果

消費者端

目錄結(jié)構(gòu)

導(dǎo)入依賴

<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>

修改yml

spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
simple:
acknowledge-mode: manual # 開啟手動確認

業(yè)務(wù)邏輯

@Slf4j
@Component
public class OrderListener implements ChannelAwareMessageListener {
    @RabbitListener(queues = "delay_queue_name") // 監(jiān)聽的是死信隊列
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 這里要注意在監(jiān)聽這條消息前 肯定會有接口點擊支付去更改支付狀態(tài)的業(yè)務(wù)邏輯 此處我是做了10s的訂單業(yè)務(wù)而已
            log.info("您在時間為：{},時有一條訂單為：{}", LocalDateTime.now().minusSeconds(10), new String(message.getBody()));
            // 下面開始接受訂單消息邏輯
            log.info("將訂單id傳入數(shù)據(jù)庫查詢訂單支付字段");
            log.info("字段為支付成功狀態(tài)就手動確認簽收");
            log.info("字段為未支付狀態(tài)就取消訂單并且回滾事務(wù)");
            channel.basicAck(deliveryTag,false);// 僅確認本次消息
        } catch (Exception e){
            log.info("出現(xiàn)異常 拒絕簽收消息 并且不重回隊列");
            channel.basicNack(deliveryTag,false,false);
        }
    }
}