首頁  >  文章  >  Java  >  Java RabbitMQ高階特性實例分析

Java RabbitMQ高階特性實例分析

WBOY
WBOY轉載
2023-04-29 20:25:05846瀏覽

    訊息的可靠投遞

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

    • confirm 確認模式

    • return 退回模式

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

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

    • #訊息從producer到ex則會傳回一個confir#Callback

    • 訊息從producer到exchange

    • #訊息從exchange—>queue投遞失敗則會傳回一個returnCallback

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

    確認模式

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

    以spring整合rabbitmq為例,修改rabbitmq設定檔,在connectionFactory中加入publisher-confirms屬性並設定值為trueJava RabbitMQ高階特性實例分析

    <!--
    * 确认模式:
    * 步骤:
    * 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,實際上並沒有這個路由,運行返回訊息如下。 Java RabbitMQ高階特性實例分析

    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()方法,讓其自動重新發送訊息,所以無限循環輸出內容Java RabbitMQ高階特性實例分析

    消費端限流Java RabbitMQ高階特性實例分析

    當我們的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);
            }
        }
    }

    此時啟動消費者再運行生產者之後,發現消費者發送了五條訊息,實際上生產者只接受到了一條訊息,達到限流作用Java RabbitMQ高階特性實例分析

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

    Java RabbitMQ高階特性實例分析

    把channel.basicAck(deliveryTag,true)的注释取消掉,即可以自动确认收到消息,重新运行消费者,接收到了另外的四条消息

    Java RabbitMQ高階特性實例分析

    Java RabbitMQ高階特性實例分析

    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...");
            }

    Java RabbitMQ高階特性實例分析

    十秒钟后,过期消息消失

    Java RabbitMQ高階特性實例分析

    设置单独某个消息过期

    设置交换机和队列

    <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);
        }

    Java RabbitMQ高階特性實例分析

    5s之后

    Java RabbitMQ高階特性實例分析

    注:

    1.如果同时设置队列过期和消息过期,系统会根据哪个过期的时间短而选用哪儿个。

    2.设置单独消息过期时,如果该消息不为第一个接受的消息,则不过期。

    死信队列

    死信队列,英文缩写:DLX 。Dead Letter Exchange(死信交换机),当消息成为Deadmessage后,可以被重新发送到另一个交换机,这个交换机就是DLX。

    Java RabbitMQ高階特性實例分析

    消息成为死信的三种情况:

    • 队列消息长度到达限制;

    • 消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false;

    • 原队列存在消息过期设置,消息到达超时时间未被消费;

    队列绑定死信交换机:

    给队列设置参数: x-dead-letter-exchange 和 x-dead-letter-routing-key

    Java RabbitMQ高階特性實例分析

    实现

    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天后,发送短信问候。

    实现方式:

    • 定时器

    • 延迟队列

    定时器的实现方式不够优雅,我们采取延迟队列的方式

    Java RabbitMQ高階特性實例分析

    不过很可惜,在RabbitMQ中并未提供延迟队列功能。

    但是可以使用:TTL+死信队列 组合实现延迟队列的效果。

    Java RabbitMQ高階特性實例分析

    配置

    <!--
    延迟队列:
            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高階特性實例分析的詳細內容。更多資訊請關注PHP中文網其他相關文章!

    陳述:
    本文轉載於:yisu.com。如有侵權,請聯絡admin@php.cn刪除