RabbitMQ

RabbitMQ
- 1.對MQ的介紹
- 2.RabbitMQ的六種模式及作業原理
- 3.hello world佇列
- 4.作業佇列模式
- 5.訊息應答機制
- - - 自動應答
    - 手動應答
    - 訊息自動進行重新入隊
- 6.RabbitMQ的持久化,不公平分發及預取值
- 7.發布確認
- 8.交換機
- - <1>交換機的認識
  - - 1.1 概念
    - 1.2Exchanges 的型別
    - 1.3無名Exchange
    - 1.4臨時佇列
    - 1.5佇列和交換機之間的系結
  - <2>交換機具體介紹
- 9.死信佇列
- - <1>認識死信佇列
  - <2>死信實戰
  - - 2.1架構圖
    - 2.2TTL模擬死信佇列

1.對MQ的介紹

說明是MQ

MQ(message queue)，從字面意思上看，本質是個佇列，FIFO 先入先出，只不過佇列中存放的內容是

message 而已，還是一種跨行程的通信機制，用于上下游傳遞訊息，在互聯網架構中，MQ 是一種非常常

見的上下游“邏輯解耦+物理解耦”的訊息通信服務，使用了 MQ 之后，訊息發送上游只需要依賴 MQ，不

用依賴其他服務，

MQ的好處
- 流量消峰
  
  舉個例子，如果訂單系統最多能處理一萬次訂單，這個處理能力應付正常時段的下單時綽綽有余，正常時段我們下單一秒后就能回傳結果，但是在高峰期，如果有兩萬次下單作業系統是處理不了的，只能限制訂單超過一萬后不允許用戶下單，使用訊息佇列做緩沖，我們可以取消這個限制，把一秒內下的訂單分散成一段時間來處理，這時有些用戶可能在下單十幾秒后才能收到下單成功的操作，但是比不能下單的體驗要好，
- 應用解耦
  
  以電商應用為例，應用中有訂單系統、庫存系統、物流系統、支付系統，用戶創建訂單后，如果耦合呼叫庫存系統、物流系統、支付系統，任何一個子系統出了故障，都會造成下單操作例外，當轉變成基于訊息佇列的方式后，系統間呼叫的問題會減少很多，比如物流系統因為發生故障，需要幾分鐘來修復，在這幾分鐘的時間里，物流系統要處理的記憶體被快取在訊息佇列中，用戶的下單操作可以正常完成，當物流系統恢復后，繼續處理訂單資訊即可，用戶感受不到物流系統的故障，提升系統的可用性，

在這里插入圖片描述

異步處理

有些服務間呼叫是異步的，例如 A 呼叫 B，B 需要花費很長時間執行，但是 A 需要知道 B 什么時候可以執行完，以前一般有兩種方式，A 過一段時間去呼叫 B 的查詢 api 查詢，或者 A 提供一個 callback api， B 執行完之后呼叫 api 通知 A 服務，這兩種方式都不是很優雅，使用訊息總線，可以很方便解決這個問題，A 呼叫 B 服務后，只需要監聽 B 處理完成的訊息，當 B 處理完成后，會發送一條訊息給 MQ，MQ 會將此訊息轉發給 A 服務，這樣 A 服務既不用回圈呼叫 B 的查詢 api，也不用提供 callback api，同樣 B 服務也不用做這些操作，A 服務還能及時的得到異步處理成功的訊息，

在這里插入圖片描述

2.RabbitMQ的六種模式及作業原理

作業模式

依次是：hello world ，作業模式，發布訂閱模式，路由模式，主題模式，發布確認模式

在這里插入圖片描述

作業原理

在這里插入圖片描述

Binding：exchange 和 queue 之間的虛擬連接，binding 中可以包含 routing key，Binding 資訊被保

存到 exchange 中的查詢表中，用于 message 的分發依據

依賴

<!--rabbitmq 依賴客戶端-->
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.8.0</version>
</dependency>

3.hello world佇列

在這里插入圖片描述

生產者

public class Producer {
    //建立佇列
    private static final String QUEUE_NAME="hello";
    public static void main(String[] args)  {
        //創建連接工場
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setUsername("guest");
        factory.setUsername("guest");

        try {
            //建立連接和信道
            //channel 實作了自動 close 介面 自動關閉 不需要顯示關閉
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();

            /**
             * 生成一個佇列,并將信道和佇列連接
             * 1.佇列名稱
             * 2.佇列里面的訊息是否持久化 默認訊息存盤在記憶體中
             * 3.該佇列是否只供一個消費者進行消費 是否進行共享 true 可以多個消費者消費
             * 4.是否自動洗掉 最后一個消費者端開連接以后 該佇列是否自動洗掉 true 自動洗掉
             * 5.其他引數
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            String message="hello world";
            /**
             * 發送一個訊息
             * 1.發送到那個交換機
             * 2.路由的 key 是哪個
             * 3.其他的引數資訊
             * 4.發送訊息的訊息體
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("訊息發送成功");
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

消費者

public class Consumer {
    //定義佇列名
    private static final String QUEUE_NAME="hello";

    public static void main(String[] args) {
        //建立連接和信道
        try {
            ConnectionFactory factory=new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();
            System.out.println("等待接收訊息");


            /**
             *1.同一個會話， consumerTag 是固定的 可以做此會話的名字， deliveryTag 每次接收訊息+1，可以做此訊息處理通道的名字，
             *2.包含訊息的位元組形式的類
             */
            DeliverCallback deliverCallback=(consumerTag,delivery)->{
                String message=new String(delivery.getBody());
                System.out.println(message);
            };
            CancelCallback cancelCallback=(consumerTag)->{
                System.out.println("訊息消費被取消");
            };
            /* 消費者消費訊息
             * 1.消費哪個佇列
             * 2.消費成功之后是否要自動應答 true 代表自動應答 false 手動應答
             * 3.消費者未成功消費的回呼
             * 4.消費者取消消費的回呼
             */
            channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);



        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

4.作業佇列模式

在這里插入圖片描述

生產者

public class Producer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("失敗");
            return;
        }
        channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,false,false,false,null);
        int i=0;
        while (true){
            String message="訊息"+i;
            i++;
            /**
             * 發送一個訊息
             * 1.發送到那個交換機
             * 2.路由的 key 是哪個
             * 3.其他的引數資訊
             * 4.發送訊息的訊息體
             */
            channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());
            System.out.println(message);
            Thread.sleep(500);
        }
    }
}

消費者

public class Consumer {
    public static void main(String[] args) {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("消費失敗");
            return;
        }
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message=new String(delivery.getBody());
            System.out.println(Thread.currentThread().getName()+"消費了"+message);
        };
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("訊息消費被取消");
        };
        Thread[] threads=new Thread[5];
        for (int i = 0; i <threads.length ; i++) {
            threads[i]=new Thread(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+"啟動等待消費");
                    channel.basicConsume(RabbitMQChannelUtil.QUEUE_NAME,true,deliverCallback,cancelCallback);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
        for (int i = 0; i <threads.length ; i++) {
            threads[i].start();
        }
    }
}

5.訊息應答機制

認識

消費者處理訊息時，可能在處理程序中掛掉，那么訊息就會丟失為了保證訊息在發送程序中不丟失，rabbitmq 引入訊息應答機制，訊息應答就是:消費者在接收到訊息并且處理該訊息之后，告訴 rabbitmq 它已經處理了rabbitmq 可以把該訊息洗掉了，

自動應答

訊息發送后立即被認為已經傳送成功，這種模式需要在高吞吐量和資料傳輸安全性方面做權衡,因為這種模式如果訊息在接收到之前，消費者那邊出現連接或者 channel 關閉，那么訊息就丟失了

手動應答

Channel.basicAck(用于肯定確認)

RabbitMQ 已知道該訊息并且成功的處理訊息，可以將其丟棄了

Channel.basicNack(用于否定確認)
Channel.basicReject(用于否定確認)

與 Channel.basicNack 相比少一個引數,不處理該訊息了直接拒絕，可以將其丟棄了

Channel.basicNack引數中Multiple(批量應答) 的解釋

multiple 的 true 和 false 代表不同意思

true 代表批量應答 channel 上未應答的訊息

比如說 channel 上有傳送 tag 的訊息 5,6,7,8 當前 tag 是 8 那么此時

5-8 的這些還未應答的訊息都會被確認收到訊息應答

false

只會應答 tag=8 的訊息 5,6,7 這三個訊息依然不會被確認收到訊息應答

在這里插入圖片描述

訊息手動應答的代碼

將手動應答開啟

/* 消費者消費訊息
 * 1.消費哪個佇列
 * 2.消費成功之后是否要自動應答 true 代表自動應答 false 手動應答
 * 3.當一個訊息發送過來后的回呼介面
 * 4.消費者取消消費的回呼
 */
boolean ack=false;
channel.basicConsume(QUEUE_NAME,ack,deliverCallback,cancelCallback);

訊息消費回呼時，使用手動應答

/**
 * 訊息發送過來后的回呼介面
 *1.同一個會話， consumerTag 是固定的 可以做此會話的名字， deliveryTag 每次接收訊息+1，可以做此訊息處理通道的名字，
 *2.訊息類
 */
DeliverCallback deliverCallback=(consumerTag,delivery)->{
    String message=new String(delivery.getBody());
    System.out.println(message);
    /**
     * 引數說明
     * 1.訊息的標記tag
     * 2.是否批量應答
     */
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
};

訊息自動進行重新入隊

如果消費者由于某些原因失去連接(其通道已關閉，連接已關倍訓 TCP 連接丟失)，導致訊息未發送 ACK 確認，RabbitMQ 將了解到訊息未完全處理，并將對其重新排隊，如果此時其他消費者可以處理，它將很快將其重新分發給另一個消費者，這樣，即使某個消費者偶爾死亡，也可以確保不會丟失任何訊息，

在這里插入圖片描述

6.RabbitMQ的持久化,不公平分發及預取值

概念

剛剛我們已經看到了如何處理任務不丟失的情況，但是如何保障當 RabbitMQ 服務停掉以后消

息生產者發送過來的訊息不丟失，默認情況下 RabbitMQ 退出或由于某種原因崩潰時，它忽視佇列

和訊息，除非告知它不要這樣做，確保訊息不會丟失需要做兩件事：我們需要將佇列和訊息都標

記為持久化，

佇列持久化

之前我們創建的佇列都是非持久化的，rabbitmq 如果重啟的化，該佇列就會被洗掉掉，如果

要佇列實作持久化需要在宣告佇列的時候把 durable(第二個) 引數設定為持久化

但是需要注意的就是如果之前宣告的佇列不是持久化的，需要把原先佇列先洗掉，或者重新

創建一個持久化的佇列，不然就會出現錯誤

channel.queueDeclare(RabbitMQChannelUtil.QUEUE_NAME,true,false,false,null);

在這里插入圖片描述

這個就是持久化佇列

訊息持久化

要想讓訊息實作持久化需要在訊息生產者修改代碼，MessageProperties.PERSISTENT_TEXT_PLAIN 添

加這個屬性，

佇列持久化為false時:

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME,null,message.getBytes());

佇列持久化為true時

channel.basicPublish("",RabbitMQChannelUtil.QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

將訊息標記為持久化并不能完全保證不會丟失訊息，盡管它告訴 RabbitMQ 將訊息保存到磁盤，但是

這里依然存在當訊息剛準備存盤在磁盤的時候但是還沒有存盤完，訊息還在快取的一個間隔點，此時并沒

有真正寫入磁盤，持久性保證并不強.更強的持久化后面發布確認會講到

不公平分發

在最開始的時候我們學習到 RabbitMQ 分發訊息采用的輪訓分發，但是在某種場景下這種策略并不是

很好，比方說有兩個消費者在處理任務，其中有個消費者 1 處理任務的速度非常快，而另外一個消費者 2

處理速度卻很慢，這個時候我們還是采用輪訓分發的化就會到這處理速度快的這個消費者很大一部分時間

處于空閑狀態，而處理慢的那個消費者一直在干活，這種分配方式在這種情況下其實就不太好，但是

RabbitMQ 并不知道這種情況它依然很公平的進行分發，

為了避免這種情況，我們設定不公平分發：

channel.basicQos(1);

在這里插入圖片描述

預取值

本身訊息的發送就是異步發送的，所以在任何時候，channel 上肯定不止只有一個訊息另外來自消費

者的手動確認本質上也是異步的，因此這里就存在一個未確認的訊息緩沖區，因此希望開發人員能限制此

緩沖區的大小，以避免緩沖區里面無限制的未確認訊息問題，這個時候就可以通過使用 basic.qos 方法設

置“預取計數”值來完成的，該值定義通道上允許的未確認訊息的最大數量，一旦數量達到配置的數量，

RabbitMQ 將停止在通道上傳遞更多訊息，除非至少有一個未處理的訊息被確認

prefetch就是預取值數

在這里插入圖片描述

7.發布確認

上文持久化中提到，當訊息持久化存入RabbitMQ磁盤時，RabbitMQ突然宕機，則訊息未成功存入，會發生訊息丟失，所以發布確認即：在訊息成功存入磁盤時，返還給生產者一個訊息，確認已經存入磁盤

具體介紹

生產者將信道設定成 confirm 模式，一旦信道進入 confirm 模式，所有在該信道上面發布的

訊息都將會被指派一個唯一的 ID(從 1 開始)，一旦訊息被投遞到所有匹配的佇列之后，broker

就會發送一個確認給生產者(包含訊息的唯一 ID)，這就使得生產者知道訊息已經正確到達目的隊

列了，如果訊息和佇列是可持久化的，那么確認訊息會在將訊息寫入磁盤之后發出，broker 回傳

給生產者的確認訊息中 delivery-tag 域包含了確認訊息的序列號，此外 broker 也可以設定

basic.ack 的 multiple 域，表示到這個序列號之前的所有訊息都已經得到了處理，

為了保證訊息不丟失：

開啟佇列持久化
開啟訊息持久化
開啟信道的發布確認

開啟發布確認的方法

channel.confirmSelect();

發布確認的模式

單個確認發布

public static void singleConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("信道建立失敗");
            return;
        }
        //開啟發布確認
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            //可以加時間引數，當訊息發送失敗或超過引數時間沒成功，則回傳false
            boolean flag=channel.waitForConfirms();
            //如果失敗可以重發
            if(flag){
                System.out.println(message+"發送成功");
            }else {
                //這里可以實作重發
                System.out.println(message+"發送失敗");
            }
        }
        long end=System.currentTimeMillis();
        System.out.println("發送"+MESSAGE_COUNT+"條訊息，耗時"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

發布一個訊息之后只有它被確認發布，后續的訊息才能繼續發布,waitForConfirmsOrDie(long)這個方法只有在訊息被確認的時候才回傳，如果在指定時間范圍內這個訊息沒有被確認那么它將拋出例外，

缺點：速度太慢

批量發布確認模式

public static void batchConfirm(){
    try {
        Channel channel=RabbitMQChannelUtil.getChannel();
        if(channel == null){
            System.out.println("建立連接失敗");
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //當100條訊息發布成功時，再確認
        int ackMessageCount=100;
        //未確認的訊息個數
        int needAckMessageCount=0;
        //開啟發布確認
        channel.confirmSelect();
        long begin=System.currentTimeMillis();
        for (int i = 0; i <MESSAGE_COUNT ; i++) {
            String message=i+"";
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            needAckMessageCount++;
            if(needAckMessageCount == ackMessageCount){
                //確認
                channel.waitForConfirms();
                needAckMessageCount=0;
            }
        }
        //判斷可能還有訊息未發送，再發送依次
        if(needAckMessageCount > 0){
            channel.waitForConfirms();
        }
        long end= System.currentTimeMillis();
        System.out.println("發送"+MESSAGE_COUNT+"條訊息，耗時"+(end-begin)+"ms");
    }catch (Exception e){
        e.printStackTrace();
    }
}

缺點:當發生故障導致發布出現問題時，不知道是哪個訊息出現問題

異步確認發布

原理

有單獨一個佇列保存確認信號

public static void asyncConfirm() throws Exception {
    try (Channel channel = RabbitMQChannelUtil.getChannel()) {
        if(channel == null){
            return;
        }
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //開啟發布確認
        channel.confirmSelect();
        /**
         * 執行緒安全有序的一個哈希表，適用于高并發的情況
         * 1.輕松的將序號與訊息進行關聯
         * 2.輕松批量洗掉條目 只要給到序列號
         * 3.支持并發訪問
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new
                ConcurrentSkipListMap<>();
        /**
         * 確認收到訊息的一個回呼
         * 1.訊息序列號
         * 2.true 可以確認小于等于當前序列號的訊息
         * false 確認當前序列號訊息
         */
        ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
            if (multiple) {
                //回傳的是小于等于當前序列號的未確認訊息 是一個 map
                ConcurrentNavigableMap<Long, String> confirmed =
                        outstandingConfirms.headMap(sequenceNumber, true);
                //清除該部分未確認訊息
                confirmed.clear();
            }else{
                //只清除當前序列號的訊息
                outstandingConfirms.remove(sequenceNumber);
            }
        };
        ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("發布的訊息"+message+"未被確認，序列號"+sequenceNumber);
        };
        /**
         * 添加一個異步確認的監聽器
         * 1.確認收到訊息的回呼
         * 2.未收到訊息的回呼
         */
        channel.addConfirmListener(ackCallback, null);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "訊息" + i;
            /**
             * channel.getNextPublishSeqNo()獲取下一個訊息的序列號
             * 通過序列號與訊息體進行一個關聯
             * 全部都是未確認的訊息體
             */
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        }
        long end = System.currentTimeMillis();
        System.out.println("發布" + MESSAGE_COUNT + "個異步確認訊息,耗時" + (end - begin) +
                "ms");
    } 
}

8.交換機

<1>交換機的認識

1.1 概念

RabbitMQ 訊息傳遞模型的核心思想是: 生產者生產的訊息從不會直接發送到佇列，實際上，通常生產

者甚至都不知道這些訊息傳遞傳遞到了哪些佇列中，

相反，生產者只能將訊息發送到交換機(exchange)，交換機作業的內容非常簡單，一方面它接收來

自生產者的訊息，另一方面將它們推入佇列，交換機必須確切知道如何處理收到的訊息，是應該把這些消

息放到特定佇列還是說把他們到許多佇列中還是說應該丟棄它們，這就的由交換機的型別來決定，

在這里插入圖片描述

1.2Exchanges 的型別

總共有以下型別：

直接(direct), 主題(topic) ,標題(headers) , 扇出(fanout)

1.3無名Exchange

channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

第一個引數是交換機的名稱，空字串表示默認或無名稱交換機：訊息能路由發送到佇列中其實

是由 routingKey(bindingkey)系結 key 指定的，如果它存在的話

1.4臨時佇列

每當我們連接到 Rabbit 時，我們都需要一個全新的空佇列，為此我們可以創建一個具有隨機名稱的佇列，或者能讓服務器為我們選擇一個隨機佇列名稱那就更好了，其次一旦我們斷開了消費者的連接，佇列將被自動洗掉，

String queueName = channel.queueDeclare().getQueue();

1.5佇列和交換機之間的系結

//宣告交換機名稱及型別
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//把該臨時佇列系結我們的 exchange 其中 routingkey(也稱之為 binding key)為空字串
 channel.queueBind(queueName, EXCHANGE_NAME, "");

<2>交換機具體介紹

Fanout 洗掉(廣播)

將接收到的所有訊息廣播到它知道的所有佇列中，

Direct (直接)

將詳細發送到對應路由鍵的佇列上去

在這里插入圖片描述

在上面這張圖中，我們可以看到 X 系結了兩個佇列，系結型別是 direct，佇列 Q1 系結鍵為 orange，

佇列 Q2 系結鍵有兩個:一個系結鍵為 black，另一個系結鍵為 green.

在這種系結情況下，生產者發布訊息到 exchange 上，系結鍵為 orange 的訊息會被發布到佇列

Q1，系結鍵為 blackgreen 和的訊息會被發布到佇列 Q2，其他訊息型別的訊息將被丟棄，

系結

//宣告交換機名稱及型別
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//把該臨時佇列系結我們的 exchange 其中 routingkey(也稱之為 binding key)為空字串
 channel.queueBind(queueName, EXCHANGE_NAME, "");

Topics(主題)

盡管使用 direct 交換機改進了我們的系統，但是它仍然存在局限性-比方說我們想接收的日志型別有

info.base 和 info.advantage，某個佇列只想 info.base 的訊息，那這個時候 direct 就辦不到了，這個時候

就只能使用 topic 型別

發送到型別是 topic 交換機的訊息的 routing_key 不能隨意寫，必須滿足一定的要求，它**必須是一個單

詞串列，以點號分隔開，這些單詞可以是任意單詞，但這個單詞串列最多不能超過 255 個位元組，
- 可以代替一個單詞
- #可以替代零個或多個單詞

在這里插入圖片描述

9.死信佇列

<1>認識死信佇列

概念

死信，顧名思義就是無法被消費的訊息，字面意思可以這樣理解，一般來說，producer 將訊息投遞到 broker 或者直接到 queue 里了，consumer 從 queue 取出訊息進行消費，但某些時候由于特定的原因導致 queue中的某些訊息無法被消費，這樣的訊息如果沒有后續的處理，就變成了死信，有死信自然就有了死信佇列，
應用場景:為了保證訂單業務的訊息資料不丟失，需要使用到 RabbitMQ 的死信佇列機制，當訊息消費發生例外時，將訊息投入死信佇列中.還有比如說: 用戶在商城下單成功并點擊去支付后在指定時間未支付時自動失效

來源

訊息超出最大存活時間過期
佇列達到最大長度(佇列滿了，無法再添加資料到 mq 中)
訊息被拒絕(basic.reject 或 basic.nack)并且 requeue=false.

<2>死信實戰

2.1架構圖

在這里插入圖片描述

2.2TTL模擬死信佇列

生產者

public class Producer {
    private static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main(String[] args) {
        try {
            Channel channel= RabbitMQChannelUtil.getChannel();
            if(channel == null){
                return;
            }
            //宣告交換機型別
            channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
            //設定訊息TTL時間
            AMQP.BasicProperties basicProperties=new AMQP.BasicProperties().builder().expiration("1000").build();
            //用作演示訊息佇列的限制個數
            for (int i = 0; i <10 ; i++) {
                String message="info"+i;
                channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes());
                System.out.println("生產者發送訊息");
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

普通消費者:啟動之后關閉，模擬接收不到訊息

public class NormalConsumer {
    //普通交換機名稱
    private static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交換機名稱
    private static final String DEAD_EXCHANGE = "dead_exchange";
    public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if(channel == null){
            return;
        }
        //宣告死信和普通交換機 型別為 direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        //宣告死信佇列
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        //死信佇列系結死信交換機與 routingkey
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
        //正常佇列系結死信佇列資訊
        Map<String, Object> params = new HashMap<>();
        //正常佇列設定死信交換機 引數 key 是固定值
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        //正常佇列設定死信 routing-key 引數 key 是固定值
        params.put("x-dead-letter-routing-key", "lisi");

        String normalQueue = "normal-queue";
        channel.queueDeclare(normalQueue, false, false, false, params);
        channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
        System.out.println("等待接收訊息.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("NormalConsumer 接收到訊息"+message);
        };
        channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
        });
    }
}

死信佇列消費者

public class DeadConsumer {
 private static final String DEAD_EXCHANGE = "dead_exchange";
 public static void main(String[] argv) throws Exception {
        Channel channel = RabbitMQChannelUtil.getChannel();
        if (channel == null) {
         return;
        }
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
        String deadQueue = "dead-queue";
        channel.queueDeclare(deadQueue, false, false, false, null);
        channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
        System.out.println("等待接收死信佇列訊息.....");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
         String message = new String(delivery.getBody(), "UTF-8");
         System.out.println("DeadConsumer 接收死信佇列的訊息" + message);
        };
        channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
        });
 }

另外兩種思路相同.

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/295126.html

標籤：其他

上一篇：elk + logback搭建

下一篇：Elasticsearch又雙⑸柿閑孤讀耍?00萬條“禁飛名單”被泄露

RabbitMQ基礎總結

RabbitMQ

目錄

1.對MQ的介紹

2.RabbitMQ的六種模式及作業原理

3.hello world佇列

4.作業佇列模式

5.訊息應答機制

自動應答

手動應答

訊息自動進行重新入隊

6.RabbitMQ的持久化,不公平分發及預取值

7.發布確認

8.交換機

<1>交換機的認識

1.1 概念

1.2Exchanges 的型別

1.3無名Exchange

1.4臨時佇列

1.5佇列和交換機之間的系結

<2>交換機具體介紹

9.死信佇列

<1>認識死信佇列

<2>死信實戰

2.1架構圖

2.2TTL模擬死信佇列

RabbitMQ基礎總結

RabbitMQ

目錄

1.對MQ的介紹

2.RabbitMQ的六種模式 及作業原理

3.hello world佇列

4.作業佇列模式

5.訊息應答機制

自動應答

手動應答

訊息自動進行重新入隊

6.RabbitMQ的持久化,不公平分發及預取值

7.發布確認

8.交換機

<1>交換機的認識

1.1 概念

1.2Exchanges 的型別

1.3無名Exchange

1.4臨時佇列

1.5佇列和交換機之間的系結

<2>交換機具體介紹

9.死信佇列

<1>認識死信佇列

<2>死信實戰

2.1架構圖

2.2TTL模擬死信佇列

2.RabbitMQ的六種模式及作業原理