Spring Boot系列(8)——RabbitMQ確認、退回模式及死信隊列

〇、什么是消息隊列

　　參考：新手也能看懂，消息隊列其實很簡單

 　　　　   RabbitMQ運行模型與名詞解釋

 

一、應答模式

　　1.什么是應答？

　　　　消息投遞到交換器（exchange）中，交換器給我們的反饋，是保障消息投遞成功的一種機制。

　　2.測試

　　配置：

#選擇確認類型為交互

 2 spring.rabbitmq.publisher-confirm-type=correlated

 

　　測試方法：

    @Test
    /**
     * the test is testing confirm-function in rabbitmq
     */
    void messageSendTestWithConfirm(){

        /*
         * 設置消息確認回調方法；
         * @ack 為true時，表示投遞成功；為false表示投遞失敗；
         * @CorrelationData 為自定義反饋信息；
         * @cause 為投遞失敗的原因；
         */
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if(!ack){
                    logger.error("correlationData："+ correlationData);
                    logger.error("ack："+ack);
                    logger.error("cause："+cause);
                }
            }
        });

        //消息內容
        Map<String,String> map = new HashMap<>();
        map.put("message","testing confire function");

        //設置自定義反饋消息
        String uuid = UUID.randomUUID().toString();
        logger.info("消息唯一ID："+uuid);
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(uuid);

        //並不存在名為“exchange-dog”的exchange
        rabbitTemplate.convertAndSend("exchange-dog","dog",map,correlationData);

    }

 

　　測試結果

 c.d.amqp.SpringBootAmqpApplicationTests  : 消息唯一ID：e6601e83-fad7-4b53-9968-c74828e62b23 2  o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [192.168.22.130:5672]
 o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#f7cdf8:0/SimpleConnection@397ef2 [delegate=amqp://guest@192.168.22.130:5672/, localPort= 8055]
 o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'exchange-dog' in vhost '/', class-id=60, method-id=40)
 c.d.amqp.SpringBootAmqpApplicationTests  : correlationData：CorrelationData [id=e6601e83-fad7-4b53-9968-c74828e62b23]
 c.d.amqp.SpringBootAmqpApplicationTests  : ack：false
 c.d.amqp.SpringBootAmqpApplicationTests  : cause：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'exchange-dog' in vhost '/', class-id=60, method-id=40)

 

　　注意：Confirm模式只管有無投遞到exchange，而不管有無發送到隊列當中。

 

二、返回模式

　　1.什么是返回模式？

　　　　當消息未投遞到queue時的反饋。

 

　　2.測試

　　配置：

#開啟返回模式

spring.rabbitmq.publisher-returns=true

 

　　測試方法：

    @Test
    void messageSendTestWithReturn(){
        /*
         * 設置消息返回回調方法；
         * 該方法執行時則表示消息投遞失敗
         * @message 為反饋信息；
         * @replyCode 一個反饋代碼，表示不同投遞失敗原因；
         * @replyText 反饋信息
         */
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                logger.error("返回消息配置："+message.getMessageProperties().toString());
                logger.error("反饋代碼："+replyCode);
                logger.error("反饋內容："+replyText);
                logger.error("exchange："+exchange);
                logger.error("routingKey："+routingKey);
            }
        });

        //消息內容
        Map<String,String> map = new HashMap<>();
        map.put("message","testing return function");

        //並不存在名為“dog”的routingKey，即投不到現有的queue里
        rabbitTemplate.convertAndSend("exchange-direct","dog",map);
    }

 

　　測試結果：

c.d.amqp.SpringBootAmqpApplicationTests  : 返回消息配置：MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]
c.d.amqp.SpringBootAmqpApplicationTests  : 反饋代碼：312 3 c.d.amqp.SpringBootAmqpApplicationTests  : 反饋內容：NO_ROUTE 4 c.d.amqp.SpringBootAmqpApplicationTests  : exchange：exchange-direct
c.d.amqp.SpringBootAmqpApplicationTests  : routingKey：dog

 

三、限流策略（手動應答消息）

　　1.為什么要限流？

　　　　若隊列中消息積壓過多，突然開啟監聽，會導致消費端崩潰。

 

　　2.如何限流？

　　　　使用RabbitMQ提供的Qos(服務質量保證)功能，如果一定數目消息的未被應答前，不再接受新消息。

 

　　3.測試

　　配置

1 #手動消息應答

2 spring.rabbitmq.listener.simple.acknowledge-mode=manual

　　

　　測試

    /*
     * 消息手動應答
     * @RabbitListener注解監聽來自指定隊列的消息
     */

    @RabbitListener(queues = "springboot-queue")
    public void revice(Message message,Channel channel) throws IOException {
        try{
            logger.info("消息ID："+message.getMessageProperties().getHeader("spring_returned_message_correlation"));
            logger.info("消息標簽："+String.valueOf(message.getMessageProperties().getDeliveryTag()));
            /* 設置Qos機制
             * 第一個參數：單條消息的大小(0表示即無限制)
             * 第二個參數：每次處理消息的數量
             * 第三個參數：是否為consumer級別（false表示僅當前channel有效）
             */
            channel.basicQos(0,1,false);
            //手動應答消息　　第一個參數是所確認消息的標識，第二參數是是否批量確認
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            
        }catch (Exception e){
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            logger.error("消息ID："+message.getMessageProperties().getHeader("spring_returned_message_correlation"));
            logger.error("接收消息發送錯誤："+e.getMessage());
        }
    }

 

四、死信隊列

　　1.什么是死信？

　　　　未被正常處理的消息。

 

　　2.出現死信的情況

　　　　(1)消息被拒絕（reject、nack），並且不重新放回原隊列（requeue=false）；

　　　　(2)消息過期（設置了Expiration）；

　　　　(3)隊列已滿。

 

　　3.配置

　　以下代碼配置了一個正常的exchange、queue和 一個專門接收死信的exchange、queue

　　在隊列中配置x-dead-letter-exchange參數，表示本隊列出現死信，則轉發到配置指定的交換器中去

    /*
     * 創建消息隊列的對象：exchange、queue、綁定規則
     */
    @Test
    void createObjectOfMQ(){

        /*
         * 在普通隊列上配置參數。表示若本隊列有死信，則轉發到配置指定的轉發器中去
         * @參數鍵：x-dead-letter-exchange
         * @參數名：接收死信的交換器
         */
        Map<String,Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange","springboot-dlx-exchange"); 14 
        amqpAdmin.declareExchange(new DirectExchange("springboot-direct"));
        amqpAdmin.declareQueue(new Queue("springboot.queue",true,false,false,arguments));
        amqpAdmin.declareBinding(new Binding("springboot.queue", Binding.DestinationType.QUEUE,"springboot-direct","springboot",null));

        //接收死信的交換器
        amqpAdmin.declareExchange(new TopicExchange("springboot-dlx-exchange"));
        //交換器收到的死信都轉發到該隊列，#表示接收所有消息
        amqpAdmin.declareQueue(new Queue("springboot-dlx.queue",true));
        amqpAdmin.declareBinding(new Binding("springboot-dlx.queue", Binding.DestinationType.QUEUE,"springboot-dlx-exchange","#",null));
    }

 

　　運行以上方法后，可在RabbitMQ管理界面看到兩個exchange、兩個queue成功創建

 

　　4.發送消息

   @Test
    void sendMessageWithTTL(){
        String str = "dlx test";

        //setExpiration表示設置該消息存活時間
        //5秒后該消息未被消息，則轉發到死信交換器
        Message message = MessageBuilder.withBody(str.getBytes())
                .setExpiration("5000")  9                 .setContentEncoding("UTF-8")
                .setMessageId("dlx-001")
                .build();

        //將消息發送到配置了"x-dead-letter-exchange"參數的隊列
        rabbitTemplate.convertAndSend("springboot-direct","springboot",message);
    }

 

　　 發送完消息，過5秒之后可以看到信息已被投遞到死信隊列中去了

 

 

 

 

setExpiration