Java RabbitMQ高階特性詳細分析

2022-08-08 18:00:43

訊息的可靠投遞

在使用 RabbitMQ 的時候，作為訊息傳送方希望杜絕任何訊息丟失或者投遞失敗場景。RabbitMQ 為我們提供了兩種方式用來控制訊息的投遞可靠性模式。

confirm 確認模式
return 退回模式

rabbitmq整個訊息投遞的路徑為：

producer—>rabbitmq broker—>exchange—>queue—>consumer

訊息從producer到exchange則會返回一個confirmCallback
訊息從exchange—>queue投遞失敗則會返回一個returnCallback

我們可以利用這兩個callback控制訊息的可靠性投遞

確認模式

訊息從 producer 到 exchange 則會返回一個 confirmCallback

以spring整合rabbitmq為例，修改rabbitmq組態檔,在connectionFactory中新增publisher-confirms屬性並設定值為true

<!--
* 確認模式：
* 步驟：
* 1. 確認模式開啟：ConnectionFactory中開啟publisher-confirms="true"
-->
<!-- 定義rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"
                               publisher-confirms="true"/>

/*
 * 確認模式：
 * 步驟：
 * 2. 在rabbitTemplate定義ConfirmCallBack回撥函數
 */
@Test
    public void queueTest(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            /**
             *
             * @param correlationData 相關設定資訊
             * @param ack exchange交換機 是否成功收到了訊息。true 成功，false代表失敗
             * @param cause 失敗原因
             */
                System.out.println("confirm方法被執行了....");
                if (ack) {
                    //接收成功
                    System.out.println("接收成功訊息" + cause);
                } else {
                    //接收失敗
                    System.out.println("接收失敗訊息" + cause);
                    //做一些處理，讓訊息再次傳送。
                }
            }
        });
        //路由鍵與佇列同名
        rabbitTemplate.convertAndSend("spring_queue", "message confirm....");
    }

因為正常向佇列中傳送了訊息，所以返回的cause值為空，如果出現異常，cause為異常原因

退回模式

訊息從 exchange–>queue 投遞失敗則會返回一個 returnCallback

1.開啟回退模式:publisher-returns=“true”

    <!-- 定義rabbitmq connectionFactory -->
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
                               port="${rabbitmq.port}"
                               username="${rabbitmq.username}"
                               password="${rabbitmq.password}"
                               virtual-host="${rabbitmq.virtual-host}"
                               publisher-returns="true"/>

2.設定Exchange處理訊息失敗的模式：setMandatory，然後設定ReturnCallBack

    @Test
    public void queueTest(){
        //1.設定交換機處理失敗訊息的模式
        rabbitTemplate.setMandatory(true);
        //2.設定ReturnCallBack
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * @param message    訊息物件
             * @param replyCode  錯誤碼
             * @param replyText  錯誤資訊
             * @param exchange   交換機
             * @param routingKey 路由鍵
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String
                    replyText, String exchange, String routingKey) {
                System.out.println("return 執行了....");
                System.out.println(message);
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);
                //處理
            }
        });
        //手動新增錯誤路由模擬錯誤發生
        rabbitTemplate.convertAndSend("spring_topic_exchange", "return123", "return message...");
    }

此處只有發生錯誤才會返回訊息，因此手動加上一個錯誤，給傳送訊息新增路由值return123，實際上並沒有這個路由，執行返回訊息如下。

Consumer Ack

ack指Acknowledge，確認。表示消費端收到訊息後的確認方式。

有三種確認方式：

自動確認:acknowledge=“none”
手動確認:acknowledge=“manual”
根據異常情況確認:acknowledge=“auto”，（這種方式使用麻煩，沒有進行學習）

其中自動確認是指，當訊息一旦被Consumer接收到，則自動確認收到，並將相應 message 從RabbitMQ 的訊息快取中移除。但是在實際業務處理中，很可能訊息接收到，業務處理出現異常，那麼該訊息就會丟失。如果設定了手動確認方式，則需要在業務處理成功後，呼叫channel.basicAck()，手動簽收，如果出現異常，則呼叫channel.basicNack()方法，讓其自動重新傳送訊息。

還是以spring整合rabbitmq為例，rabbitmq組態檔中設定確認方式

<rabbit:listener-container connection-factory="connectionFactory"
acknowledge="manual">
.....

監聽類程式碼如下：

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收轉換訊息
            System.out.println(new String(message.getBody()));
            //2. 處理業務邏輯
            System.out.println("處理業務邏輯...");
            int i = 3/0;//出現錯誤
            // 3. 手動簽收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒絕簽收
            /*
             *第三個引數：requeue：重回佇列。如果設定為true，則訊息重新回到queue，broker會
             *重新傳送該訊息給消費端
             */
            channel.basicNack(deliveryTag,true,true);
            //channel.basicReject(deliveryTag,true);
        }
    }
}

因為出現異常呼叫channel.basicNack()方法，讓其自動重新傳送訊息，所以無限迴圈輸出內容

消費端限流

當我們的 Rabbitmq 伺服器積壓了有上萬條未處理的訊息時，我們隨便開啟一個消費者使用者端，會出現這樣情況: 巨量的訊息瞬間全部推播過來，但是我們單個使用者端無法同時處理這麼多資料!當資料量特別大的時候，我們對生產端限流肯定是不科學的，因為有時候並行量就是特別大，有時候並行量又特別少，我們無法約束生產端，這是使用者的行為。所以我們應該對消費端限流，rabbitmq提供了一種qos（服務質量保證）功能，即在非自動確認訊息的前提下，如果一定數目的訊息（給channel或者consume設定Qos值）未被確認前，不進行消費新訊息。

1.確保ack機制為手動確認

2.listener-container設定屬性perfetch = 1,表示消費端每次從mq拉去一條訊息來消費，直到手動確認消費完畢後，才會繼續拉去下一條訊息。

<rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual" prefetch="1">
        <rabbit:listener ref="topicListenerACK" queue-names="spring_topic_queue_well2"/>
</rabbit:listener-container>

生產者，傳送五條訊息

    @Test
    public void topicTest(){
/**
 * 引數1：交換機名稱
 * 引數2：路由鍵名
 * 引數3：傳送的訊息內容
 */
        for (int i=0;i<5;i++){
            rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.a", "傳送到spring_topic_exchange交換機xzk.cn的訊息"+i);
        }
    }
}

生產者註釋掉channel.basicAck(deliveryTag,true)即不確認收到訊息

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收轉換訊息
            System.out.println(new String(message.getBody()));
            //2. 處理業務邏輯
            System.out.println("處理業務邏輯...");
            // 3. 手動簽收
            //channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒絕簽收
            /*
             *第三個引數：requeue：重回佇列。如果設定為true，則訊息重新回到queue，broker會
             *重新傳送該訊息給消費端
             */
            channel.basicNack(deliveryTag,true,true);
        }
    }
}

此時啟動消費者再執行生產者之後，發現消費者傳送了五條訊息，實際上生產者只接受到了一條訊息，達到限流作用

觀察rabbitmq控制檯，發現有1條unack訊息。4條ready訊息，還沒到達consumer。和我們設定的prefetchCount=1限流情況相符。

把channel.basicAck(deliveryTag,true)的註釋取消掉，即可以自動確認收到訊息，重新執行消費者，接收到了另外的四條訊息

TTL(Time To Live)

Time To Live，訊息過期時間設定

設定某個佇列為過期佇列

設定交換機，佇列以及佇列過期時間為10000ms

 <!--ttl-->
    <rabbit:queue name="test_queue_ttl" id="test_queue_ttl">
        <rabbit:queue-arguments>
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        </rabbit:queue-arguments>
    </rabbit:queue>
    <rabbit:topic-exchange name="test_exchange_ttl">
        <rabbit:bindings>
            <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>
        </rabbit:bindings>
    </rabbit:topic-exchange>

生產者傳送10條訊息

    @Test
    public void testTtl() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...");
        }

十秒鐘後，過期訊息消失

設定單獨某個訊息過期

設定交換機和佇列

<rabbit:queue name="test_queue_ttl" id="test_queue_ttl"/>
<rabbit:topic-exchange name="test_exchange_ttl">
    <rabbit:bindings>
        <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/>     
    </rabbit:bindings>
</rabbit:topic-exchange>

生產者傳送特定過期訊息，用到了MessagePostProcessor這個api

 @Test
    public void testTtl() {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //1.設定message資訊
                message.getMessageProperties().setExpiration("5000");//訊息的過期時間
                //2.返回該訊息
                return message;
            }
        };
        //訊息單獨過期
        rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...",messagePostProcessor);
    }

5s之後

注:

1.如果同時設定佇列過期和訊息過期，系統會根據哪個過期的時間短而選用哪兒個。

2.設定單獨訊息過期時，如果該訊息不為第一個接受的訊息，則不過期。

死信佇列

死信佇列，英文縮寫：DLX 。Dead Letter Exchange（死信交換機），當訊息成為Deadmessage後，可以被重新傳送到另一個交換機，這個交換機就是DLX。

訊息成為死信的三種情況：

佇列訊息長度到達限制；
消費者拒接消費訊息，basicNack/basicReject,並且不把訊息重新放入原目標佇列,requeue=false；
原佇列存在訊息過期設定，訊息到達超時時間未被消費；

佇列繫結死信交換機：

給佇列設定引數： x-dead-letter-exchange 和 x-dead-letter-routing-key

實現

1.宣告正常的佇列(test_queue_dlx)和交換機(test_exchange_dlx)

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
    <!--正常佇列繫結死信交換機-->
    <rabbit:queue-arguments>
        <!--x-dead-letter-exchange：死信交換機名稱-->
        <entry key="x-dead-letter-exchange" value="exchange_dlx" />
        <!--3.2 x-dead-letter-routing-key：傳送給死信交換機的routingkey-->
        <entry key="x-dead-letter-routing-key" value="dlx.hehe" />
        <!--4.1 設定佇列的過期時間 ttl-->
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        <!--4.2 設定佇列的長度限制 max-length -->
        <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="test_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx">
        </rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

2.宣告死信佇列(queue_dlx)和死信交換機(exchange_dlx)

<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

3.生產端測試

/**
* 傳送測試死信訊息：
* 1. 過期時間
* 2. 長度限制
* 3. 訊息拒收
*/
@Test
public void testDlx(){
    //1. 測試過期時間，死信訊息
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條訊息，我會死嗎？");
    //2. 測試長度限制後，訊息死信
    /* for (int i = 0; i < 20; i++) {
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條訊息，我會死嗎？");
    }*/
    //3. 測試訊息拒收
    //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條訊息，我會死嗎？");
}

4.消費端監聽

public class DlxListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收轉換訊息
            System.out.println(new String(message.getBody()));
            //2. 處理業務邏輯
            System.out.println("處理業務邏輯...");
            int i = 3/0;//出現錯誤
            //3. 手動簽收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            System.out.println("出現異常，拒絕接受");
            //4.拒絕簽收，不重回佇列 requeue=false
            channel.basicNack(deliveryTag,true,false);
        }
    }
}

<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx">
</rabbit:listener>

延遲佇列

延遲佇列，即訊息進入佇列後不會立即被消費，只有到達指定時間後，才會被消費。c

需求：

1.下單後，30分鐘未支付，取消訂單，回滾庫存。

2.新使用者註冊成功7天后，傳送簡訊問候。

實現方式：

定時器
延遲佇列

定時器的實現方式不夠優雅，我們採取延遲佇列的方式

不過很可惜，在RabbitMQ中並未提供延遲佇列功能。

但是可以使用：TTL+死信佇列組合實現延遲佇列的效果。

設定

<!--
延遲佇列：
        1. 定義正常交換機（order_exchange）和佇列(order_queue)
        2. 定義死信交換機（order_exchange_dlx）和佇列(order_queue_dlx)
        3. 繫結，設定正常佇列過期時間為30分鐘
-->
<!-- 定義正常交換機（order_exchange）和佇列(order_queue)-->
<rabbit:queue id="order_queue" name="order_queue">
<!-- 繫結，設定正常佇列過期時間為30分鐘-->
    <rabbit:queue-arguments>
        <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
        <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
    </rabbit:queue-arguments>
</rabbit:queue>
<rabbit:topic-exchange name="order_exchange">
    <rabbit:bindings>
        <rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>
<!-- 定義死信交換機（order_exchange_dlx）和佇列(order_queue_dlx)-->
<rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue>
<rabbit:topic-exchange name="order_exchange_dlx">
    <rabbit:bindings>
        <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding>
    </rabbit:bindings>
</rabbit:topic-exchange>

生產端測試

@Test
public void testDelay() throws InterruptedException {
    //1.傳送訂單訊息。 將來是在訂單系統中，下單成功後，傳送訊息
    rabbitTemplate.convertAndSend("order_exchange","order.msg","訂單資訊：id=1,time=2019年8月17日16:41:47");
    /*//2.列印倒計時10秒
    for (int i = 10; i > 0 ; i--) {
        System.out.println(i+"...");
        Thread.sleep(1000);
    }*/
}

消費端監聽

public class OrderListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
		long deliveryTag = message.getMessageProperties().getDeliveryTag();
		try {
			//1.接收轉換訊息
			System.out.println(new String(message.getBody()));
			//2. 處理業務邏輯
			System.out.println("處理業務邏輯...");
			System.out.println("根據訂單id查詢其狀態...");
			System.out.println("判斷狀態是否為支付成功");
			System.out.println("取消訂單，回滾庫存....");
			//3. 手動簽收
			channel.basicAck(deliveryTag,true);
		} catch (Exception e) {
			//e.printStackTrace();
			System.out.println("出現異常，拒絕接受");
			//4.拒絕簽收，不重回佇列 requeue=false
			channel.basicNack(deliveryTag,true,false);
		}
	}
}

<rabbit:listener ref="orderListener" queue-names="order_queue_dlx">
</rabbit:listener>

到此這篇關於Java RabbitMQ高階特性詳細分析的文章就介紹到這了,更多相關Java RabbitMQ特性內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！