搭建SpringBoot项目,用于演示

springboot版本

<!-- spring boot --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>2.3.2.RELEASE</version> <type>pom</type> <!-- import 导入父工程的配置--> <scope>import</scope> </dependency>

消费与提供方的pom.xml

<dependencies> <!-- spring-boot-starter-web spring-boot-starter-actuator绑定在一块 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-actuator</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!--2. rabbitmq--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> </dependency> </dependencies>

提供端的application.yml

# 配置RabbitMQ的基本信息 ip 端口 username password 虚拟机.. spring: rabbitmq: host: 192.168.93.132 # ip port: 5672 username: xiaofu password: xiaofu virtual-host: /springboot # 旧版本 开启 confirm 确认模式 # publisher-confirms: true # 新版的开启 confirm 确认模式 publisher-confirm-type: correlated # 开启 return 退回模式 publisher-returns: true

消费端的application.yml

spring: rabbitmq: host: 192.168.93.132 #主机ip port: 5672 #端口 username: xiaofu password: xiaofu virtual-host: /springboot # 旧版本 开启 confirm 确认模式 # publisher-confirms: true # 新版的开启 confirm 确认模式 publisher-confirm-type: correlated # 开启 return 退回模式 publisher-returns: true listener: # RabbitMQ模式使用simple simple支持事务的 simple: # Consumer ACK机制：设置为手动签收 acknowledge-mode: manual prefetch: 1 # 限流,配置1 表示消费端每次向MQ拉取最大一条消息 # direct 是不支持事务的 # direct: # acknowledge-mode: manual # ACK机制：设置为手动签收 # retry: # enabled: true # 是否支持重试 # max-attempts: 3 # 重试机制,3次

1.消息的可靠投递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提 供了两种方式用来控制消息的投递可靠性模式。 ⚫ confirm 确认模式 ⚫ return 退回模式 rabbitmq 整个消息投递的路径为： producer—>rabbitmq broker—>exchange—>queue—>consumer ⚫ 消息从 producer 到 exchange 则会返回一个 confirmCallback 。 ⚫== 消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。== 我们将利用这两个 callback 控制消息的可靠性投递

持久化 • exchange要持久化 • queue要持久化 • message要持久化生产方确认Confirm消费方确认AckBroker高可用

1.1配置confirm 确认模式与return 退回模式

# 旧版本 开启 confirm 确认模式 # publisher-confirms: true # 新版的开启 confirm 确认模式 publisher-confirm-type: correlated # 开启 return 退回模式 publisher-returns: true

1.2创建用于测试消息的可靠投递的交换机与队列

package com.fs.rabbitmq.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /* 消息的可靠投递 创建测试交换机与队列 测试交换机可靠性记得在application.yml中配置 #开启 confirm 确认模式 设置为默认的自动确认模式 publisher-confirm-type: none */ @Configuration public class RabbitMQConfigConfirmAndReturn { //创建交换机 @Bean public Exchange exchangeConfirm(){ //创建一个Direct：定向，把消息交给符合指定routing key 的队列的交换机 return ExchangeBuilder.directExchange("test_Exchange_Confirm").build(); } //创建一个队列 @Bean public Queue queueConfirm(){ //创建一个队列而且是持久的 return QueueBuilder.durable("test_Queue_Confirm").build(); } // 队列和交换机绑定关系 Binding /* 1. 指定哪个队列 2. 指定哪个交换机 3. routing key */ @Bean public Binding bindingConfirm(@Qualifier("exchangeConfirm") Exchange exchange,@Qualifier("queueConfirm") Queue queue){ //把队列绑定在交换机上指定routingKey没有参数 return BindingBuilder.bind(queue).to(exchange).with("testConfirm").noargs(); } // return 退回模式 //使用test_Exchange_Confirm这个交换机 }

1.3 ProducerTest测试类中编写测试confirm与return测试方法

1.3.1首先在测试类中注入RabbitTemplate

@SpringBootTest @RunWith(SpringRunner.class) public class ProducerTest { //1.注入RabbitTemplate @Autowired private RabbitTemplate rabbitTemplate; }

1.3.2测试方法

/** * 确认模式: 该模式是来校验消息是否发送成功到交换机中 * 步骤: * 1.确认开启: publisher-confirm-type: none * 2.在rabbitTemplate定义一个confirmCallBack回调函数 */ @Test public void testConfirm(){ //使用rabbitTemplate的确认回调方法 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * @param correlationData 相关的配置信息 * @param ack 代表了Exchange交换机是否收到了消息,true表示收到了消息,false表示交换机没有收到消息 * @param cause 失败的原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println("发送消息后,回调方法执行了~~~"); if (ack){ System.out.println("发送消息成功,启动成功方案:"+cause); }else { System.out.println("发送消息失败,启动失败方案:"+cause); } } }); //发送消息,假设写错交换机的名称,肯定会发送到Exchange失败,就会执行我们的confirmCallBack回调方法 rabbitTemplate.convertAndSend("test_Exchange_Confirm","testConfirm","测试Confirm确认模式~~~"); } /** * 回退模式: 该模式是用来校验该消息是否从Exchange交换机成功路由到了queue队列中 * 当Exchange路由到queue失败后,就会执行这个ReturnCallBack方法 * * 步骤: * 1.开启回退模式: publisher-returns: true * 2.设置ReturnCallBack * 3,设置Exchange处理的消息的模式 * 1.如果消息没有路由到Queue中,则丢弃消息(默认) * 2.如果消息没有路由到Queue中,返回给消息到发送方的ReturnCallBack方法 */ @Test public void testReturn() { //设置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) { //当我们的消息发送从Exchange交换机发送到Queue错误后就会执行这个回调方法 System.out.println("ReturnCallBack 执行了~~~"); System.out.println(message); System.out.println(replyCode); System.out.println(replyText); System.out.println(exchange); System.out.println(routingKey); } }); //发送消息, /* 测试1: 使用正确的Exchange与routingKey执行成功,不会执行我们的ReturnCallBack回退方法 测试2: 使用正确的Exchange与错误的不存在的routingKey,就会执行我们的ReturnCallBack回退方法 */ rabbitTemplate.convertAndSend("test_Exchange_Confirm", "testConfirm111", "testConfirm~~~发送消息,测试回退模式"); }

1.3.3 测试消息可靠投递

confirm测试 return测试

2.Consumer Ack

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。

有三种确认方式： • 自动确认：acknowledge=“none” • 手动确认：acknowledge=“manual” • 根据异常情况确认：acknowledge=“auto”，（这种方式使用麻烦，不作讲解）

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

2.1 消费端application.yml中配置

需要开启手动签收消息

listener: # RabbitMQ模式使用simple simple支持事务的 simple: # Consumer ACK机制：设置为手动签收 acknowledge-mode: manual prefetch: 1 # 限流,配置1 表示消费端每次向MQ拉取最大一条消息

2.2 在消费端创建监听类

在方法上使用下面的注解,监听的队列 @RabbitListener(queues = “队列名称”)

下面的代码监听的是我们上面测试confirm的队列

package com.fs.Queuelistener; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; /** * Consumer ACK机制：默认自动签收 * 1. 设置手动签收。acknowledge="manual" * 2. 让监听器类实现ChannelAwareMessageListener接口 * 3. 如果消息成功处理，则调用channel的 basicAck()签收 * 4. 如果消息处理失败，则调用channel的basicNack()拒绝签收，broker重新发送给consumer */ @Component public class AckListener { @RabbitListener(queues = "test_Queue_Confirm") public void testAck(Message message, Channel channel) throws IOException { //得到消息的唯一deliveryTag long deliveryTag = message.getMessageProperties().getDeliveryTag(); //模拟接收到消息消费的逻辑 try{ //接收到消息进行消费 System.out.println(new String(message.getBody())); System.out.println("消息到了ACK机制中~~~"); //模拟执行逻辑错误 // int i = 1/0; //手动签收消息 /* deliveryTag:表示收到的消息的参数标签(消息的唯一id) 第二个参数:是否签收多条消息(批量签收消息) */ channel.basicAck(deliveryTag,true); }catch (Exception e){ //当我们上面的逻辑出现错误,就不会签收消息,我们在catch中就执行拒绝签收 System.out.println("消费逻辑出现异常~~~消息被Ack机制重回队列"); //拒绝签收 /* 第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue的尾部，broker会重新发送该消息给消费端,false为丢弃改消息,若设置了死信队列,就会交给死信队列 */ channel.basicNack(deliveryTag,true,false); } } }

2.3 测试ACK

启动主启动:ConsumerSpringbootApplication 在提供方发送消息 在消费方查看消息被消费

3.TTL 全称 Time To Live（存活时间/过期时间）

➢ TTL 全称 Time To Live（存活时间/过期时间）。 ➢ 当消息到达存活时间后，还没有被消费，会被自动清除。 ➢ RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

➢ 设置队列过期时间使用参数：x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。 ➢ 设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。 ➢ 如果两者都进行了设置，以时间短的为准。

3.1 在提供方编写TTL的交换机与队列的创建代码

package com.fs.rabbitmq.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /* 测试RabbitMQ的TTL */ @Configuration public class RabbitMQConfigTTL { //创建交换机 @Bean public Exchange exchangeTtl(){ //创建一个Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列 的交换机 return ExchangeBuilder.topicExchange("test_Exchange_TTL").build(); } //创建队列 @Bean public Queue queueTtl(){ //创建一个队列,设置消息过期时间为10秒 return QueueBuilder.durable("test_Queue_TTL").withArgument("x-message-ttl",10000).build(); } //绑定交换机与队列 @Bean public Binding bindingTtl(@Qualifier("exchangeTtl") Exchange exchange, @Qualifier("queueTtl") Queue queue){ //将队列绑定在topic通配符交换机上设置路由规则routingKey,没有参数 return BindingBuilder.bind(queue).to(exchange).with("ttl.#").noargs(); } }

3.2 提供方测试代码编写

/** * TTL:过期时间 * 1. 队列统一过期 * * 2. 消息单独过期 * 结果: * 如果设置了消息的过期时间，也设置了队列的过期时间，它以时间短的为准。 * 队列过期后，会将队列所有消息全部移除。 * 消息过期后，只有消息在队列顶端，才会判断其是否过期(移除掉) */ @Test public void testTTL(){ //* 1. 队列统一过期 //发送10条消息,不去消费,查看web控制台10秒后这10条消息是否会被丢弃 // for (int i = 0; i < 10; i++) { // //调用方法 // rabbitTemplate.convertAndSend("test_Exchange_TTL","ttl.hehe.xf","测试TTL超时时间队列消息发送~~~"+i); // } //* 2. 消息单独过期 // 消息后处理对象，设置一些消息的参数信息,发送消息的时候传递该参数,那么这些消息就会具有该参数 //该对象是一个接口,使用匿名类部内来创建实现类 MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { //设置发送消息的参数 @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000");//设置消息过期时间为5秒 return message; } }; //再次发生一条消息,使用我们设置好的消息参数对消息进行封装 //发送成功后,去看我们的队列中的这条消息是否是5秒过期,因为我们这个消息是在队列的顶端,等待被消费,而且过期时间短于队列统一时间,所以优先我们这单条消息过期时间 // rabbitTemplate.convertAndSend("test_Exchange_TTL","ttl.hehe.fs","我被使用了消息参数,5秒后过期~~~",messagePostProcessor); //我们再次极端的测试,让我们这条消息不在队列的顶端 //这条i==5的消息设置5秒过期,但是他在队列的中间,5秒后已经过期,但是不会被队列移除掉,当队列统一的过期时间到了,就会随着统一被队列丢弃 或者交给死信交换机 //因为队列只会移除队列顶端的过期消息,例如当有消费者来消费这10条消息后,但是i=5这条消息 //已经过期,当消费到这条消息时,它就在队列的顶端,就会判断该消息是否过期, //若过期,者就会移除,或者交给 死信交换机 //不会发送给消费者消费的 for (int i = 0; i < 10; i++) { if (i == 5) { //消息单独过期 rabbitTemplate.convertAndSend("test_Exchange_TTL", "ttl.hehe.xf", "我被使用了消息参数,5秒后过期~~~而且在队列的中间,我会不会5秒后过期呢?", messagePostProcessor); } else { //不过期的消息 rabbitTemplate.convertAndSend("test_Exchange_TTL", "ttl.hehe.xf", "我发送了消息...."); } } }

3.3 测试发送,查询queue队列中的消息存活时间

3.3.1 测试 队列统一过期

将1. 队列统一过期这段代码注释放开,把其余代码注释,然后点击运行

3.3.2 测试 消息单独过期

使用这个类MessagePostProcessor来封装我们发生消息的属性参数

3.3.3 测试 ,让我们这条消息不在队列的顶端

4.死信队列 DLX 。Dead Letter Exchange（死信交换机）

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以 被重新发送到另一个交换机，这个交换机就是DLX。 消息成为死信的三种情况：

队列消息长度到达限制；消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机： 给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

4.1 创建用于测试死信队列的交换机与队列

死信队列： 1. 声明正常的队列(test_queue_dlx)和正常交换机(test_exchange_dlx) 2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx) 3. 正常队列绑定死信交换机 设置两个参数： * x-dead-letter-exchange：死信交换机名称 * x-dead-letter-routing-key：发送给死信交换机的routingkey

package com.fs.rabbitmq.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; import java.util.Map; /* 死信队列 死信队列： 1. 声明正常的队列(test_queue_dlx)和正常交换机(test_exchange_dlx) 2. 声明死信队列(queue_dlx)和死信交换机(exchange_dlx) 3. 正常队列绑定死信交换机 设置两个参数： * x-dead-letter-exchange：死信交换机名称 * x-dead-letter-routing-key：发送给死信交换机的routingkey */ @Configuration public class RabbitMQDeadMessageConfig { //创建自定义 死信交换机 逻辑认为是用来做死信服务的 @Bean public Exchange exchangeDlx(){ return ExchangeBuilder.topicExchange("exchange_del").build(); } //创建自定义 死信队列 逻辑认为是用来做死信服务的 @Bean public Queue queueDlx(){ return QueueBuilder.durable("queue_dlx").build(); } //将自定义的死信队列绑定在一块 @Bean public Binding bindingDlx(@Qualifier("exchangeDlx") Exchange exchange,@Qualifier("queueDlx") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("dlx.*").noargs(); } //创建正常接收消息的交换机 @Bean public Exchange exchangeNormalDlx(){ return ExchangeBuilder.topicExchange("exchange_Normal_DLX").build(); } //创建正常接收消息的队列,绑定我们的死信交换机 @Bean public Queue queueNormalDlx(){ return QueueBuilder.durable("queue_Normal_DLX")//正常队列的名称 .withArgument("x-dead-letter-exchange","exchange_del")//设置改队列的死信交换机 .withArgument("x-dead-letter-routing-key","dlx.xf")//设置该队列的发送消息时指定的routingkey .withArgument("x-message-ttl",10000)//设置队列中消息的过期时间 .withArgument("x-max-length",10).build();//设置队列的最大容量 } // @Bean // public Queue queueNormalDlx(@Qualifier("exchangeDlx") Exchange exchange,@Qualifier("queueDlx") Queue queue){ // Map<String, Object> args = new HashMap<>(); // // set the queue with a dead letter feature // args.put("x-dead-letter-exchange", exchange);//设置该队列的死信交换机 // args.put("x-dead-letter-routing-key", "dlx.xf");//设置该队列的发送消息时指定的routingkey // args.put("x-message-ttl",10000);//设置队列中消息的过期时间 // args.put("x-max-length",10);//设置队列的最大容量 // return new Queue("queue_Normal_DLX", true, false, false, args); // } //将正常的交换机与队列绑定 @Bean public Binding bindingNormalDlx(@Qualifier("exchangeNormalDlx") Exchange exchange,@Qualifier("queueNormalDlx") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("test.dlx.#").noargs(); } }

4.2 编写测试类发送测试死信消息

* 发送测试死信消息： * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 /** * 发送测试死信消息： * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 */ @Test public void testDlx() throws InterruptedException { //测试过期时间,死信消息,首先发送给了正常的交换机,交换机路由到正常的队列,然后该队列的消息由于设置了10秒过期,10秒内没有被消费 // 过期后就交给死信交换机,然后由死信交换机路由到死信队列,然后被消费掉 // rabbitTemplate.convertAndSend("exchange_Normal_DLX","test.dlx.xf","我发送了一条10秒后就过期的消息~~~"); //测试队列长度,当一次性发送超过队列长度的消息,队列就会将多余的消息交给死信交换机 //由于我们创建队列的时候,改队列的长度为10,那么就有10 条消息被第一时间交给死信交换机,然后在等10秒,10秒后队列中的10条消息没有被消费,也会交给死信交换机 //由执行控制台结果得知,队列是先进先出的原则先进的0-9会被后进的10-19挤出来,所以0-9先变成死信消息,而10-19是10秒过期后未被消费成的死信消息 // for (int i = 0; i < 20; i++) { // Thread.sleep(10); // rabbitTemplate.convertAndSend("exchange_Normal_DLX","test.dlx.xf","我发送了多条10秒后就过期的消息~~~"+i); // } //测试消费端拒收消息,拒收的消息也不返回发送的队列,就会变成死信消息,就交给死信交换机处理 rabbitTemplate.convertAndSend("exchange_Normal_DLX","test.dlx.xf","我发送了消费端出错不消费的消息~~~"); }

4.3 编写消费端的监听队列类 DlxListener 与 TestDlxListener

DlxListener

package com.fs.Queuelistener; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; /* 监听死信队列中的消息 */ @Component public class DlxListener { //监听死信队列 @RabbitListener(queues = "queue_dlx") public void testDlx(Message message, Channel channel) throws IOException { //得到消息唯一标识 long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //消费死信队列中的消息 System.out.println(new String(message.getBody())); //手动关闭 channel.basicAck(deliveryTag,true); }catch (Exception e){ //上面代码逻辑出现错误 e.printStackTrace(); //拒绝接收,从新发送 channel.basicNack(deliveryTag,true,true); } } }

TestDlxListener

package com.fs.Queuelistener; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; import java.io.IOException; /* 监听正常队列的消息,然后异常拒收,也不返回给发送队列,使消息成为死信消息,交给死信交换机 */ @Component public class TestDlxListener { @RabbitListener(queues = "queue_Normal_DLX") public void testDlxListener(Message message, Channel channel) throws IOException { //得到笑嘻嘻唯一标识 long deliveryTag = message.getMessageProperties().getDeliveryTag(); try{ //消费消息 //模拟消费出错 int i = 1/0; //手动提交 channel.basicAck(deliveryTag,true); }catch (Exception e){ System.out.println("消息消费出现异常,拒绝签收"); //拒绝接收消息 //出错后将消息丢弃,不返回给发送队列 拒绝签收，不重回队列 requeue=false channel.basicNack(deliveryTag,true,false); } } }

4.4 测试

启动消费端的主启动

4.4.1 测试 过期时间 消息过期后交给死信交换机被消费掉

先将TestDlxListener类中的@Component注释掉 ,将注释掉的测试代码打开,后面的代码注释掉,在run

4.4.2 测试队列 长度限制

将测试长度限制的代码放开,其余代码注释,点击run

4.4.3 消息拒收

模拟业务错误,启动消费端主启动

4.5 死信队列小结

死信交换机和死信队列和普通的没有区别当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列

消息成为死信的三种情况： 3. 队列消息长度到达限制； 4. 消费者拒接消费消息，并且不重回队列； 5. 原队列存在消息过期设置，消息到达超时时间未被消费；

5 延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。 很可惜，在RabbitMQ中并未提供延迟队列功能。 但是可以使用：TTL+死信队列 组合实现延迟队列的效果。

5.1 提供方创建用于测试延迟队列的交换机与队列

package com.fs.rabbitmq.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /* RabbitMQ是没有实现延迟队列的 延迟队列的正常queue是没有消费者的,否则生产的消息会被立马消费掉,就不会交给死信交换机,达不到延迟队列效果 但是我们可以通过 使用TTL 加上(DLX)死信队列组合实现延迟队列的效果 延迟队列： 1. 定义正常交换机（order_exchange）和队列(order_queue) 2. 定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx) 3. 绑定，设置正常队列过期时间为30分钟 */ @Configuration public class RabbitMQDelayQueueConfig { //定义死信交换机（order_exchange_dlx）和队列(order_queue_dlx) @Bean public Exchange orderExchangeDlx(){ return ExchangeBuilder.topicExchange("order_exchange_dlx").build(); } //定义死信队列 @Bean public Queue orderQueueDlx(){ return QueueBuilder.durable("order_queue_dlx").build(); } //将死信交换机与死信队列相互绑定 @Bean public Binding orderBindingDlx(@Qualifier("orderExchangeDlx") Exchange exchange,@Qualifier("orderQueueDlx") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("dlx.order.#").noargs(); } //定义正常交换机（order_exchange）和队列(order_queue) @Bean public Exchange orderExchange(){ return ExchangeBuilder.topicExchange("order_exchange").build(); } //定义队列 @Bean public Queue orderQueue(){ return QueueBuilder.durable("order_queue") .withArgument("x-dead-letter-exchange","order_exchange_dlx")//绑定死信交换机 .withArgument("x-dead-letter-routing-key","dlx.order.xf")//绑定routingKey value路由规则dlx.order.# .withArgument("x-message-ttl",10000)//给这个队列添加过期时间 测试就使用10秒过期时间 .build(); } //将正常交换机与队列相互绑定 @Bean public Binding orderBinding(@Qualifier("orderExchange") Exchange exchange,@Qualifier("orderQueue") Queue queue){ return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs(); } }

5.2 消费端编写监听测试延迟队列的队列

package com.fs.Queuelistener; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.stereotype.Component; /* 测试 延迟队列效果实现 消费死信队列中的消息 */ @Component public class OrderListener implements ChannelAwareMessageListener { /* 监听死信队列的消息,并消费 */ @RabbitListener(queues = "order_queue_dlx") 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); } } }

5.3 编写测试代码,发送消息

/* 发送消息到队列中,消息10秒到期,然后消费端监听死信队列,并消费 */ @Test public void testDelay() throws InterruptedException { //1.发送订单消息。 将来是在订单系统中，下单成功后，发送消息 rabbitTemplate.convertAndSend("order_exchange", "order.msg", "订单信息：id=1,time=2019年8月17日16:41:47"); //2.打印倒计时10秒,模拟消息等待10秒后消息过期后,在消费端消费死信 for (int i = 10; i > 0 ; i--) { System.out.println(i+"..."); Thread.sleep(1000); } } }

5.4 测试

启动消费端主启动

run测试方法

6 日志与监控

RabbitMQ默认日志存放路径： /var/log/rabbitmq/rabbit@xxx.log 日志包含了RabbitMQ的版本号、Erlang的版本号、RabbitMQ服务节点名称、cookie的hash值、 RabbitMQ配置文件地址、内存限制、磁盘限制、默认账户guest的创建以及权限配置等等。

6.1 命令

7 消息追踪

在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况。对于RabbitMQ而言，可能 是因为生产者或消费者与RabbitMQ断开了连接，而它们与RabbitMQ又采用了不同的确认机制；也 有可能是因为交换器与队列之间不同的转发策略；甚至是交换器并没有与任何队列进行绑定，生产者 又不感知或者没有采取相应的措施；另外RabbitMQ本身的集群策略也可能导致消息的丢失。这个时 候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。 在RabbitMQ中可以使用Firehose和rabbitmq_tracing插件功能来实现消息追踪。

7.1 消息追踪-Firehose

firehose的机制是将生产者投递给rabbitmq的消息，rabbitmq投递给消费者的消息按照指定的格式 发送到默认的exchange上。这个默认的exchange的名称为amq.rabbitmq.trace，它是一个topic类 型的exchange。发送到这个exchange上的消息的routing key为 publish.exchangename 和 deliver.queuename。其中exchangename和queuename为实际exchange和queue的名称，分别 对应生产者投递到exchange的消息，和消费者从queue上获取的消息。 注意：打开 trace 会影响消息写入功能，适当打开后请关闭。 rabbitmqctl trace_on：开启Firehose命令 rabbitmqctl trace_off：关闭Firehose命令

7.2 消息追踪-rabbitmq_tracing

rabbitmq_tracing和Firehose在实现上如出一辙，只不过rabbitmq_tracing的方式比Firehose多了一 层GUI的包装，更容易使用和管理。 启用插件：rabbitmq-plugins enable rabbitmq_tracing

8 消息可靠性保障

100%确保消息发送成功

8.1 消息可靠性保障–消息补偿

8.2 消息幂等性保障–乐观锁机制

幂等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。也就是说，其任 意多次执行对资源本身所产生的影响均与一次执行的影响相同。 在MQ中指，消费多条相同的消息，得到与消费该消息一次相同的结果。