11-RabbitMQ高级特性-消息可靠性投递
消息的可靠投递
在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
- confirm 确认模式
- return 退回模式
rabbitmq 整个消息投递的路径为：producer--->rabbitmq broker--->exchange--->queue--->consumer
- 消息从 producer 到 exchange 则会返回一个 confirmCallback 。
- 消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递
案例
1. confirm 确认模式
1.1 工程搭建
创建一个空的 maven 工程 rabbitmq-producer-spring：
1.2. 添加依赖
修改pom.xml文件内容为如下：
代码语言：javascript复制<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.lijw</groupId>
<artifactId>rabbitmq-producer-spring</artifactId>
<version>1.0-SNAPSHOT</version>
<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>
</project>
1.3. 配置整合
- 创建
rabbitmq.properties连接参数等配置文件；
rabbitmq.host=127.0.0.1
rabbitmq.port=5672
rabbitmq.username=libai
rabbitmq.password=libai
rabbitmq.virtual-host=/test
2.创建 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}"/>
<!--定义管理交换机、队列-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>
上面是 rabbitmq 初始化的配置，下面我们来定义 交换机与队列。
1.4 定义交换机与队列
<!--消息可靠性投递（生产端） -->
<!-- 1.定义队列 -->
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm"/>
<!-- 2.定义交换机 -->
<rabbit:direct-exchange name="test_exchange_confirm">
<!-- 3. 绑定交换机与队列 -->
<rabbit:bindings>
<rabbit:binding queue="test_queue_confirm" key="confirm"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>
1.5 开启确认模式
代码语言：javascript复制确认模式开启：ConnectionFactory中开启publisher-confirms="true"
1.6 编写测试方法，尝试发送消息
代码语言：javascript复制在rabbitTemplate定义ConfirmCallBack回调函数
package com.lijw;
import org.junit.Test;
import org.junit.runner.RunWith;
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;
/**
* @author Aron.li
* @date 2022/3/4 20:41
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
// 定义队列
static final String QUEUE_NAME = "test_queue_confirm";
// 定义交换机
static final String EXCHANGE_NAME = "test_exchange_confirm";
// 定义路由键
static final String ROUTING_KEY = "confirm";
/**
* 确认模式：
* 步骤：
* 1. 确认模式开启：ConnectionFactory中开启publisher-confirms="true"
* 2. 在rabbitTemplate定义ConfirmCallBack回调函数
*/
@Test
public void testConfirm() {
// 2. 定义回调
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("confirm方法被执行了");
//通过ack判断是否发送消息成功
if (ack) {
// 发送成功
System.out.println("ack: " ack ", 发送成功消息: " cause);
} else {
// 发送失败
System.out.println("ack: " ack ", 发送失败消息: " cause);
// 错一些失败处理，让消息再次发送。
}
}
});
// 3. 发送消息
rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, "test msg confirm 123.......");
// 4. 阻止程序中断导致 channel 中断，如果 channel 中断，那么 confirm 返回的 ack 都只会是 false
while (true) {
}
}
}
- 首先我们正常执行发送的方法，如下：
- 那么我们故意修改错误的交换机名称，让发送消息失败，如下：
以上就是 confirm 确认模式了。
2. return 退回模式
代码语言：javascript复制回退模式： 当消息发送给Exchange后，Exchange路由到Queue失败是 才会执行 ReturnCallBack
2.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-confirms="true"
publisher-returns="true"
/>
2.2 编写测试方法，验证 returnCallBack 方法
代码语言：javascript复制@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
// 定义队列
static final String QUEUE_NAME = "test_queue_confirm";
// 定义交换机
static final String EXCHANGE_NAME = "test_exchange_confirm";
// 定义路由键
static final String ROUTING_KEY = "confirm";
/**
* 回退模式： 当消息发送给Exchange后，Exchange路由到Queue失败是 才会执行 ReturnCallBack
* 步骤：
* 1. 开启回退模式:publisher-returns="true"
* 2. 设置ReturnCallBack
* 3. 设置Exchange处理消息的模式：
* 3.1. 如果消息没有路由到Queue，则丢弃消息（默认）
* 3.2. 如果消息没有路由到Queue，返回给消息发送方ReturnCallBack
*/
@Test
public void testReturn() {
// 1. 设置Exchange处理消息的模式
rabbitTemplate.setMandatory(true); // 如果不设置，则默认丢弃消息
// 2. 设置Exchange如果消息没有路由到Queue，返回给消息发送方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("returnCallback 执行了...");
System.out.println(message);
System.out.println(replyCode);
System.out.println(replyText);
System.out.println(exchange);
System.out.println(routingKey);
// 错误处理... 后续篇章介绍
}
});
// 3. 发送消息
rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, "test msg returnCallBack 123.......");
System.out.println("发送消息: test msg returnCallBack 123.......");
// 4. 阻止程序中断导致 channel 中断，如果 channel 中断，那么 returnCallBack 将不会执行
while (true) {
}
}
}
- 首先我们正常发送消息，如下：
- 那么下面我们来触发调用 returnCallback 方法， 触发的方式是让消息达到 exchange，但是无法路由到 queue。也就是修改一个错误的 ROUTING_KEY 就可以了。
小结
- 设置ConnectionFactory的publisher-confirms="true" 开启 确认模式。
- 使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。
- 设置ConnectionFactory的publisher-returns="true" 开启 退回模式。
- 使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。
- 在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。 使用channel下列方法，完成事务控制：
- txSelect(), 用于将当前channel设置成transaction模式
- txCommit()，用于提交事务
- txRollback(),用于回滚事务
