04-21 02:42 阅读 136

关注

RabbitMQ——实战篇3（Spring集成高级特性：死信队列，消息丢失，延迟队列）

文章目录

• 其他文章地址

• 1、高级特性

• 2、代码实现（具体的都在注释上）

• 2.2.1、pom

• 2.2.2、配置文件

• 2.2.3、测试ACK

• 2.2.4、死信队列测试

• 2.2.5、延迟队列

• 2.2.6、限流

• 2.2.7、启动类

• 2.1.1、pom

• 2.1.2、配置文件

• 2.1.3、测试服务端

• 2.1、服务端

• 2.2、客户端

其他文章地址

1、RabbitMQ——单机版安装（3.6.5）
2、RabbitMQ——入门篇
3、RabbitMQ——实战篇1（原生API）
4、RabbitMQ——实战篇2（Spring集成）
5、RabbitMQ——实战篇3（Spring集成高级特性：死信队列，消息丢失，延迟队列）
6、RabbitMQ——实战篇4（SpringBoot集成）

项目地址：https://gitee.com/zhouzhz/rabbitmq

1、高级特性

1、消费端限流
2、TTL
3、死信队列
4、延迟队列
5、消息可靠性投递
6、Consumer ACK

2、代码实现（具体的都在注释上）

2.1、服务端

2.1.1、pom

<dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.1.7.RELEASE</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>5.1.7.RELEASE</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.0</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build>

2.1.2、配置文件

rabbitmq.properties

rabbitmq.host=192.168.0.66rabbitmq.port=5672rabbitmq.username=zhzmq
rabbitmq.password=zhzmq
rabbitmq.virtual-host=/zhztest

spring-rabbitmq-producer.xml

<?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:rabbitmq.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对象操作可以在代码中方便发送消息--><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"><rabbit:bindings><rabbit:binding queue="test_queue_confirm" key="confirm"></rabbit:binding></rabbit:bindings></rabbit:direct-exchange><!--######################################消息可靠性投递 结束 ##########################################--><!--####################################TTL 开始 ################################--><!--ttl--> <rabbit:queue name="test_queue_ttl" id="test_queue_ttl"> <!--设置queue的参数--> <rabbit:queue-arguments> <!--x-message-ttl指队列的过期时间--> <entry key="x-message-ttl" value="10000" 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><!--######################################TTL 结束 ##########################################--><!--#######################################死信队列 开始 #########################################--><!--死信队列：1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx)3. 正常队列绑定死信交换机
                设置两个参数：* x-dead-letter-exchange：死信交换机名称* x-dead-letter-routing-key：发送给死信交换机的routingkey--><!--1. 声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)--><rabbit:queue name="test_queue_dlx" id="test_queue_dlx"><!--3. 正常队列绑定死信交换机--><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：发送给死信交换机的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><!--######################################死信队列 结束##########################################--><!--########################################延迟队列 开始########################################--><!--延迟队列：1. 定义正常交换机（order_exchange）和队列(order_queue)2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx)3. 绑定，设置正常队列过期时间为30分钟--><!-- 1. 定义正常交换机（order_exchange）和队列(order_queue)--><rabbit:queue id="order_queue" name="order_queue"><!--3. 绑定，设置正常队列过期时间为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><!--2. 定义死信交换机（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><!--######################################延迟队列 结束 ##########################################--></beans>

2.1.3、测试服务端

package com.zhz.test;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.lang.NonNull;import org.springframework.lang.Nullable;import org.springframework.test.context.ContextConfiguration;import org.springframework.test.context.junit4.SpringJUnit4Cla***unner;/**
 * @author ：zhz
 * @date ：Created in 2021/02/06
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 * @description:
 **/@RunWith(SpringJUnit4Cla***unner.class)@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")public class HighProducerTest {@Autowiredprivate RabbitTemplate rabbitTemplate;/**
     * 解决丢失消息：
     *      消息传递路劲：producer--->rabbitmq broker--->exchange--->queue--->consumer
     *      生产者：
     *          1、confirm模式配合return模式，confirm模式一般是到达不了exchange才会触发，return模式是到达不了queue才会触发
     *              1、设置ConnectionFactory的publisher-confirms="true" 开启 确认模式。
     *              使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在
     *              方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。
     *              2、设置ConnectionFactory的publisher-returns="true" 开启 退回模式。
     *              使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，
     *              如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。
     *              并执行回调函数returnedMessage。
     *          2、setChannelTransacted();不建议使用，因为吞吐量减低250倍
     *      rabbitmq端：
     *          exchange持久化，queue持久化，message持久化
     *      消费者：（手动ACK）
     *          在rabbit:listener-container标签中设置acknowledge属性，设置ack方式 none：自动确认，manual：手动确认
     *          如果在消费端没有出现异常，则调用channel.basicAck(deliveryTag,false);方法确认签收消息
     *          如果出现异常，则在catch中调用 basicNack或 basicReject，拒绝消息，让MQ重新发送消息。
     *      broker端：
     *          高可用
     */@Testpublic void testConfirm() {//回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/**
             *
             * @param correlationData 相关配置信息
             * @param ack   exchange交换机 是否成功收到了消息。true 成功，false代表失败
             * @param cause 失败原因
             */@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("confirm方法执行-------------");if (ack) {//接收成功System.out.println("接收成功：" + cause);} else {//接收失败System.out.println("接收失败：" + cause);//消息重试}}});rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "消息确认");//进行睡眠操作===》正常业务操作中不需要try {Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();}}//return模式@Testpublic void testReturn() {//设置交换机处理失败消息的模式   为true的时候，消息达到不了 队列时，会将消息重新返回给生产者rabbitTemplate.setMandatory(true);//定义回调rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/**
             *
             * @param message   消息对象
             * @param replyCode 错误码
             * @param replyText 错误信息
             * @param exchange  交换机
             * @param routingKey 路由键
             */@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("执行中------");System.out.println(message + "----" + replyCode + "----" + replyText + "----" + exchange + "----" + routingKey);//处理}});rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "消息回退中");//进行睡眠操作try {Thread.sleep(5000);} catch (Exception e) {e.printStackTrace();}}/**
     * 解决消费者端限流：
     *      在<rabbit:listener-container> 中配置 prefetch属性设置消费端一次拉取多少消息
     *      消费端的确认模式一定为手动确认。acknowledge="manual"
     *///批量发送消息，让消费者每次拉去指定的数量@Testpublic void testQos(){for (int i = 0; i < 10; i++) {//发送消息rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "消息发送中");}}/**
     * TTL:过期时间
     *  1. 队列统一过期
     *
     *  2. 消息单独过期
     *
     * 如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。
     * 队列过期后，会将队列所有消息全部移除。
     * 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉)
     *
     */@Testpublic void testTtl(){for (int i = 0; i < 10; i++) {// 发送消息rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.zhz", "测试ttl");}}/**
     * 发送测试死信消息：
     *  1. 过期时间
     *  2. 长度限制
     *  3. 消息拒收
     */@Testpublic void testDlx(){//1. 测试过期时间，死信消息rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.hehe","测试死信队列消息过期时间");//2. 测试长度限制后，消息死信for (int i = 0; i < 20; i++) {rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.hehe","测试死信队列消息长度限制后");}//3. 测试消息拒收rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.zhz","测试死信队列消息拒收");}/**
     * 模拟延迟队列
     * 应用场景：下单成功之后，发送消息（取消下单）
     */@Testpublic void testDelay() throws InterruptedException {//1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息：id=1,time=2021年2月...");//2.打印倒计时10秒for (int i = 10; i > 0 ; i--) {System.out.println(i+"...");Thread.sleep(1000);}}}

2.2、客户端

2.2.1、pom

 <dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.1.7.RELEASE</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.8.RELEASE</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>5.1.7.RELEASE</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.0</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build>

2.2.2、配置文件

rabbitmq.properties

rabbitmq.host=192.168.0.66rabbitmq.port=5672rabbitmq.username=zhzmq
rabbitmq.password=zhzmq
rabbitmq.virtual-host=/zhztest

spring-rabbitmq-consumer.xml

<?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:rabbitmq.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}"/><context:component-scan base-package="com.zhz.listener" /><!--定义监听器容器
      acknowledge="manual":手动签收
      prefetch="1":每次抓取多少条消息--><!--定义监听器容器  acknowledge="manual"  prefetch="1"  --><rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"><rabbit:listener ref="ackListener" queue-names="test_queue_confirm"></rabbit:listener><!--<rabbit:listener ref="qosListener" queue-names="test_queue_confirm"></rabbit:listener>--><!--定义监听器，监听正常队列--><!--<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"></rabbit:listener>--><!--延迟队列效果实现：  一定要监听的是 死信队列！！！--><rabbit:listener ref="orderListener" queue-names="order_queue_dlx"></rabbit:listener></rabbit:listener-container></beans>

2.2.3、测试ACK

package com.zhz.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;/**
 * @author ：zhz
 * @date ：Created in 2021/02/06
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 * @description: ack也就是手动提交
 **/@Componentpublic class AckListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {//1、获取消息的idlong deliveryTag = message.getMessageProperties().getDeliveryTag();try {//2、获取消息System.out.println("message:" + new String(message.getBody()));//3、进行业务处理System.out.println("=====进行业务处理====");//模拟出现异常//int  i = 5/0;//4、进行消息签收channel.basicAck(deliveryTag, false);System.out.println("收到了消息:" + deliveryTag);} catch (Exception e) {//拒绝签收 /*
            第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
             */channel.basicNack(deliveryTag, false, true);}}}

2.2.4、死信队列测试

package com.zhz.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;/**
 * @author ：zhz
 * @date ：Created in 2021/02/06
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 * @description: 死信队列
 **/@Componentpublic class DlxListener implements ChannelAwareMessageListener {@Overridepublic 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=falsechannel.basicNack(deliveryTag,true,false);}}}

2.2.5、延迟队列

package com.zhz.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;/**
 * @author ：zhz
 * @date ：Created in 2021/02/06
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 * @description: 延迟队列
 **/@Componentpublic class OrderListener implements ChannelAwareMessageListener {@Overridepublic 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=falsechannel.basicNack(deliveryTag,true,false);}}}

2.2.6、限流

package com.zhz.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;/**
 * @author ：zhz
 * @date ：Created in 2021/02/06
 * @version: V1.0
 * @slogan: 天下风云出我辈，一入代码岁月催
 * @description: 限流
 **/@Componentpublic class QosListener implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {  //获取到的消息System.out.println(new String(message.getBody()));Thread.sleep(1000);//处理业务逻辑//进行消息的签收//channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}}

2.2.7、启动类

package com.zhz.listener;import org.springframework.context.ApplicationContext;import org.springframework.context.support.ClassPathXmlApplicationContext;public class Test {public static void main(String[] args) {//初始化IOC容器ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring-rabbitmq-consumer.xml");}}