RabbitMQ高级特性:死信队列

科技新语

发布于 2023-04-07 09:03:52

2.8K00

代码可运行

文章被收录于专栏：用户9378866的专栏用户9378866的专栏

运行总次数：0

代码可运行

什么是死信队列?

死信队列,英文缩写DLX,Dead Letter Exchange(死信交换机),当消息成为Dead message(消息过期)后,可以被重新发送到另一个交换机,这个交换机就算是DLX,其实死信交换机(队列)和正常交换机(队列)没有什么区别

为什么叫死信队列但是翻译过来叫死信交换机呢,因为RabbitMQ比较特殊,其他MQ只有队列没有交换机这个概念的

正常来说,队列设置了过期时间,当消息到了队列之后,在过期时间内没有被消费,那么这个消息就被丢弃了,但是如果这个队列绑定了一个DLX死信队列(交换机),那么就算消息过期了也不会被直接丢弃掉,而是会发送给死信交换机,那么死信交换机又可以绑定其他队列,将这些消息存储到其他队列,从而又可以进行消息消费,就算这个意思,过程如图所示

什么情况下消息成为死信队列

消息成为死信队列的三种情况

1 队列消息长度达到限制

比如说给队列最大存储长度为10,当11条消息进来的时候,第11条消息进不去了，那么第11条消息就是死信

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

消费者使用basicNack/basicReject,并且requeue=false,表示消费者拒绝重新消费该消息

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

原来的队列存在过期时间,但是到了过期时间还没有消费该消息

队列绑定交换机的方式

给队列设置两个参数 x-dead-letter-exchange(设置交换机的名称)和x-dead-letter-routing-key(发送消息时指定的routingkey)

死信队列代码实现

声明两套交换机队列,一套是正常的交换机队列,一套是死信的交换机队列

1 声明正常队列和交换机test_queue_dlx test_exchange_dlx

2 声明死信队列和死信交换机 queue_dlx exchange_dlx

3 正常队列绑定死信交换机

设置两个参数

1死信交换机名称 x-dead-letter-exchange

2发送给死信交换机的路由键 x-dead-letter-routing-key

1 消息过期时间的测试

测试过期时间的死信给正常的交换机发送消息,过了存活时间消息自动从正常队列跑到死信队列

2 队列长度限制的测试

如果发送成功那么因为设置了最大长度是10,只会有10条进行正常队列剩下的会跑到死信队列,过了10s后正常队列中的消息也会自动跑到死信队列中

3 消费者消息拒收的限制

消费者Consumer监听正常的队列,然后让消息拒绝接收并且不重回队列

由于消费者拒收消息,消息会直接跑到私信队列中

最终相关测试代码

生产者配置

 <? xml version="1.0" encoding="UTF-8" ?>

<beans xmlns="http://www.springframework.org/schema/beans"

 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

 xmlns:context="http://www.springframework.org/schema/context"

 xmlns:rabbit="http://www.springframework.org/schema/rabbit"

 xsi:schemaLocation="http://www.springframework.org/schema/beans

 http://www.springframework.org/schema/beans/spring-beans.xsd

 http://www.springframework.org/schema/context

 https://www.springframework.org/schema/context/spring-context.xsd

 http://www.springframework.org/schema/rabbit

 http://www.springframework.org/schema/rabbit/spring-rabbit.xsd" >

    <!--加载配置文件-->

 <context:property-placeholder location="classpath:application.properties" />



    <!-- 定义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"

 publisher-returns="true"



 />



    <!--定义管理交换机,队列-->

 <rabbit:admin connection-factory="connectionFactory" />



    <!--定义rabbitTemplate对象操作可以在代码中调用api发送消息-->

 <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" />



    <!--消息的可靠性投递(生产端)-->

 <!--队列-->

 <rabbit:queue id="test_queue_confirm" name="test_queue_confirm" ></rabbit:queue>

    <!--交换机 广播-->

 <rabbit:direct-exchange name="test_exchange_confirm" >

        <!--绑定queue-->

 <rabbit:bindings>

            <rabbit:binding queue="test_queue_confirm" key="confirm" ></rabbit:binding>

        </rabbit:bindings>

    </rabbit:direct-exchange>



    <!--ttl消息存活时间-->

 <!--声明队列queue-->

 <rabbit:queue name="test_queue_ttl" id="test_queue_ttl" >

            <!--设置queue队列存活时间-->

 <rabbit:queue-arguments>

            <!--通过entry设置属性 键值对应的格式 key=键 value=值 value-type表示值的数据类型-->



 <!--设置队列存活为10s 默认单位为毫秒

 由于存换时间是int类型 这边声明的是字符串 需要使用value-type表名值是整形的 不指定会报错的 -->

 <!--x-message-ttl 指队列的过期时间-->

 <entry key="x-message-ttl" value="100000" value-type="java.lang.Integer" ></entry>

      

        </rabbit:queue-arguments>

    </rabbit:queue>



    <!--声明交换机-->

 <rabbit:topic-exchange name="test_exchange_ttl" >

        <!--绑定交换机和队列-->

 <rabbit:bindings>

            <rabbit:binding pattern="ttl.#" queue="test_queue_ttl" ></rabbit:binding>

        </rabbit:bindings>

    </rabbit:topic-exchange>





    <!--

 死信队列

 1 声明正常队列和交换机test_queue_dlx   test_exchange_dlx

 2 声明死信队列和死信交换机  queue_dlx exchange_dlx

 3 正常队列绑定死信交换机

 设置两个参数

 1死信交换机名称  x-dead-letter-exchange

 2发送给死信交换机的路由键 x-dead-letter-routing-key

 -->



 <!-- 1 声明正常队列和交换机test_queue_dlx   test_exchange_dlx-->

 <rabbit:queue id="test_queue_dlx" name="test_queue_dlx" >

        <!-- 3正常队列绑定死信交换机 使用arguments和entry设置参数-->

 <rabbit:queue-arguments>

            <!--3.1死信交换机名称  x-dead-letter-exchange-->

 <entry key="x-dead-letter-exchange" value="exchange_dlx" />

            <!--3.2发送给死信交换机的路由键 x-dead-letter-routing-key-->

 <entry key="x-dead-letter-routing-key" value="dlx.hehe" />

            <!--4让消息完成死信-->

 <!--4.1 设置队列过期时间 ttl 10s-->

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

    </rabbit:topic-exchange>



    <!-- 2 声明死信队列和死信交换机  queue_dlx exchange_dlx-->

 <rabbit:queue id="queue_dlx" name="queue_dlx" />

    <!--设置正常的交换机-->

 <rabbit:topic-exchange name="exchange_dlx" >

        <!--正常交换机绑定正常队列-->

 <rabbit:bindings>

            <rabbit:binding pattern="dlx.#" queue="queue_dlx" />

        </rabbit:bindings>

    </rabbit:topic-exchange>





</beans>
复制代码

消费者配置

 <? xml version="1.0" encoding="UTF-8" ?>

<beans xmlns="http://www.springframework.org/schema/beans"

 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

 xmlns:context="http://www.springframework.org/schema/context"

 xmlns:rabbit="http://www.springframework.org/schema/rabbit"

 xsi:schemaLocation="http://www.springframework.org/schema/beans

 http://www.springframework.org/schema/beans/spring-beans.xsd

 http://www.springframework.org/schema/context

 https://www.springframework.org/schema/context/spring-context.xsd

 http://www.springframework.org/schema/rabbit

 http://www.springframework.org/schema/rabbit/spring-rabbit.xsd" >

    <!--加载配置文件-->

 <context:property-placeholder location="classpath:application.properties" />



    <!-- 定义rabbitmq connectionFactory -->

 <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"

 port="${rabbitmq.port}"

 username="${rabbitmq.username}"

 password="${rabbitmq.password}"

 virtual-host="${rabbitmq.virtual-host}"



 />



    <!--配置监听器bean 扫描这个包的路径 配置类需要加注解-->

 <context:component-scan base-package="com.wyh.listener" />

    <!--定义监听器    acknowledge="" 设置签收方式 -->

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

        <!--加载对应监听器的类 具体的类名和监听的队列名-->

 <!--  <rabbit:listener ref="qasCKListener" queue-names="test_queue_confirm"  />-->

 <!--

 <rabbit:listener ref="rabbitMQACKListener" queue-names="test_queue_confirm"  />

-->

 <!--监听正常队列 拒绝签收并不允许重返队列 成为死信队列-->

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

    </rabbit:listener-container>



</beans>
复制代码

生产者发送消息

package com.producer.test;



import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.amqp.AmqpException;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.MessagePostProcessor;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.test.context.ContextConfiguration;

import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;



 /**

 *  @prog ram: SpringBoot_RabbitMQ_Advanced

 *  @description:  测试确认模式消息是否发送成功

 *  @author:  魏一鹤

 *  @createDate:  2022-04-04 23:10

 **/



 //spring配置文件

@RunWith(SpringJUnit4ClassRunner.class)

 //加载文件的路径

@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml" )

public class ProducerTest {

    //注入 RabbitTemplate

 @Autowired

    private RabbitTemplate rabbitTemplate;



    /**

 * 确认模式:

 * 步骤

 * 1确认模式的开启:在connectionFactory中开启,默认是false不开启的 publisher-confirms="true"

 * 2回调函数的编写:在RabbitTemplate模板工具类定义ConfirmCallBack(回调函数).当消息发送出去的时候回调函数会自动执行,返回true(成功)或者false(失败)

 **/

 @Test

    public void testConfirm() {

        //定义确认模式的回调函数

  rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            @Override

            //匿名内部类

 /**

 * confirm有三个参数 下面一一说明

 * CorrelationData correlationData 相关的配置信息

 * boolean ack  exchange交换机是否成功收到了消息 true成功false失败

 * String cause 失败原因 ack=false

 **/

  public void confirm(CorrelationData correlationData, boolean ack, String cause) {

                System.out.println( "确认模式的回调函数被执行了!" ); //确认模式的回调函数被执行了!

 System.out.println( "消息是否发送成功?" +ack);

                if(ack){

                    //交换机接收成功生产者发送的消息

 System.out.println( "接收成功消息！原因是:" +cause);

                }else{

                    //交换机接收没有成功生产者发送的消息

 System.out.println( "接收失败消息！原因是:" +cause);

                    //将来会做处理,就算消息发送失败也会重新去发送消息,保证消息第二次发送成功

 }

            }

        });

        //发送消息

  rabbitTemplate.convertAndSend( "test_exchange_confirm" , "confirm" , "message confirm..." );

    }





    /**

 * 回退模式:当消息发送给Exchange交换机后,交换机路由到queue失败时才会执行ReturnCallBack

 * 步骤

 * 1回退模式的开启:在connectionFactory中开启,默认是false不开启的 publisher-returns="true"

 * 2回调函数的编写:在RabbitTemplate模板工具类定义ConfirmCallBack(回调函数).当消息发送出去的时候回调函数会自动执行,返回true(成功)或者false(失败)

 * 3设置Exchange处理消息的模式 它有两种模式

 *   3.1 第一种模式:如果消息没有路由到queue队列,则会丢弃消息(默认的方式)

 *   3.2 第二种模式:如果消息没有路由到queue队列,则返回给消息的发送方ReturnCallBack

 **/



 @Test

    public void testReturn() {



        //由于处理消息的模式默认是如果消息没有路由到queue队列,则会丢弃消息

 //所以需要设置交换机处理消息的模式,交换机会把消息返回给对应的生产者,生产者通过监听就能拿到消息

  rabbitTemplate.setMandatory(true);

        //编写returnCallBack回调函数

  rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback(){

            @Override

            //匿名内部类

 /**

 * returnedMessage有五个参数 下面一一说明

 * Message message 消息对象

 * int replyCode    返回编码 错误码

 * String replyText 错误信息

 * String exchange  交换机

 * String routingKey 路由键

 **/

  public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

                System.out.println( "返回模式的回调函数被执行了!" );

                System.out.println( "消息对象:" +message);

                System.out.println( "返回编码(错误码):" +replyCode);

                System.out.println( "错误信息:" +replyText);

                System.out.println( "交换机:" +exchange);

                System.out.println( "路由键:" +routingKey);

                //将来会做处理 把信息重新路由



 //-----------------------------------------打印信息

 //返回模式的回调函数被执行了!

 //消息对象:(Body:'message confirm...' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])

 //返回编码(错误码):312

 //错误信息:NO_ROUTE

 //交换机:test_exchange_confirm

 //路由键:confirm111

 }



        });

        //发送消息

  rabbitTemplate.convertAndSend( "test_exchange_confirm" , "confirm" , "message confirm..." );

    }



    //专门测试限流循环发送消息

 @Test

    public void testSend() {

        for (int i = 0; i < 10; i++) {

            //发送消息

  rabbitTemplate.convertAndSend( "test_exchange_confirm" , "confirm" , "message confirm..." );

        }

    }



    //测试ttl

 /**

 *  ttl 过期时间分为两种

 *  1队列统一过期

 *  2消息单独过期

 *  如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准！

 *  如果队列过期后,会将队列所有消息全部移除,因为是统一的

 *  消息过期后,只有消息在队列顶端(快要被消费),才会判断其是否过期,如果过期就会被移除掉

 **/

 @Test

    public void testTTL() {

        //消息单独过期 在convertAndSend方法中心新增一个参数 MessagePostProcessor 加载匿名内部类和它的重写方法

 //MessagePostProcessor 消息处理对象 设置一些消息参数信息



 //发送消息

  for (int i = 0; i < 10; i++) {

            if(i==5){

                //过期的消息

  rabbitTemplate.convertAndSend( "test_exchange_ttl" , "ttl.weiyihe" , "message ttl..." ,new MessagePostProcessor(){

                    @Override

                    public Message postProcessMessage(Message message) throws AmqpException {

                        //设置message的信息  setExpiration 设置消息过期时间

 message.getMessageProperties().setExpiration( "10000" );

                        //返回该消息

  return message;

                    }

                });

            }else{

                //不过期的消息

  rabbitTemplate.convertAndSend( "test_exchange_ttl" , "ttl.weiyihe" , "message ttl..." );

            }

        }

    }





    /**

 * 发送测试私信消息

 * 1、消息过期时间的测试

 * 2、队列长度限制的测试

 * 3、消费者消息拒收的限制

 **/

 @Test

    public void testDlx(){

        //1、测试过期时间的死信 给正常的交换机发送消息

 //如果消息正常发送到正常的交换机,过了10s会自动去死信队列

 //发送消息

 //rabbitTemplate.convertAndSend("test_exchange_dlx", "test.dlx.haha", "我是一条消息,我会死吗?");



 //2、队列长度限制的测试

 //发送20条信息 现在x-max-length是10

 //如果发送成功 那么因为设置了最大长度是10,只会有10条进行正常队列

 // 剩下的会跑到死信队列,过了10s后正常队列中的消息也会自动跑到死信队列中

 //for (int i = 0; i < 20; i++) {

 //    //发送消息

 //    rabbitTemplate.convertAndSend("test_exchange_dlx", "test.dlx.haha", "我是一条消息,我会死吗?");

 //}



 //3 消费者消息拒收的限制

 //消费者Consumer监听正常的队列,然后让消息拒绝接收并且不重回队列

  rabbitTemplate.convertAndSend( "test_exchange_dlx" , "test.dlx.haha" , "我是一条消息,我会死吗?" );



    }

}
复制代码

消费者拒收消息

package com.wyh.listener;



import com.rabbitmq.client.Channel;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import org.springframework.stereotype.Component;



import java.util.concurrent.TimeUnit;



 /**

 *  @program:  SpringBoot_RabbitMQ_Advanced

 *  @description:  RabbitMQ 死信通信

 *  @author:  魏一鹤

 *  @createDate:  2022-04-06 20:30

 **/



 //包扫描注解 把bean加载到spring容器中

@Component

 //实现MessageListener接口并重写onMessage方法

public class DLXCKListener 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 num=3/0;

            channel.basicAck(deliveryTag,true);

        } catch (Exception e) {

            System.out.println( "出现异常,消费者拒绝签收！" );

            //死信队列 拒绝签收requeue=false 将消息路由到死信队列中

 channel.basicNack(deliveryTag,true,false);



        }

    }

}
复制代码