Java RabbitMQ高級(jí)特性詳細(xì)分析

更新時(shí)間：2022年08月08日 11:30:10 作者：Cavewang

為了保證消息的可靠性傳輸，包括投遞消息的生產(chǎn)方能投遞成功，和消息消費(fèi)的消費(fèi)方正確消費(fèi)，RabbitMQ 提供了兩個(gè)確認(rèn)機(jī)制，由于消息按照流通的順序從左到右，因此為保證可靠性，MQ必須對(duì) Producer進(jìn)行確認(rèn)，Consumer 必須對(duì) MQ 進(jìn)行確認(rèn)

消息的可靠投遞

在使用 RabbitMQ 的時(shí)候，作為消息發(fā)送方希望杜絕任何消息丟失或者投遞失敗場(chǎng)景。RabbitMQ 為我們提供了兩種方式用來(lái)控制消息的投遞可靠性模式。

confirm 確認(rèn)模式
return 退回模式

rabbitmq整個(gè)消息投遞的路徑為：

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

消息從producer到exchange則會(huì)返回一個(gè)confirmCallback
消息從exchange—>queue投遞失敗則會(huì)返回一個(gè)returnCallback

我們可以利用這兩個(gè)callback控制消息的可靠性投遞

確認(rèn)模式

消息從 producer 到 exchange 則會(huì)返回一個(gè) confirmCallback

以spring整合rabbitmq為例，修改rabbitmq配置文件,在connectionFactory中添加publisher-confirms屬性并設(shè)置值為true

<!--
* 確認(rèn)模式：
* 步驟：
* 1. 確認(rèn)模式開(kāi)啟：ConnectionFactory中開(kāi)啟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"/>

/*
 * 確認(rèn)模式：
 * 步驟：
 * 2. 在rabbitTemplate定義ConfirmCallBack回調(diào)函數(shù)
 */
@Test
    public void queueTest(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            /**
             *
             * @param correlationData 相關(guān)配置信息
             * @param ack exchange交換機(jī) 是否成功收到了消息。true 成功，false代表失敗
             * @param cause 失敗原因
             */
                System.out.println("confirm方法被執(zhí)行了....");
                if (ack) {
                    //接收成功
                    System.out.println("接收成功消息" + cause);
                } else {
                    //接收失敗
                    System.out.println("接收失敗消息" + cause);
                    //做一些處理，讓消息再次發(fā)送。
                }
            }
        });
        //路由鍵與隊(duì)列同名
        rabbitTemplate.convertAndSend("spring_queue", "message confirm....");
    }

因?yàn)檎Ｏ蜿?duì)列中發(fā)送了消息，所以返回的cause值為空，如果出現(xiàn)異常，cause為異常原因

退回模式

消息從 exchange–>queue 投遞失敗則會(huì)返回一個(gè) returnCallback

1.開(kāi)啟回退模式: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.設(shè)置Exchange處理消息失敗的模式：setMandatory，然后設(shè)置ReturnCallBack

    @Test
    public void queueTest(){
        //1.設(shè)置交換機(jī)處理失敗消息的模式
        rabbitTemplate.setMandatory(true);
        //2.設(shè)置ReturnCallBack
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * @param message    消息對(duì)象
             * @param replyCode  錯(cuò)誤碼
             * @param replyText  錯(cuò)誤信息
             * @param exchange   交換機(jī)
             * @param routingKey 路由鍵
             */
            @Override
            public void returnedMessage(Message message, int replyCode, String
                    replyText, String exchange, String routingKey) {
                System.out.println("return 執(zhí)行了....");
                System.out.println(message);
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);
                //處理
            }
        });
        //手動(dòng)添加錯(cuò)誤路由模擬錯(cuò)誤發(fā)生
        rabbitTemplate.convertAndSend("spring_topic_exchange", "return123", "return message...");
    }

此處只有發(fā)生錯(cuò)誤才會(huì)返回消息，因此手動(dòng)加上一個(gè)錯(cuò)誤，給發(fā)送消息添加路由值return123，實(shí)際上并沒(méi)有這個(gè)路由，運(yùn)行返回消息如下。

Consumer Ack

ack指Acknowledge，確認(rèn)。表示消費(fèi)端收到消息后的確認(rèn)方式。

有三種確認(rèn)方式：

自動(dòng)確認(rèn):acknowledge=“none”
手動(dòng)確認(rèn):acknowledge=“manual”
根據(jù)異常情況確認(rèn):acknowledge=“auto”，（這種方式使用麻煩，沒(méi)有進(jìn)行學(xué)習(xí)）

其中自動(dòng)確認(rèn)是指，當(dāng)消息一旦被Consumer接收到，則自動(dòng)確認(rèn)收到，并將相應(yīng) message 從RabbitMQ 的消息緩存中移除。但是在實(shí)際業(yè)務(wù)處理中，很可能消息接收到，業(yè)務(wù)處理出現(xiàn)異常，那么該消息就會(huì)丟失。如果設(shè)置了手動(dòng)確認(rèn)方式，則需要在業(yè)務(wù)處理成功后，調(diào)用channel.basicAck()，手動(dòng)簽收，如果出現(xiàn)異常，則調(diào)用channel.basicNack()方法，讓其自動(dòng)重新發(fā)送消息。

還是以spring整合rabbitmq為例，rabbitmq配置文件中設(shè)置確認(rèn)方式

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

監(jiān)聽(tīng)類(lèi)代碼如下：

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收轉(zhuǎn)換消息
            System.out.println(new String(message.getBody()));
            //2. 處理業(yè)務(wù)邏輯
            System.out.println("處理業(yè)務(wù)邏輯...");
            int i = 3/0;//出現(xiàn)錯(cuò)誤
            // 3. 手動(dòng)簽收
            channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒絕簽收
            /*
             *第三個(gè)參數(shù)：requeue：重回隊(duì)列。如果設(shè)置為true，則消息重新回到queue，broker會(huì)
             *重新發(fā)送該消息給消費(fèi)端
             */
            channel.basicNack(deliveryTag,true,true);
            //channel.basicReject(deliveryTag,true);
        }
    }
}

因?yàn)槌霈F(xiàn)異常調(diào)用channel.basicNack()方法，讓其自動(dòng)重新發(fā)送消息，所以無(wú)限循環(huán)輸出內(nèi)容

消費(fèi)端限流

當(dāng)我們的 Rabbitmq 服務(wù)器積壓了有上萬(wàn)條未處理的消息時(shí)，我們隨便打開(kāi)一個(gè)消費(fèi)者客戶(hù)端，會(huì)出現(xiàn)這樣情況: 巨量的消息瞬間全部推送過(guò)來(lái)，但是我們單個(gè)客戶(hù)端無(wú)法同時(shí)處理這么多數(shù)據(jù)!當(dāng)數(shù)據(jù)量特別大的時(shí)候，我們對(duì)生產(chǎn)端限流肯定是不科學(xué)的，因?yàn)橛袝r(shí)候并發(fā)量就是特別大，有時(shí)候并發(fā)量又特別少，我們無(wú)法約束生產(chǎn)端，這是用戶(hù)的行為。所以我們應(yīng)該對(duì)消費(fèi)端限流，rabbitmq提供了一種qos（服務(wù)質(zhì)量保證）功能，即在非自動(dòng)確認(rèn)消息的前提下，如果一定數(shù)目的消息（給channel或者consume設(shè)置Qos值）未被確認(rèn)前，不進(jìn)行消費(fèi)新消息。

1.確保ack機(jī)制為手動(dòng)確認(rèn)

2.listener-container配置屬性perfetch = 1,表示消費(fèi)端每次從mq拉去一條消息來(lái)消費(fèi)，直到手動(dòng)確認(rèn)消費(fèi)完畢后，才會(huì)繼續(xù)拉去下一條消息。

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

生產(chǎn)者，發(fā)送五條消息

    @Test
    public void topicTest(){
/**
 * 參數(shù)1：交換機(jī)名稱(chēng)
 * 參數(shù)2：路由鍵名
 * 參數(shù)3：發(fā)送的消息內(nèi)容
 */
        for (int i=0;i<5;i++){
            rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.a", "發(fā)送到spring_topic_exchange交換機(jī)xzk.cn的消息"+i);
        }
    }
}

生產(chǎn)者注釋掉channel.basicAck(deliveryTag,true)即不確認(rèn)收到消息

public class AckListener implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //1.接收轉(zhuǎn)換消息
            System.out.println(new String(message.getBody()));
            //2. 處理業(yè)務(wù)邏輯
            System.out.println("處理業(yè)務(wù)邏輯...");
            // 3. 手動(dòng)簽收
            //channel.basicAck(deliveryTag,true);
        } catch (Exception e) {
            //e.printStackTrace();
            //4.拒絕簽收
            /*
             *第三個(gè)參數(shù)：requeue：重回隊(duì)列。如果設(shè)置為true，則消息重新回到queue，broker會(huì)
             *重新發(fā)送該消息給消費(fèi)端
             */
            channel.basicNack(deliveryTag,true,true);
        }
    }
}

此時(shí)啟動(dòng)消費(fèi)者再運(yùn)行生產(chǎn)者之后，發(fā)現(xiàn)消費(fèi)者發(fā)送了五條消息，實(shí)際上生產(chǎn)者只接受到了一條消息，達(dá)到限流作用

觀察rabbitmq控制臺(tái)，發(fā)現(xiàn)有1條unack消息。4條ready消息，還沒(méi)到達(dá)consumer。和我們?cè)O(shè)置的prefetchCount=1限流情況相符。

把channel.basicAck(deliveryTag,true)的注釋取消掉，即可以自動(dòng)確認(rèn)收到消息，重新運(yùn)行消費(fèi)者，接收到了另外的四條消息

TTL(Time To Live)

Time To Live，消息過(guò)期時(shí)間設(shè)置

設(shè)置某個(gè)隊(duì)列為過(guò)期隊(duì)列

設(shè)置交換機(jī)，隊(duì)列以及隊(duì)列過(guò)期時(shí)間為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>

生產(chǎn)者發(fā)送10條消息

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

十秒鐘后，過(guò)期消息消失

設(shè)置單獨(dú)某個(gè)消息過(guò)期

設(shè)置交換機(jī)和隊(duì)列

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

生產(chǎn)者發(fā)送特定過(guò)期消息，用到了MessagePostProcessor這個(gè)api

 @Test
    public void testTtl() {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //1.設(shè)置message信息
                message.getMessageProperties().setExpiration("5000");//消息的過(guò)期時(shí)間
                //2.返回該消息
                return message;
            }
        };
        //消息單獨(dú)過(guò)期
        rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...",messagePostProcessor);
    }

5s之后

注:

1.如果同時(shí)設(shè)置隊(duì)列過(guò)期和消息過(guò)期，系統(tǒng)會(huì)根據(jù)哪個(gè)過(guò)期的時(shí)間短而選用哪兒個(gè)。

2.設(shè)置單獨(dú)消息過(guò)期時(shí)，如果該消息不為第一個(gè)接受的消息，則不過(guò)期。

死信隊(duì)列

死信隊(duì)列，英文縮寫(xiě)：DLX 。Dead Letter Exchange（死信交換機(jī)），當(dāng)消息成為Deadmessage后，可以被重新發(fā)送到另一個(gè)交換機(jī)，這個(gè)交換機(jī)就是DLX。

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

隊(duì)列消息長(zhǎng)度到達(dá)限制；
消費(fèi)者拒接消費(fèi)消息，basicNack/basicReject,并且不把消息重新放入原目標(biāo)隊(duì)列,requeue=false；
原隊(duì)列存在消息過(guò)期設(shè)置，消息到達(dá)超時(shí)時(shí)間未被消費(fèi)；

隊(duì)列綁定死信交換機(jī)：

給隊(duì)列設(shè)置參數(shù)： x-dead-letter-exchange 和 x-dead-letter-routing-key

實(shí)現(xiàn)

1.聲明正常的隊(duì)列(test_queue_dlx)和交換機(jī)(test_exchange_dlx)

<rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
    <!--正常隊(duì)列綁定死信交換機(jī)-->
    <rabbit:queue-arguments>
        <!--x-dead-letter-exchange：死信交換機(jī)名稱(chēng)-->
        <entry key="x-dead-letter-exchange" value="exchange_dlx" />
        <!--3.2 x-dead-letter-routing-key：發(fā)送給死信交換機(jī)的routingkey-->
        <entry key="x-dead-letter-routing-key" value="dlx.hehe" />
        <!--4.1 設(shè)置隊(duì)列的過(guò)期時(shí)間 ttl-->
        <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/>
        <!--4.2 設(shè)置隊(duì)列的長(zhǎng)度限制 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.聲明死信隊(duì)列(queue_dlx)和死信交換機(jī)(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.生產(chǎn)端測(cè)試

/**
* 發(fā)送測(cè)試死信消息：
* 1. 過(guò)期時(shí)間
* 2. 長(zhǎng)度限制
* 3. 消息拒收
*/
@Test
public void testDlx(){
    //1. 測(cè)試過(guò)期時(shí)間，死信消息
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條消息，我會(huì)死嗎？");
    //2. 測(cè)試長(zhǎng)度限制后，消息死信
    /* for (int i = 0; i < 20; i++) {
    rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條消息，我會(huì)死嗎？");
    }*/
    //3. 測(cè)試消息拒收
    //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一條消息，我會(huì)死嗎？");
}

4.消費(fèi)端監(jiān)聽(tīng)

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

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

延遲隊(duì)列

延遲隊(duì)列，即消息進(jìn)入隊(duì)列后不會(huì)立即被消費(fèi)，只有到達(dá)指定時(shí)間后，才會(huì)被消費(fèi)。c

需求：

1.下單后，30分鐘未支付，取消訂單，回滾庫(kù)存。

2.新用戶(hù)注冊(cè)成功7天后，發(fā)送短信問(wèn)候。

實(shí)現(xiàn)方式：

定時(shí)器
延遲隊(duì)列

定時(shí)器的實(shí)現(xiàn)方式不夠優(yōu)雅，我們采取延遲隊(duì)列的方式

不過(guò)很可惜，在RabbitMQ中并未提供延遲隊(duì)列功能。

但是可以使用：TTL+死信隊(duì)列組合實(shí)現(xiàn)延遲隊(duì)列的效果。

配置

<!--
延遲隊(duì)列：
        1. 定義正常交換機(jī)（order_exchange）和隊(duì)列(order_queue)
        2. 定義死信交換機(jī)（order_exchange_dlx）和隊(duì)列(order_queue_dlx)
        3. 綁定，設(shè)置正常隊(duì)列過(guò)期時(shí)間為30分鐘
-->
<!-- 定義正常交換機(jī)（order_exchange）和隊(duì)列(order_queue)-->
<rabbit:queue id="order_queue" name="order_queue">
<!-- 綁定，設(shè)置正常隊(duì)列過(guò)期時(shí)間為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>
<!-- 定義死信交換機(jī)（order_exchange_dlx）和隊(duì)列(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>

生產(chǎn)端測(cè)試

@Test
public void testDelay() throws InterruptedException {
    //1.發(fā)送訂單消息。 將來(lái)是在訂單系統(tǒng)中，下單成功后，發(fā)送消息
    rabbitTemplate.convertAndSend("order_exchange","order.msg","訂單信息：id=1,time=2019年8月17日16:41:47");
    /*//2.打印倒計(jì)時(shí)10秒
    for (int i = 10; i > 0 ; i--) {
        System.out.println(i+"...");
        Thread.sleep(1000);
    }*/
}

消費(fèi)端監(jiān)聽(tīng)

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

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

到此這篇關(guān)于Java RabbitMQ高級(jí)特性詳細(xì)分析的文章就介紹到這了,更多相關(guān)Java RabbitMQ特性?xún)?nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Java RabbitMQ高級(jí)特性詳細(xì)分析

目錄

消息的可靠投遞

確認(rèn)模式

退回模式

Consumer Ack

消費(fèi)端限流

TTL(Time To Live)

設(shè)置某個(gè)隊(duì)列為過(guò)期隊(duì)列

設(shè)置單獨(dú)某個(gè)消息過(guò)期

死信隊(duì)列

延遲隊(duì)列

相關(guān)文章

最新評(píng)論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具