2. 程式人生 > 實用技巧 >Rabbitmq的死信佇列和延時佇列

Rabbitmq的死信佇列和延時佇列

阿新 • • 發佈：2020-08-23

一、死信佇列

死信佇列其實和普通的佇列沒啥大的區別，都需要建立自己的Queue、Exchange,然後通過RoutingKey繫結到Exchange上去，只不過死信佇列的RoutingKey和Exchange要作為引數，繫結到正常的佇列上去，一種應用場景是正常佇列裡面的訊息被basicNack或者reject時，訊息就會被路由到正常佇列繫結的死信佇列中，還有一種還有常用的場景就是開啟了自動簽收，然後消費者消費訊息時出現異常，超過了重試次數，那麼這條訊息也會進入死信佇列，如果配置了話，當然還有其他的應用場景，這裡不一一討論。

1.1、死信佇列和交換器配置

這裡有兩個佇列，正常的業務佇列emailQueue

和與之繫結的死信佇列，這裡只演示，手動簽收，消費者捕獲異常Nack

1.1.2、yml配置

spring:
 rabbitmq:
    host: 192.168.99.12
    port: 5672
    username: guest
    password: guest
    # 傳送確認
    publisher-confirms: true
    # 路由失敗回撥
    publisher-returns: true
    template:
        # 必須設定成true 訊息路由失敗通知監聽者，false 將訊息丟棄
        mandatory: true
    listener:
      simple:
        # 每次從RabbitMQ獲取的訊息數量
        prefetch: 1
        default-requeue-rejected: false
        # 每個佇列啟動的消費者數量
        concurrency: 1
        # 每個佇列最大的消費者數量
        max-concurrency: 1
        # 簽收模式為手動簽收-那麼需要在程式碼中手動ACK
        acknowledge-mode: manual
#郵件佇列
email:
  queue:
    name: demo.email
	
#郵件交換器名稱
exchange:
  name: demoTopicExchange

#死信佇列
dead:
  letter:
    queue:
      name: demo.dead.letter
    exchange:
      name: demoDeadLetterTopicExchange

1.1.3、死信佇列配置

/**
 * rabbitmq 配置
 *
 * @author DUCHONG
 * @since 2020-08-23 14:05
 **/
@Configuration
@Slf4j
public class RabbitmqConfig {


    @Value("${email.queue.name}")
    private String emailQueue;
    @Value("${exchange.name}")
    private String topicExchange;
    @Value("${dead.letter.queue.name}")
    private String deadLetterQueue;
    @Value("${dead.letter.exchange.name}")
    private String deadLetterExchange;

    @Bean
    public Queue emailQueue() {

        Map<String, Object> arguments = new HashMap<>(2);
        // 繫結死信交換機
        arguments.put("x-dead-letter-exchange", deadLetterExchange);
        // 繫結死信的路由key
        arguments.put("x-dead-letter-routing-key", deadLetterQueue+".#");

        return new Queue(emailQueue,true,false,false,arguments);
    }

	
    @Bean
    TopicExchange emailExchange() {
        return new TopicExchange(topicExchange);
    }


    @Bean
    Binding bindingEmailQueue() {
        return BindingBuilder.bind(emailQueue()).to(emailExchange()).with(emailQueue+".#");
    }

    
    //私信佇列和交換器
    @Bean
    public Queue deadLetterQueue() {
        return new Queue(deadLetterQueue);
    }

    @Bean
    TopicExchange deadLetterExchange() {
        return new TopicExchange(deadLetterExchange);
    }

    @Bean
    Binding bindingDeadLetterQueue() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(deadLetterQueue+".#");
    }

}

1.2、訊息傳送方

@Configuration
@EnableScheduling
@Slf4j
public class ScheduleController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Value("${exchange.name}")
    private String topicExchange;

    @Scheduled(cron = "0 0/2 * * * ?")
    public void sendEmailMessage() {

        String msg = RandomStringUtils.randomAlphanumeric(8);
        JSONObject email=new JSONObject();
        email.put("content",msg);
        email.put("to","[email protected]");
        CorrelationData correlationData=new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(topicExchange,"demo.email.x",email.toJSONString(),correlationData);
        log.info("---傳送 email 訊息---{}---messageId---{}",email,correlationData.getId());
    }


}

1.3、訊息消費方

@Component
@Slf4j
public class MessageHandler {

    
   /**
     * 郵件消費者
     * @param message
     * @param channel
     * @param headers
     * @throws IOException
     */
    @RabbitListener(queues ="demo.email")
    @RabbitHandler
    public void handleEmailMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {

        try {

            String msg=new String(message.getBody(), CharEncoding.UTF_8);
            JSONObject jsonObject = JSON.parseObject(msg);
            jsonObject.put("messageId",headers.get("spring_returned_message_correlation"));
            log.info("---接受到訊息---{}",jsonObject);
			//主動異常
			int m=1/0;
            //手動簽收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
        catch (Exception e) {
            log.info("handleEmailMessage捕獲到異常,拒絕重新入隊---訊息ID---{}",headers.get("spring_returned_message_correlation"));
            //異常，ture 重新入隊,或者false,進入死信佇列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);

        }
    }

    /**
     * 死信消費者，自動簽收開啟狀態下，超過重試次數，或者手動簽收，reject或者Nack
     * @param message
     */
    @RabbitListener(queues = "demo.dead.letter")
    public void handleDeadLetterMessage(Message message, Channel channel,@Headers Map<String,Object> headers) throws IOException {

        //可以考慮資料庫記錄，每次進來查數量，達到一定的數量，進行預警，人工介入處理
        log.info("接收到死信訊息:---{}---訊息ID---{}", new String(message.getBody()),headers.get("spring_returned_message_correlation"));
		//回覆ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

 
}

1.4、結果

二、延時佇列

延時佇列顧名思義，不是及時的佇列，也就是傳送者發給的訊息要延時一段時間，消費者才能接受的到，這裡有個典型的應用場景就是訂單30分鐘內未支付就關閉訂單，當然死信佇列也是可以實現的，這裡只演示訊息的延時消費邏輯，訂單邏輯就一個判斷，這裡不做討論。

2.1、延時佇列和交換器配置

使用延時佇列之前，需要先安裝延時佇列外掛，安裝方法，前面已經介紹過了，這裡放個連結

延時佇列外掛安裝

2.1.1、yml配置

spring:
    rabbitmq:
        host: 192.168.99.12
        port: 5672
        username: guest
        password: guest
        # 傳送確認
        publisher-confirms: true
        # 路由失敗回撥
        publisher-returns: true
        template:
            # 必須設定成true 訊息路由失敗通知監聽者，false 將訊息丟棄
            mandatory: true
        #消費端
        listener:
            simple:
                # 每次從RabbitMQ獲取的訊息數量
                prefetch: 1
                default-requeue-rejected: false
                # 每個佇列啟動的消費者數量
                concurrency: 1
                # 每個佇列最大的消費者數量
                max-concurrency: 1
                # 簽收模式為手動簽收-那麼需要在程式碼中手動ACK
                acknowledge-mode: manual
#郵件佇列
email:
    queue:
        name: demo.email

#郵件交換器名稱
exchange:
    name: demoTopicExchange

#死信佇列
dead:
    letter:
        queue:
            name: demo.dead.letter
        exchange:
            name: demoDeadLetterTopicExchange

#延時佇列
delay:
    queue:
        name: demo.delay
    exchange:
        name: demoDelayTopicExchange

2.1.2、延時佇列配置

/**
 * rabbitmq 配置
 *
 * @author DUCHONG
 * @since 2020-08-23 14:05
 **/
@Configuration
@Slf4j
public class RabbitmqConfig {


    @Value("${email.queue.name}")
    private String emailQueue;
    @Value("${exchange.name}")
    private String topicExchange;
    @Value("${dead.letter.queue.name}")
    private String deadLetterQueue;
    @Value("${dead.letter.exchange.name}")
    private String deadLetterExchange;
    @Value("${delay.queue.name}")
    private String delayQueue;
    @Value("${delay.exchange.name}")
    private String delayExchange;

    @Bean
    public Queue emailQueue() {

        Map<String, Object> arguments = new HashMap<>(2);
        // 繫結死信交換機
        arguments.put("x-dead-letter-exchange", deadLetterExchange);
        // 繫結死信的路由key
        arguments.put("x-dead-letter-routing-key", deadLetterQueue+".#");

        return new Queue(emailQueue,true,false,false,arguments);
    }


    @Bean
    TopicExchange emailExchange() {
        return new TopicExchange(topicExchange);
    }


    @Bean
    Binding bindingEmailQueue() {
        return BindingBuilder.bind(emailQueue()).to(emailExchange()).with(emailQueue+".#");
    }


    //私信佇列和交換器
    @Bean
    public Queue deadLetterQueue() {
        return new Queue(deadLetterQueue);
    }

    @Bean
    TopicExchange deadLetterExchange() {
        return new TopicExchange(deadLetterExchange);
    }

    @Bean
    Binding bindingDeadLetterQueue() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(deadLetterQueue+".#");
    }
    //延時佇列
    @Bean
    public Queue delayQueue() {
        return new Queue(delayQueue);
    }

    @Bean
    CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "topic");
        //引數二為型別：必須是x-delayed-message
        return new CustomExchange(delayExchange, "x-delayed-message", true, false, args);

    }

    @Bean
    Binding bindingDelayQueue() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(delayQueue+".#").noargs();
    }
}

2.2、訊息傳送方

30分鐘時間太久了，這裡延時2分鐘來看效果

@Configuration
@EnableScheduling
@Slf4j
public class ScheduleController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Value("${exchange.name}")
    private String topicExchange;

    @Value("${delay.exchange.name}")
    private String delayTopicExchange;

    @Scheduled(cron = "0 0/1 * * * ?")
    public void sendEmailMessage() {

        String msg = RandomStringUtils.randomAlphanumeric(8);
        JSONObject email=new JSONObject();
        email.put("content",msg);
        email.put("to","[email protected]");
        CorrelationData correlationData=new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(topicExchange,"demo.email.x",email.toJSONString(),correlationData);
        log.info("---傳送 email 訊息---{}---messageId---{}",email,correlationData.getId());
    }


    @Scheduled(cron = "0 0/1 * * * ?")
    public void sendDelayOrderMessage() throws Exception{

        //訂單號 id實際是儲存訂單後返回的，這裡用uuid代替
        String orderId = UUID.randomUUID().toString();
        // 模擬訂單資訊
        JSONObject order=new JSONObject();
        order.put("orderId",orderId);
        order.put("goodsName","vip充值");
        order.put("orderAmount","99.00");
        CorrelationData correlationData=new CorrelationData(orderId);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setMessageId(orderId);
        //30分鐘時間太長，這裡延時120s消費
        messageProperties.setHeader("x-delay", 120000);
        Message message = new Message(order.toJSONString().getBytes(CharEncoding.UTF_8), messageProperties);

        rabbitTemplate.convertAndSend(delayTopicExchange,"demo.delay.x",message,correlationData);

        log.info("---傳送 order 訊息---{}---orderId---{}",order,correlationData.getId());
        //睡一會，為了看延遲效果
        TimeUnit.MINUTES.sleep(10);
    }
}

2.3、訊息消費方

@Component
@Slf4j
public class MessageHandler {


    /**
     * 郵件傳送
     * @param message
     * @param channel
     * @param headers
     * @throws IOException
     */
    @RabbitListener(queues ="demo.email")
    @RabbitHandler
    public void handleEmailMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {

        try {

            String msg=new String(message.getBody(), CharEncoding.UTF_8);
            JSONObject jsonObject = JSON.parseObject(msg);
            jsonObject.put("messageId",headers.get("spring_returned_message_correlation"));
            log.info("---接受到訊息---{}",jsonObject);
			//主動異常
			int m=1/0;
            //手動簽收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
        catch (Exception e) {
            log.info("handleEmailMessage捕獲到異常,拒絕重新入隊---訊息ID---{}", headers.get("spring_returned_message_correlation"));
            //異常，ture 重新入隊,或者false,進入死信佇列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);

        }
    }

    /**
     * 死信消費者，自動簽收開啟狀態下，超過重試次數，或者手動簽收，reject或者Nack
     * @param message
     */
    @RabbitListener(queues = "demo.dead.letter")
    public void handleDeadLetterMessage(Message message, Channel channel,@Headers Map<String,Object> headers) throws IOException {

        //可以考慮資料庫記錄，每次進來查數量，達到一定的數量，進行預警，人工介入處理
        log.info("接收到死信訊息:---{}---訊息ID---{}", new String(message.getBody()),headers.get("spring_returned_message_correlation"));
		//回覆ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    /**
     * 延時佇列消費
     * @param message
     * @param channel
     * @param headers
     * @throws IOException
     */
    @RabbitListener(queues ="demo.delay")
    @RabbitHandler
    public void handleOrderDelayMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {

        try {

            String msg=new String(message.getBody(), CharEncoding.UTF_8);
            JSONObject jsonObject = JSON.parseObject(msg);
            log.info("---接受到訂單訊息---orderId---{}",message.getMessageProperties().getMessageId());
            log.info("---訂單資訊---order---{}",jsonObject);
            //業務邏輯，根據訂單id獲取訂單資訊，如果還未支付，設定關閉狀態，如果已支付，不做任何處理
            //手動簽收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
        catch (Exception e) {
            log.info("handleOrderDelayMessage捕獲到異常,重新入隊---orderId---{}", headers.get("spring_returned_message_correlation"));
            //異常，ture 重新入隊,或者false,進入死信佇列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);

        }
    }

}

2.4、結果

執行結果顯示，同一個訂單號的訊息，傳送過後2分鐘，消費者才接受到，符合預期。

相關推薦

Rabbitmq的死信佇列和延時佇列

一、死信佇列 死信佇列其實和普通的佇列沒啥大的區別，都需要建立自己的Queue、Exchange,然後通過RoutingKey繫結到Exchange上去，只不過死信佇列的RoutingKey和Exchange要作為引數，繫結到正常的佇列上去，一種應用場

RabbitMQ從零到叢集高可用.NetCore(.NET5) - 死信佇列，延時佇列（轉）

一、死信佇列 描述：Q1佇列綁定了x-dead-letter-exchange(死信交換機)為X2，x-dead-letter-routing-key（死信路由key)指向Q2(佇列2)

面試官：RabbitMQ過期時間設定、死信佇列、延時佇列怎麼設計？

哈嘍！大家好，我是小奇，一位不靠譜的程式設計師 小奇打算以輕鬆幽默的對話方式來分享一些技術，如果你覺得通過小奇的文章學到了東西，那就給小奇一個贊吧

RabbitMQ死性佇列，延時佇列

1.前言 之前一直聽說過RabbitMQ的死性佇列，延時佇列，但是沒有深入的瞭解，結合自己實際工作中使用到的RabbitMQ，現在整理出來當做個人學習筆記。

TP6(thinkphp6)佇列與延時佇列

安裝 在此我就不再略過TP6的專案建立過程了，大致就是安裝composer工具，安裝成功以後，再使用composer去建立專案即可。

運用.NetCore例項講解RabbitMQ死信佇列,延時佇列

目錄一、死信佇列二、延時佇列三、延時訊息設定不同過期時間四、延時訊息用延時外掛的方式實現一、死信佇列

SpringBoot使用RabbitMQ延時佇列(小白必備)

1.什麼是MQ MQ，是一種跨程序的通訊機制，用於上下游傳遞訊息。 在網際網路架構中，MQ是一種非常常見的上下游“邏輯解耦+物理解耦”的訊息通訊服務。

rabbitmq 延時佇列 外掛方式實現 每條訊息都延時自己時間

上篇文章的延時是加到佇列上的 通過死信過時推送 ，缺點就是不能每條訊息定義自己的過時時間而且每次有新的過時時間，要新建一個交換機和佇列

【日常摘要】- RabbitMq實現延時佇列

簡介 什麼是延時佇列？ 一種帶有延遲功能的訊息佇列 過程： 使用場景 比如存在某個業務場景

springboot-rabbitmq:實現延時佇列

延時佇列應用於什麼場景 延時佇列顧名思義，即放置在該佇列裡面的訊息是不需要立即消費的，而是等待一段時間之後取出消費。那麼，為什麼需要延遲消費呢？我們來看以下的場景 網上商城下訂單後30分鐘後沒有完成支付，

Springboot整合Rabbitmq實現延時佇列之rabbitmq_delayed_message_exchange外掛方式

很多時候我們想定時去做某件事情的時候我們會首先想到定時任務，quartz是個不錯的選擇，但是也有缺點，假如配置在專案中，叢集部署會有重複執行的問題，如果持久化在mysql中，解決了叢集的問題，但是過於依賴mysql，

docker安裝rabbitmq延時佇列外掛

docker安裝rabbitMQ延時佇列外掛（delayed_message_exchange） 1. 查詢Docker容器中的RabbitMQ映象

RabbitMQ延時佇列應用場景

應用場景 我們系統未付款的訂單，超過一定時間後，需要系統自動取消訂單並釋放佔有物品

基於golang的簡單分散式延時佇列服務的實現

一、引言 背景 我們在做系統時，很多時候是處理實時的任務，請求來了馬上就處理，然後立刻給使用者以反饋。但有時也會遇到非實時的任務，比如確定的時間點發布重要公告。或者需要在使用者做了一件事情的X分鐘/Y小時

一口氣說出Java 6種延時佇列的實現方法(面試官也得服)

五一期間原計劃是寫兩篇文章，看一本技術類書籍，結果這五天由於自律性過於差，禁不住各種誘惑，我連電腦都沒開啟過，計劃完美宣告失敗。所以在這能看出和大佬之間的差距，人家沒白沒夜的更文，比你優秀的人比你更努

延時佇列 DelayQueue

當用戶超時未支付時，給使用者發提醒訊息。另一種場景是，超時未付款，訂單自動取消。通常，訂單建立的時候可以向延遲佇列種插入一條訊息，到時間自動執行。其實，也可以用臨時表，把這些未支付的訂單放到一個臨時表

DelayQueue延時佇列

一、DelayQueue是什麼 　　DelayQueue是一個無界的BlockingQueue，用於放置實現了Delayed介面的物件，其中的物件只能在其到期時才能從佇列中取走。這種佇列是有序的，即隊頭物件的延遲到期時間最長。注意：不能將nul

詳解Java中的延時佇列 DelayQueue

當用戶超時未支付時，給使用者發提醒訊息。另一種場景是，超時未付款，訂單自動取消。通常，訂單建立的時候可以向延遲佇列種插入一條訊息，到時間自動執行。其實，也可以用臨時表，把這些未支付的訂單放到一個臨時表

延時佇列 DelayQueue 的簡單使用

技術標籤：佇列java 使用場景：物聯網應用中，硬體版本升級。需要通過線上操作，對某一型別裝置，在某個時間點進行升級。

延時佇列的6個實現方案

一、延時佇列的應用 什麼是延時佇列？顧名思義：首先它要具有佇列的特性，再給它附加一個延遲消費佇列訊息的功能，也就是說可以指定佇列中的訊息在哪個時間點被消費。延時佇列在專案中的應用還是比較多的，尤其像電商