RabbitMQ详解（下）
一：序
通过《RabbitMQ详解（上）》一文中，我们可以知道RabbitMQ的一些基本的原生用法，如交换机的创建及消息的投递，但是在企业中我们大部分都是把RabbitMQ集成到SpringBoot中的，所以原生的方式我们则不怎么使用到，下面我将和大家一起走入SpringBoot整合RabbitMQ的世界。
下面全部案例代码：UseSpringBootIntegrateRabbitMQ
二：一个简单的案例
为了不那么麻烦，我就以一个SpringBoot来处理消息的发送和接收；通过Postman发送请求到Controller，再由Controller调用生产者，生产者把消息推送到Brock（交换机）,再由交换机具体路由到指定队列，然后由指定的消费者监听队列获取消息；这就是一个我要实现的完整流程，以直接交换机来举例。
如下是SpringBoot整合RabbitMQ的基本案例，实现直接交换机模式，具体流程如上图：
<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <!--注意：spring-boot-starter-parent版本若达到3.0.0 则只支持JDK17，而2.7.10是支持JDK8的最后一个版本--> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.7.10</version> </parent> <groupId>cn.xw</groupId> <artifactId>HelloWorld</artifactId> <version>0.0.1-SNAPSHOT</version> <name>HelloWorld</name> <description>HelloWorld</description> <dependencies> <!--Spring Boot的核心启动器，包含了自动配置、日志和YAML--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> <!--去掉logback配置，要不然冲突--> <exclusions> <exclusion> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-logging</artifactId> </exclusion> </exclusions> </dependency> <!-- 引入SpringBoot的log4j2依赖启动坐标；这坐标包含具体的log4j2的坐标和连接Slf4j的适配器 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-log4j2</artifactId> </dependency> <!--SpringBootWeb启动依赖坐标--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!--RabbitMQ启动依赖坐标--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <!-- JSON格式化坐标--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>2.0.26</version> </dependency> <!--Lombok坐标导入--> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.26</version> <scope>provided</scope> </dependency> <!-- RabbitMQ测试坐标 --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <version>3.0.3</version> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <!--配置maven编译版本--> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.8.1</version> <configuration> <source>1.8</source><!--源代码使用的JDK--> <target>1.8</target><!--target需要生成的目标class文件的编译版本--> <encoding>UTF-8</encoding><!--字符集编码，防止中文乱码--> <failOnError>true</failOnError><!--指示即使存在编译错误，构建是否仍将继续--> <failOnWarning>false</failOnWarning><!--指示即使存在编译警告，构建是否仍将继续--> <showDeprecation>false</showDeprecation><!--设置是否显示使用不推荐API的源位置--> <showWarnings>false</showWarnings><!--设为true若要显示编译警告，请执行以下操作--> <meminitial>128M</meminitial><!--编译器使用的初始化内存--> <maxmem>512M</maxmem><!--编译器使用的最大内存--> </configuration> </plugin> </plugins> </build> </project>
<?xml version="1.0" encoding="UTF-8" ?> <!--monitorInterval属性值（秒数）为一个非零值来让Log4j每隔指定的秒数来重新读取配置文件，可以用来动态应用Log4j配置--> <Configuration status="info" monitorInterval="30"> <!--用来自定义一些变量--> <Properties> <!--变量定义--> <Property name="myPattern" value="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/> <Property name="dir_url">./logs</Property> </Properties> <!--使用Appenders元素可以将日志事件数据写到各种目标位置--> <Appenders> <!-- 默认打印到控制台 --> <Console name="ConsoleAppend" target="SYSTEM_OUT"> <!-- 默认打印格式 --> <PatternLayout pattern="${myPattern}"/> </Console> <!-- 打印到日志文件上 --> <File name="FileAppend" fileName="${dir_url}/fileLog.log" bufferedIO="true" immediateFlush="true"> <PatternLayout> <pattern>${myPattern}</pattern> </PatternLayout> </File> </Appenders> <!--定义logger，只有定义了logger并引入的appender，appender才会生效--> <Loggers> <!-- 默认打印日志级别为 error --> <Root level="INFO"> <AppenderRef ref="ConsoleAppend"/> <AppenderRef ref="FileAppend"/> </Root> </Loggers> </Configuration>
server:
port: 8081
spring:
## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties
rabbitmq:
host: 49.235.99.193
port: 5672
username: admin
password: 123
virtual-host: test
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 11:49 * 用来测试RabbitMQ的生产者发送消息（对象）到消费者中的一系列传输 */ @Data public class MessageSendDTO implements Serializable { private static final long serialVersionUID = 5905249092659173678L; private Integer msgID; // 消息ID private String msgType; // 消息类型 private Object msgBody; // 消息体 }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 17:55 * RabbitMQ配置类 */ @Configuration public class RabbitMQConfig { //定义1：简单的直接交换机名称；定义2：简单队列名称；定义3：路由key public static final String SIMPLE_DIRECT_EXCHANGE = "simpleDirectExchange"; public static final String SIMPLE_QUEUE_NAME = "simpleQueueName"; public static final String SIMPLE_KEY = "simpleKey"; /*** * 创建交换机信息 * @return Exchange */ @Bean("simpleDirectExchange") //注：Bean对象可以不写名称，默认就是方法名 public Exchange simpleDirectExchange() { //这个ExchangeBuilder就是我们当初使用的如下方式一样： // channel.exchangeDeclare("交换机名称", "交换机类型",true, false, false, null); return ExchangeBuilder.directExchange(SIMPLE_DIRECT_EXCHANGE).durable(true).build(); } /*** * 创建队列信息 * @return Queue */ @Bean("simpleQueueName") public Queue simpleQueueName() { //这个QueueBuilder就是我们当初使用的如下方式一样： // channel.queueDeclare("队列名称", true, false, false, null); return QueueBuilder.durable(SIMPLE_QUEUE_NAME).build(); } /*** * 队列信息绑定到交换机上 * @param simpleDirectExchange 简单的直接交换机 * @param simpleQueueName 简单的队列 * @return Binding */ @Bean("simpleQueueBindSimpleExchange") public Binding simpleQueueBindSimpleExchange(@Qualifier(value = "simpleDirectExchange") Exchange simpleDirectExchange, @Qualifier(value = "simpleQueueName") Queue simpleQueueName) { //这个BindingBuilder就是我们当初使用的如下方式一样： // channel.queueBind("队列名称", "交换机名称", "路由key"); return BindingBuilder.bind(simpleQueueName).to(simpleDirectExchange).with(SIMPLE_KEY).noargs(); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 23:52 */ @Slf4j //使用lombok自带的日志注解，具体实现是slf4j+log4j2 @RestController @RequestMapping("/simple") @RequiredArgsConstructor public class SimpleController { //使用SLF4J来获取Logger对象;(注意导包：import org.slf4j.Logger; import org.slf4j.LoggerFactory;) //Logger logger = LoggerFactory.getLogger(this.getClass()); //注入生产者对象 private final SimpleProducer simpleProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String produce(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //发送消息到生产者 simpleProducer.productionSimpleMessage(msg); return "请求发送成功，并已接收"; } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 23:30 * 生产者 */ @Slf4j @Component @RequiredArgsConstructor public class SimpleProducer { //注入RabbitTemplate对象 private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 生产者发送的消息 */ public void productionSimpleMessage(MessageSendDTO msg) { log.info("生产者接收到消息，并发送到Brock的交换机...."); //消息转换为JSON格式发送，并发送到Brock的交换机 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //convertAndSend("交换机名称","路由key","发送消息内容")，其实和原生的： // channel.basicPublish("交换机名称","路由key","其它参数","消息"); // 使用convertAndSend默认消息是持久化的，如我们当初原生设置的 其它参数：MessageProperties.PERSISTENT_TEXT_PLAIN rabbitTemplate.convertAndSend(RabbitMQConfig.SIMPLE_DIRECT_EXCHANGE, RabbitMQConfig.SIMPLE_KEY, bytes); } } //---------------------------------------------------------------------- /** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 23:29 * 这是一个简单的消费者 */ @Slf4j @Component public class SimpleConsumer { /*** * 简单消息处理（监听） * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.SIMPLE_QUEUE_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割 public void messageSimpleHandle(String msgData, Message message, Channel channel) { //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); log.info("消息由消费者消费：{}，并消费完成", msg); } }
1：整合RabbitMQ的常用配置信息
## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties，以实际版本为主 # base spring.rabbitmq.host: 服务Host spring.rabbitmq.port: 服务端口 spring.rabbitmq.username: 登陆用户名 spring.rabbitmq.password: 登陆密码 spring.rabbitmq.virtual-host: 连接到rabbitMQ的vhost spring.rabbitmq.addresses: 指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host) spring.rabbitmq.requested-heartbeat: 指定心跳超时，单位秒，0为不指定；默认60s spring.rabbitmq.publisher-confirm-type: 是否启用【发布确认】 spring.rabbitmq.publisher-returns: 是否启用【发布返回】 spring.rabbitmq.connection-timeout: 连接超时，单位毫秒，0表示无穷大，不超时 spring.rabbitmq.parsed-addresses: # ssl spring.rabbitmq.ssl.enabled: 是否支持ssl spring.rabbitmq.ssl.key-store: 指定持有SSL certificate的key store的路径 spring.rabbitmq.ssl.key-store-password: 指定访问key store的密码 spring.rabbitmq.ssl.trust-store: 指定持有SSL certificates的Trust store spring.rabbitmq.ssl.trust-store-password: 指定访问trust store的密码 spring.rabbitmq.ssl.algorithm: ssl使用的算法，例如，TLSv1.1 # cache spring.rabbitmq.cache.channel.size: 缓存中保持的channel数量 spring.rabbitmq.cache.channel.checkout-timeout: 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel spring.rabbitmq.cache.connection.size: 缓存的连接数，只有是CONNECTION模式时生效 spring.rabbitmq.cache.connection.mode: 连接工厂缓存模式：CHANNEL 和 CONNECTION # listener spring.rabbitmq.listener.simple.auto-startup: 是否启动时自动启动容器 spring.rabbitmq.listener.simple.acknowledge-mode: 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto spring.rabbitmq.listener.simple.concurrency: 最小的消费者数量 spring.rabbitmq.listener.simple.max-concurrency: 最大的消费者数量 spring.rabbitmq.listener.simple.prefetch: 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量. spring.rabbitmq.listener.simple.transaction-size: 指定一个事务处理的消息数量，最好是小于等于prefetch的数量. spring.rabbitmq.listener.simple.default-requeue-rejected: 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系） spring.rabbitmq.listener.simple.idle-event-interval: 多少长时间发布空闲容器时间，单位毫秒 spring.rabbitmq.listener.simple.retry.enabled: 监听重试是否可用 spring.rabbitmq.listener.simple.retry.max-attempts: 最大重试次数 spring.rabbitmq.listener.simple.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔 spring.rabbitmq.listener.simple.retry.multiplier: 应用于上一重试间隔的乘数 spring.rabbitmq.listener.simple.retry.max-interval: 最大重试时间间隔 spring.rabbitmq.listener.simple.retry.stateless: 重试是有状态or无状态 # template spring.rabbitmq.template.mandatory: 启用强制信息；默认false spring.rabbitmq.template.receive-timeout: receive() 操作的超时时间 spring.rabbitmq.template.reply-timeout: sendAndReceive() 操作的超时时间 spring.rabbitmq.template.retry.enabled: 发送重试是否可用 spring.rabbitmq.template.retry.max-attempts: 最大重试次数 spring.rabbitmq.template.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔 spring.rabbitmq.template.retry.multiplier: 应用于上一重试间隔的乘数 spring.rabbitmq.template.retry.max-interval: 最大重试时间间隔
三：工作队列+消息应答+消息分发
这里的工作队列我就使用生产者发送消息到直接交换机，再由直接交换机通过路由分发到队列中，再由消费者（多个）来消费队列的消息，具体的流程图如下（在下面的每小节完成消息应答和消息分发）：
1：普通的工作队列
因为具体的其它代码（配置文件等）在第一章已经给出了，下面我就主要粘出具体的代码
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 17:43 * RabbitMQ配置类 */ @Configuration public class RabbitMQConfig { //直接交换机名称 public static final String DIRECT_EXCHANGE = "directExchange"; //队列名称A public static final String QUEUE_A_NAME = "queueAName"; //路由key public static final String ROUTE_KEY = "routeKeyName"; /*** * 创建交换机信息 * @return Exchange */ @Bean("directExchange") public Exchange directExchange() { return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).durable(true).build(); } /*** * 创建队列A信息 * @return Queue */ @Bean("queueAName") public Queue queueAName() { return QueueBuilder.durable(QUEUE_A_NAME).build(); } /*** * 队列绑定到交换机上，通过路由key * @param directExchange 交换机信息 * @param queueAName A队列绑定 * @return Binding */ @Bean("directExchangeBindAQueue") public Binding directExchangeBindAQueue(@Qualifier(value = "directExchange") Exchange directExchange, @Qualifier(value = "queueAName") Queue queueAName) { return BindingBuilder.bind(queueAName).to(directExchange).with(ROUTE_KEY).noargs(); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:39 * 测试生产者 */ @Component @RequiredArgsConstructor public class TestProducer { //注入rabbitTemplate对象 private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ROUTE_KEY, bytes); } } //---------------------------------------------------------------- /** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 23:29 * 这是监听队列A的消费者（A、B） */ @Slf4j @Component public class QueueConsumer { /*** * 消费者A（监听）队列queueAName * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割 public void messageSimpleHandleA(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) throws InterruptedException, IOException { //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); //假设消费者A处理消息慢，每8秒处理一条 Thread.sleep(8000); log.info("A：消息由消费者A消费：{}，并消费完成", msg); } /*** * 消费者B（监听）队列queueAName * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}) //只需要监听队列即可,多个则在{}里面逗号分割 public void messageSimpleHandleB(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) throws InterruptedException, IOException { //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); //假设消费者B处理消息快，每2秒处理一条 Thread.sleep(2000); log.info("B：消息由消费者B消费：{}，并消费完成", msg); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:37 */ @Slf4j //使用lombok自带的日志注解，具体实现是slf4j+log4j2 @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //使用SLF4J来获取Logger对象;(注意导包：import org.slf4j.Logger; import org.slf4j.LoggerFactory;) //Logger logger = LoggerFactory.getLogger(this.getClass()); //注入生产者对象 private final TestProducer testProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSend(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //循环发送消息 for (int i = 97; i <= 106; i++) { MessageSendDTO build = MessageSendDTO.builder().msgID(i) .msgType("testType") .msgBody(msg.getMsgBody() + new String(new char[]{(char) i, (char) i})).build(); //发送消息 testProducer.producerSendMsg(build); } return "请求发送成功，并已接收"; } }
我们运行后可以发现，生产者发送10条信息到队列，然后就由2个消费者监听获取队列的信息，其中消费者A每隔8秒消费一条消息，消费者B每隔2秒消费一条消息，所以能者多劳才对，但是恰恰相法，默认使用的是轮询的方式，每个消费者都会消费5条消息；其实这种是不行的，其中消费者B大部分属于空闲时间，应该执行更多信息才对，下面我就来优化代码，使用不公平分发（其实这个就是我在上篇说到的预取值那一节）
2：消息分发（预取值）
具体在上篇的预取值一节介绍是啥了，这里我就这样使用了，我们对上面的代码进行一些简单的处理：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test listener: simple: prefetch: 1 # 设置预取值为1（当为1时也被称为不公平分发）
#消费者每次从队列获取的消息数量 (默认一次250个)
#通过查看后台管理器中queue的unacked数量
3：消息接收应答
消费者处理完消息后向队列发送消息处理完成请求，那么我们就得开启消息接收应答方式；非SpringBoot整合RabbitMQ的默认方式是消息一旦发送给消费者就代表应答了，然后队列就删除已发送的消息了；但是我们不希望这样，因为在处理的过程中出现问题后，那条消息就没了，我们希望消息处理完成后再给队列发送应答成功；这就得修改配置和消费者了：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test listener: simple: acknowledge-mode: manual prefetch: 1 # 消息确认模式： # acknowledge-mode: none # 自动确认，则不需要我们手动确认消息，而是消息一旦发送给消费者就代表完成确认了 # acknowledge-mode: auto # 根据情况确认（默认值），若程序出现异常则不确认，若成功执行完成则确认 # acknowledge-mode: manual # 手动确认，需要我们写确认方法
然后我们需要在消费者上面编写手动确认代码：
@Slf4j @Component public class QueueConsumer { /*** * 消费者A（监听）队列queueAName * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}, ackMode = "MANUAL")//监听队列即可,多个则在{}里面逗号分割 public void messageSimpleHandleA(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 Message message, Channel channel) throws InterruptedException, IOException { //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); //假设消费者A处理消息慢，每8秒处理一条 Thread.sleep(8000); log.info("A：消息由消费者A消费：{}，并消费完成", msg); //手动确认，注：这个deliveryTag可以通过message.getMessageProperties().getDeliveryTag()拿到 channel.basicAck(deliveryTag, false); } /*** * 消费者B（监听）队列queueAName * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_A_NAME}, ackMode = "MANUAL") //监听队列即可,多个则在{}里面逗号分割 public void messageSimpleHandleB(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 Message message, Channel channel) throws InterruptedException, IOException { //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); //假设消费者B处理消息快，每2秒处理一条 Thread.sleep(2000); //模拟判断我是否需要手动确认(若随机不是2则确认消费，否则拒绝，继续交由队列) if (Math.ceil(Math.random() * 4) != 2) { log.info("B：消息由消费者B消费：{}，并消费完成", msg); //手动确认 channel.basicAck(deliveryTag, false); } else { log.info("B：消息由消费者B消费：{}，并消费失败，丢回队列", msg);
// 消息编号我们也可以通过message取出来，不用deliveryTag，在message可以获取更多的信息 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); } } }
我们需要添加ackMode = "MANUAL"，并且编写指定的手动确认代码，消费者B为了可以更好的模拟，我可能会随机执行不确认，并且丢回队列
具体的手动确认的代码和方法请参考我之前上篇的：手动应答
四：扇出交换机(Fanout 发布订阅)
前面我们已经使用过直接交换机了，下面将说明一下如何在SpringBoot里整合RabbitMQ来实现扇出交换机使用，具体的扇出交换机的介绍我在上篇已经详细介绍了，下面直接放出代码：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 14:05 * 扇出交换机配置 */ @Configuration public class RabbitMQConfig { //扇出交换机名称 public static final String EXCHANGE_NAME = "fanoutDemo"; //创建两个消息队列 public static final String QUEUE_A = "queueA"; public static final String QUEUE_B = "queueB"; /*** * 创建交换机信息 * @return Exchange */ @Bean("fanoutDemo") public Exchange fanoutDemo() { return ExchangeBuilder.fanoutExchange(EXCHANGE_NAME).durable(true).build(); } /*** * 创建队列A * @return Queue */ @Bean("queueA") public Queue queueA() { return QueueBuilder.durable(QUEUE_A).build(); } /*** * 创建队列B * @return Queue */ @Bean("queueB") public Queue queueB() { return QueueBuilder.durable(QUEUE_B).build(); } /*** * 队列A绑定到扇出交换机 * @param fanoutDemo 交换机名称 * @param queueA 队列A * @return Binding */ @Bean("fanoutBindQueueA") public Binding fanoutBindQueueA(@Qualifier(value = "fanoutDemo") Exchange fanoutDemo, @Qualifier(value = "queueA") Queue queueA) { return BindingBuilder.bind(queueA).to(fanoutDemo).with("").noargs(); } /*** * 队列B绑定到扇出交换机 * @param fanoutDemo 交换机名称 * @param queueB 队列B * @return Binding */ @Bean("fanoutBindQueueB") public Binding fanoutBindQueueB(@Qualifier(value = "fanoutDemo") Exchange fanoutDemo, @Qualifier(value = "queueB") Queue queueB) { return BindingBuilder.bind(queueB).to(fanoutDemo).with("").noargs(); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 15:01 */ @Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, "", bytes); log.info("生产者发送信息完成，已经交由给交换机....."); } } //====================================== /** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 15:04 * 消费者信息 */ @Slf4j @Component @RequiredArgsConstructor public class QueueConsumer { /*** * 消费者A */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_A}) public void testConsumerA(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) { log.info("接收到队列A信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class)); } /*** * 消费者B */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_B}) public void testConsumerB(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) { //注：若消费失败（报错）会自动手动不确认，并且把消息放到队列中，然后又被这个队列消费，最终死循环 int a = 1 / 0; log.info("接收到队列B信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class)); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:37 */ @Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSend(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //发送消息 testProducer.producerSendMsg(msg); return "请求发送成功，并已接收"; } }
上面的案例我采用了一个生产者发送消息到扇出交换机，再由扇出交换机发布消息到每个绑定过这个扇出交换机的队列，然后再由消费者消费每个队列的消息。
五：主题交换机（Topics 匹配模式）
其实主题交换机就是比之前的交换机灵活，它可以按照匹配的方式路由消息到队列，具体的主题交换机的介绍在上篇已经给出介绍了，在这我只对之前的使用原生方式实现的，再使用SpringBoot整合RabbitMQ来实现一下，按照上篇的图示实现：
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 23:10 * 主题交换机配置类 */ @Configuration public class RabbitMQConfig { //交换机名称 public static final String TOPIC_EXCHANGE = "TopicExchange"; //队列Q1名称 public static final String Q1 = "Q1Queue"; //队列Q2名称 public static final String Q2 = "Q2Queue"; //路由绑定关系 Routing Key public static final String Q1_KEY = "*.orange.*"; //路由绑定关系 Routing Key 1 public static final String Q2_KEY_A = "*.*.rabbit"; //路由绑定关系 Routing Key 2 public static final String Q2_KEY_B = "lazy.#"; /*** * 主题交换机 * @return Exchange */ @Bean("topicExchange") public Exchange topicExchange() { return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE).durable(true).build(); } /*** * 队列1信息 * @return Queue */ @Bean("q1Queue") public Queue q1Queue() { return QueueBuilder.durable(Q1).build(); } /*** * 队列2信息 * @return Queue */ @Bean("q2Queue") public Queue q2Queue() { return QueueBuilder.durable(Q2).build(); } /*** * 绑定关系，Q1Queue队列绑定的匹配路由为*.orange.* * @param topicExchange 交换机 * @param q1Queue 队列1 * @return Binding */ @Bean("bindingA") public Binding bindingA(@Qualifier("topicExchange") Exchange topicExchange, @Qualifier("q1Queue") Queue q1Queue) { return BindingBuilder.bind(q1Queue).to(topicExchange).with(Q1_KEY).noargs(); } /*** * 绑定关系，Q2Queue队列绑定的匹配路由为*.*.rabbit * @param topicExchange 交换机 * @param q2Queue 队列2 * @return Binding */ @Bean("bindingB1") public Binding bindingB1(@Qualifier("topicExchange") Exchange topicExchange, @Qualifier("q2Queue") Queue q2Queue) { return BindingBuilder.bind(q2Queue).to(topicExchange).with(Q2_KEY_A).noargs(); } /*** * 绑定关系，Q2Queue队列绑定的匹配路由为lazy.# * @param topicExchange 交换机 * @param q2Queue 队列2 * @return Binding */ @Bean("bindingB2") public Binding bindingB2(@Qualifier("topicExchange") Exchange topicExchange, @Qualifier("q2Queue") Queue q2Queue) { return BindingBuilder.bind(q2Queue).to(topicExchange).with(Q2_KEY_B).noargs(); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 15:01 */ @Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 */ public void producerSendMsg() { //消息任务准备 HashMap<String, String> sendMsg = new HashMap<>(); sendMsg.put("quick.orange.rabbit", "被队列 Q1 Q2 接收到"); sendMsg.put("lazy.orange.elephant", "被队列 Q1 Q2 接收到"); sendMsg.put("quick.orange.fox", "被队列 Q1 接收到"); sendMsg.put("lazy.brown.fox", "被队列 Q2 接收到"); sendMsg.put("lazy.pink.rabbit", "虽然满足两个绑定规则但两个规则都是在Q2队列，所有只要Q2接收一次"); sendMsg.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃"); sendMsg.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃"); sendMsg.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2"); //循环发送消息任务 for (Map.Entry<String, String> msg : sendMsg.entrySet()) { String routKey = msg.getKey(); //主题路由key String message = msg.getValue();//消息任务 //创建对象 MessageSendDTO build = MessageSendDTO.builder().msgBody("基本信息：" + message + " 路由信息：" + routKey).build(); //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(build).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, routKey, bytes); } log.info("生产者发送信息完成，已经交由给交换机....."); } } //+++++++++++++++++++++++++++++++++++++++ /** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 15:04 * 消费者信息 */ @Slf4j @Component @RequiredArgsConstructor public class QueueConsumer { /*** * 消费者1 */ @RabbitListener(queues = {RabbitMQConfig.Q1}) public void testConsumerA(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) { log.info("接收到队列1信息；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class)); } /*** * 消费者2 */ @RabbitListener(queues = {RabbitMQConfig.Q2}) public void testConsumerB(@Payload String msgData, //这个是生产者发送的JSON消息 Message message, Channel channel) { log.info("接收到队列2信息......；信息为：{}", JSONObject.parseObject(msgData, MessageSendDTO.class)); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:37 */ @Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @return String */ @PostMapping("/produce") public String msgSend() { log.info("Controller接收到请求并把请求的信息交由生产者"); //发送消息 testProducer.producerSendMsg(); return "请求发送成功，并已接收"; } }
六：死信队列（重要）
在上篇我们详细介绍了死信队列，并使用最原生的方式实现死信队列，这里我将在SpringBoot里面整合RabbitMQ来实现死信队列，具体的开之前的接介绍，下面我只编写具体的代码，部分代码在上面已经给出：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test listener: simple: acknowledge-mode: manual prefetch: 1
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-17 10:56 */ @Configuration public class RabbitMQConfig { //直接交换机名称 public static final String EXCHANGE_NAME = "MsgHandleExchange"; //队列名称 public static final String QUEUE_NAME = "MsgHandleQueue"; //路由key public static final String ROUTING_KEY = "MsgHandleKey"; //声明死信交换机名称 public static final String DLX_EXCHANGE = "DLXExchange"; //声明死信队列名称 public static final String DLX_QUEUE = "DLXQueue"; //声明路由绑定关系 Routing Key 死信交换机到死信队列 public static final String DLX_KEY = "DLXKey"; //+++++++++++++++++配置了直连交换机和队列的关系 /*** * 一个普通的直连交换机 * @return Exchange */ @Bean("msgHandleExchange") public Exchange msgHandleExchange() { //第一种方式：使用new的方式，是什么类型交换机我们就创建什么xxxExchange //参数1：exchange: 交换机名称 //参数2：durable: 是否需要持久化 //参数3：autoDelete: 当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange //参数4：arguments: 扩展参数，用于扩展AMQP协议定制化使用 //Exchange directExchange = new DirectExchange(EXCHANGE_NAME,true,false); //当前Exchange是否用于RabbitMQ内部使用，默认为False //使用默认即可directExchange.isInternal(); //第二种方式：使用Builder方式 return ExchangeBuilder.directExchange(EXCHANGE_NAME).durable(true).build(); } /*** * 一个普通的队列 * @return Queue */ @Bean("msgHandleQueue") public Queue msgHandleQueue() { //第一种方式：使用new的方式；这种和原生创建一样 //参数一：队列名称 //参数二：队列里的消息是否持久化，默认消息保存在内存中，默认false //参数三：该队列是否只供一个消费者进行消费的独占队列，则为 true（仅限于此连接），false（默认，可以多个消费者消费） //参数四：是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除 true 自动删除，默认false //参数五：构建队列的其它属性，看下面扩展参数 //Queue queue = new Queue(QUEUE_NAME,true,false,false,null); //原生： channel.queueDeclare(QUEUE_NAME, true, false, false, null); //第二种方式：使用Builder方式 //~~~~~~~~~~~~~~~~~~~~~~~~~~设置死信参数Start //绑定死信队列（参数设置） Map<String, Object> arguments = new HashMap<>(); //正常队列设置死信交换机 参数key是固定值；（就是说死去的消息发送到哪个交换机） arguments.put("x-dead-letter-exchange", DLX_EXCHANGE); //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值（就是说死去的消息在交换机里通过什么路由发送到死信队列） arguments.put("x-dead-letter-routing-key", DLX_KEY); //设置正常队列的长度限制 为3 //arguments.put("x-max-length",3); //队列设置消息过期时间 60 秒 //arguments.put("x-message-ttl", 60 * 1000); //~~~~~~~~~~~~~~~~~~~~~~~~~~设置死信参数End return QueueBuilder.durable(QUEUE_NAME).withArguments(arguments).build(); } /*** * 队列绑定到交换机 * @param msgHandleExchange 交换机名称 * @param msgHandleQueue 队列名称 * @return Binding */ @Bean("queueBindDirectExchange") public Binding queueBindDirectExchange(@Qualifier("msgHandleExchange") Exchange msgHandleExchange, @Qualifier("msgHandleQueue") Queue msgHandleQueue) { return BindingBuilder.bind(msgHandleQueue).to(msgHandleExchange).with(ROUTING_KEY).noargs(); } //+++++++++++++++++配置死信交换机的一系列信息 /*** * 死信交换机 * @return Exchange */ @Bean("DLXExchange") public Exchange dLXExchange() { return ExchangeBuilder.directExchange(DLX_EXCHANGE).durable(true).build(); } /*** * 死信队列 * @return Queue */ @Bean("DLXQueue") public Queue dLXQueue() { return QueueBuilder.durable(DLX_QUEUE).build(); } /*** * 死信交换机上面绑定死信队列 * @return Binding */ @Bean("dlxQueueBindDlxExchange") public Binding dlxQueueBindDlxExchange(@Qualifier("DLXExchange") Exchange dLXExchange, @Qualifier("DLXQueue") Queue dLXQueue) { //死信交换机上面绑定死信队列 return BindingBuilder.bind(dLXQueue).to(dLXExchange).with(DLX_KEY).noargs(); } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:37 */ @Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSend(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //发送消息 testProducer.producerSendMsg(msg); return "请求发送成功，并已接收"; } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-17 14:01 * 死信消费者 */ @Slf4j @Component public class DLXConsumer { /*** * 死信消费者 */ @RabbitListener(queues = {RabbitMQConfig.DLX_QUEUE}, ackMode = "MANUAL") public void dlxConsumerTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //死信队列名称 String consumerQueue = message.getMessageProperties().getConsumerQueue(); //死信交换机名称 String receivedExchange = message.getMessageProperties().getReceivedExchange(); //路由key String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey(); log.info("死信消费者从死信队列：{} 获取死信消息：{}，并处理完成手动确认", consumerQueue, messageSendDTO); channel.basicAck(deliveryTag, false); } } //--------------------------------------------- /** * @author AnHui OuYang * @version 1.0 * created at 2023-04-16 15:04 * 消费者信息 */ @Slf4j @Component @RequiredArgsConstructor public class QueueConsumer { /*** * 消费者 * @param msgData 具体的消息 * @param deliveryTag 处理消息的编号 * @param routingKey 当前的路由key * @param message message对象 * @param channel 信道对象 */ @RabbitListener(queues = {RabbitMQConfig.QUEUE_NAME}, ackMode = "MANUAL") // MANUAL必须大写 public void testConsumerA(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收JSON的数据转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //模拟判断我是否需要手动确认(若随机不是2则确认消费，否则拒绝，交由死信队列) if (Math.ceil(Math.random() * 4) != 2) { log.info("处理完成接收到的队列信息为：{}，从{}路由过来的数据", messageSendDTO, routingKey); //手动确认 channel.basicAck(deliveryTag, false); } else { log.info("未处理完成接收到的队列信息为：{}，从{}路由过来的数据", messageSendDTO, routingKey); channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); } } }
@Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, bytes); log.info("生产者发送信息完成，已经交由给交换机....."); } }
七：延迟队列（基于死信）
1：在队列上设置TTL
针对延迟队列我上篇已经有了基本介绍，其实延迟队列就是基于TTL过期时间来完成的，把消息推送到普通的队列里并不会被消费，而是等待这条消息的TTL时间到期才会被丢弃到死信队列中，其实那个具体的死信队列才是我们将来要真实要处理的消息，消息TTL过期到死信队列，最终有死信消费者完成最终的消费；说白了就是借助普通队列的延迟时间达到延迟消费。
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test listener: simple: acknowledge-mode: manual prefetch: 1
@Configuration public class RabbitMQConfig { //延迟直接交换机 public static final String DELAY_DIRECT_EXCHANGE = "delayDirectExchange"; //延迟队列 public static final String DELAY_TTL_QUEUE = "delayTTLQueue"; //延迟队列连接延迟交换机路由key public static final String DELAY_ROUTING_KEY = "delayRoutingKey"; //死信交换机 public static final String DEAD_EXCHANGE = "deadExchange"; //死信队列 public static final String DEAD_QUEUE = "deadQueue"; //死信队列绑定死信交换机路由key public static final String DEAD_ROUTING_KEY = "deadRoutingKey"; //编写延迟队列和延迟交换机一些列配置 /*** * 延迟交换机 * @return Exchange */ @Bean("delayDirectExchange") public Exchange delayDirectExchange() { return ExchangeBuilder.directExchange(DELAY_DIRECT_EXCHANGE).durable(true).build(); } /*** * 延迟普通队列 * @return Queue */ @Bean("delayTTLQueue") public Queue delayTTLQueue() { //绑定死信队列（参数设置） Map<String, Object> arguments = new HashMap<>(); //正常队列设置死信交换机 参数key是固定值；（就是说死去的消息发送到哪个交换机） arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE); //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值（就是说死去的消息在交换机里通过什么路由发送到死信队列） arguments.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY); //设置正常队列的长度限制 为3 //arguments.put("x-max-length",3); //队列设置消息过期时间 20 秒 arguments.put("x-message-ttl", 20 * 1000); return QueueBuilder.durable(DELAY_TTL_QUEUE).withArguments(arguments).build(); } /*** * 延迟队列绑定到延迟交换机上 * @param delayDirectExchange 延迟交换机 * @param delayTTLQueue 延迟队列 * @return Binding */ @Bean("delayQueueBindDelayExchange") public Binding delayQueueBindDelayExchange(@Qualifier("delayDirectExchange") Exchange delayDirectExchange, @Qualifier("delayTTLQueue") Queue delayTTLQueue) { return BindingBuilder.bind(delayTTLQueue).to(delayDirectExchange).with(DELAY_ROUTING_KEY).noargs(); } //编写死信队列和死信交换机，和它们绑定关系的配置 /*** * 死信交换机 * @return Exchange */ @Bean("deadExchange") public Exchange deadExchange() { return ExchangeBuilder.directExchange(DEAD_EXCHANGE).durable(true).build(); } /*** * 死信队列 * @return Queue */ @Bean("deadQueue") public Queue deadQueue() { return QueueBuilder.durable(DEAD_QUEUE).build(); } /*** * 死信队列绑定死信交换机 * @param deadExchange 死信交换机 * @param deadQueue 死信队列 * @return Binding */ @Bean("deadQueueBindDeadExchange") public Binding deadQueueBindDeadExchange(@Qualifier("deadExchange") Exchange deadExchange, @Qualifier("deadQueue") Queue deadQueue) { return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs(); } }
@Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_DIRECT_EXCHANGE,RabbitMQConfig.DELAY_ROUTING_KEY,bytes); log.info("生产者发送信息完成，已经交由给延迟直接交换机....."); } } //========================== @Slf4j @Component public class DeadConsumer { /*** * 死信消费者 */ @RabbitListener(queues = {RabbitMQConfig.DEAD_QUEUE}, ackMode = "MANUAL") public void dlxConsumerTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //死信队列名称 String consumerQueue = message.getMessageProperties().getConsumerQueue(); //死信交换机名称 String receivedExchange = message.getMessageProperties().getReceivedExchange(); //路由key String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey(); log.info("死信消费者从死信队列：{} 获取死信消息：{}，并处理完成手动确认", consumerQueue, messageSendDTO); channel.basicAck(deliveryTag, false); } }
@Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的get请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSend(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //发送消息 testProducer.producerSendMsg(msg); return "请求发送成功，并已接收"; } }
通过上面的案例我们可以实现延迟的效果，把消息发送到延迟队列中，并不会对这部分消息进行消费，而是等待过期后自带丢到死信交换机中，并由死信交换机转发到死信队列中，并进行消费者消费；之前案例我们设置了一个延迟20秒的延迟队列，但是现在若有一个需求，对一些数据进行10秒的延迟，这时我们就需要手动去创建一个队列，并设置延迟时间为10秒；这时我们就发现并不灵活，每次延迟不同时间都需要添加一个延迟队列，所以在队列级别上设置延迟是不灵活的，所以不推荐；下面将进行一个优化。
2：对每条消息设置延迟TTL
上面说了对队列设置TTL延迟来实现延迟消息不是太灵活，所以下面将对之前的代码进行一个简单的优化，创建一个全新的没有延迟的队列，只是对发送的每条消息进行一个延迟（在生产者设置消息延迟并发送），这样就可以实现一个队列里面的每条消息的延迟时间不一样，说干就干，参考下图：
原本代码不变，只是对其部分类进行了优化，添加部分代码：
第一步：在RabbitMQConfig添加这部分代码： //简单队列，无延迟 public static final String SIMPLE_QUEUE = "simpleQueue"; //简单队列连接延迟交换机路由key public static final String SIMPLE_ROUTING_KEY = "simpleRoutingKey"; /*** * 一个简单无延迟的队列 * @return Queue */ @Bean("simpleQueue") public Queue simpleQueue() { //绑定死信队列（参数设置） Map<String, Object> arguments = new HashMap<>(); //正常队列设置死信交换机 参数key是固定值；（就是说死去的消息发送到哪个交换机） arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE); //正常队列设置死信交换机到死信队列绑定Routing Key 参数key是固定值（就是说死去的消息在交换机里通过什么路由发送到死信队列） arguments.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY); //设置正常队列的长度限制 为3 //arguments.put("x-max-length",3); //队列设置消息过期时间 20 秒 //arguments.put("x-message-ttl", 20 * 1000); return QueueBuilder.durable(SIMPLE_QUEUE).withArguments(arguments).build(); } /*** * 简单队列绑定到延迟交换机上 * @param delayDirectExchange 延迟交换机 * @param simpleQueue 简单队列 * @return Binding */ @Bean("simpleQueueBindDelayExchange") public Binding simpleQueueBindDelayExchange(@Qualifier("delayDirectExchange") Exchange delayDirectExchange, @Qualifier("simpleQueue") Queue simpleQueue) { return BindingBuilder.bind(simpleQueue).to(delayDirectExchange).with(SIMPLE_ROUTING_KEY).noargs(); } 第二步：在TestController添加这一个资源请求： /*** * 基本的POST请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @param ttl 过期时间 * @return String */ @PostMapping("/produceA/{ttl}") public String msgSendSimple(@RequestBody MessageSendDTO msg, @PathVariable(value = "ttl") Integer ttl) { log.info("Controller接收到请求并把请求的信息交由生产者：{}，其中消息的过期时间为：{} s", msg, ttl); //发送消息 testProducer.producerSendMsgSimple(msg, ttl); return "请求发送成功，并已接收"; } 第三步：在TestProducer生产者类添加生产者方法： /*** * 生产者方法 * @param msg 消息 * @param ttl 过期时间 */ public void producerSendMsgSimple(MessageSendDTO msg, Integer ttl) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.DELAY_DIRECT_EXCHANGE, RabbitMQConfig.SIMPLE_ROUTING_KEY, bytes, message -> { //这条消息的过期时间也被设置成了ttl秒 , 超过ttl秒未处理则执行到此消息后被丢弃（记住是执行到此消息后被丢弃，后面说明） message.getMessageProperties().setExpiration(String.valueOf(ttl * 1000)); //设置好了一定要返回 return message; }); }
得出结论：上面的程序不是特别完美（有问题），虽然设置消息过期时间，但是在队列中，不管后进的延迟消息多短，都得等前面的延迟消息过期或消费，否则后面的延迟消息会一直等待，即使延迟消息过了也会等待（最终等到这条消息执行时，会先判断是否过期，没过期继续等待，阻塞后面的消息，过期或者被消费则下一个执行）；可以看看队列先进先出的原则。
八：延迟队列（基于插件）
上面基于死信队列实现延迟存在问题，在这我将使用插件的方式来解决上面遗留的问题，这种方式得需要我们下载插件，安装插件后，交换机就会多出来一个”x-delayed-message“的类型，我们创建时选择这种类型；
我们需要下载插件：rabbitmq_delayed_message_exchange-3.11.1.ez 具体的版本可以测试，我用的当前时间最新的。
安装方式：把下载的插件复制到RabbitMQ的plugins目录里：
如我这具体路径是：/usr/local/rabbitmq_server-3.11.13/plugins；复制完成后就可以进行安装操作，
执行：rabbitmq-plugins enable rabbitmq_delayed_message_exchange；不用指定版本号。
直接执行成功会打印：....started 1 plugins.(代码安装成功一个插件)
现在我们安装完插件后想实现延迟消息则不用那么麻烦了，也不用写死信交换机了，只需要正常的队列创建和交换机创建即可，下面是基本的流程图：
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-17 17:10 * RabbitMQ配置类 */ @Configuration public class RabbitMQConfig { //直接延迟交换机名称 public static final String DELAYED_EXCHANGE = "delayedExchange"; //延迟队列名称（但不是在队列中延迟） public static final String DELAYED_QUEUE = "delayedQueue"; //绑定路由key public static final String DELAYED_ROUTING_KEY = "delayedRoutingKey"; /*** * 创建交换机消息 * @return Exchange */ @Bean("delayedExchange") public CustomExchange delayedExchange() { //因为通过ExchangeBuilder没有那个延迟交换机的类型，所以我们使用其它交换机 //其它参数 Map<String, Object> args = new HashMap<>(); //自定义交换机的类型；（虽然设置的是延迟交换机，但是具体四大类型还是得有） args.put("x-delayed-type", "direct"); //参数：交换机名称、交换机类型、是否持久化交换机、是否断开自动删除、其它参数 return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, args); } /*** * 队列名称 * @return Queue */ @Bean("delayedQueue") public Queue delayedQueue() { return QueueBuilder.durable(DELAYED_QUEUE).build(); } /*** * 绑定关系 * @param delayedExchange 交换机消息 * @param delayedQueue 队列消息 * @return Binding */ @Bean("delayedQueueBindDelayedExchange") public Binding delayedQueueBindDelayedExchange(@Qualifier(value = "delayedExchange") CustomExchange delayedExchange, @Qualifier(value = "delayedQueue") Queue delayedQueue) { return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs(); } }
@Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 * @param delayTime 延迟时间 */ public void producerSendMsgDelay(MessageSendDTO msg, Integer delayTime) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.DELAYED_EXCHANGE, RabbitMQConfig.DELAYED_ROUTING_KEY, bytes, message -> { //这条消息的过期时间被设置过期delayTime秒（在交换机中被延迟，时间一到则被路由到队列） message.getMessageProperties().setDelay(delayTime * 1000); //设置好了一定要返回 return message; }); } } // -------------------------------------------------------------- @Slf4j @Component public class ConsumerA { /*** * 消费者 */ @RabbitListener(queues = {RabbitMQConfig.DELAYED_QUEUE}, ackMode = "MANUAL") public void dlxConsumerTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //队列名称 String consumerQueue = message.getMessageProperties().getConsumerQueue(); //交换机名称 String receivedExchange = message.getMessageProperties().getReceivedExchange(); //路由key String receivedRoutingKey = message.getMessageProperties().getReceivedRoutingKey(); log.info("消费者从队列：{} 获取消息：{}，并处理完成手动确认", consumerQueue, messageSendDTO); channel.basicAck(deliveryTag, false); } }
@Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的POST请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @param delayTime 延迟时间 * @return String */ @PostMapping("/produceMsg/{delayTime}") public String msgSendSimple(@RequestBody MessageSendDTO msg, @PathVariable(value = "delayTime") Integer delayTime) { log.info("Controller接收到请求并把请求的信息交由生产者：{}，其中消息的过期时间为：{} s", msg, delayTime); //发送消息 testProducer.producerSendMsgDelay(msg, delayTime); return "请求发送成功，并已接收"; } }
九：消息发布确认
之前介绍了消息应答，它保证了消费者和队列之间的关系达到消息不丢失；但是现在要如何保证生产者投递消息到交换机以及交换机到队列的数据不丢失呢？这我们就使用到了消息的发布确认，看过上篇的人会知道，之前我使用原生方式实现了单个发布确认、批量发布确认和异步批量确认，但现在我要以SpringBoot整合RabbitMQ的方式来完成消息的发布确认。
1：一个简单的代码准备
这里我先编写一个没有实现消息的发布确认代码，针对这个代码后面做一些列的优化：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test listener: simple: acknowledge-mode: manual prefetch: 1
@Configuration public class RabbitMQConfig { //直接交换机 public static final String CONFIRM_EXCHANGE = "confirmExchange"; //队列 public static final String CONFIRM_QUEUE = "confirmQueue"; //绑定路由key public static final String CONFIRM_ROUTING_KEY = "confirmRoutingKey"; /*** * 创建交换机消息 * @return Exchange */ @Bean("confirmExchange") public Exchange confirmExchange() { return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true).build(); } /*** * 队列名称 * @return Queue */ @Bean("confirmQueue") public Queue confirmQueue() { return QueueBuilder.durable(CONFIRM_QUEUE).build(); } /*** * 绑定关系 * @param confirmExchange 交换机消息 * @param confirmQueue 队列消息 * @return Binding */ @Bean("delayedQueueBindDelayedExchange") public Binding delayedQueueBindDelayedExchange(@Qualifier(value = "confirmExchange") Exchange confirmExchange, @Qualifier(value = "confirmQueue") Queue confirmQueue) { return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY).noargs(); } }
@Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsgDelay(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE, RabbitMQConfig.CONFIRM_ROUTING_KEY, bytes); } } //======================================= @Slf4j @Component public class ConsumerA { /*** * 消费者 */ @RabbitListener(queues = {RabbitMQConfig.CONFIRM_QUEUE}, ackMode = "MANUAL") public void dlxConsumerTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //队列名称 String consumerQueue = message.getMessageProperties().getConsumerQueue(); log.info("消费者从队列：{} 获取消息：{}，并处理完成手动确认", consumerQueue, messageSendDTO); channel.basicAck(deliveryTag, false); } }
@Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的POST请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSendSimple(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者：{}", msg); //发送消息 testProducer.producerSendMsgDelay(msg); return "请求发送成功，并已接收"; } }
这里我们就完成了一个普通的消息发送到消费者消费代码了，其实也就是文章开头的简单的Demo，不过我们需要对其优化。
2：发布确认
这里说的是生产者发送消息到交换机失败的情况下回调：
Ⅰ：我们得编写一个RabbitMQMyCallBack的回调类，具体如下： @Slf4j @Component @RequiredArgsConstructor public class RabbitMQMyCallBack implements RabbitTemplate.ConfirmCallback { //注入rabbitTemplate对象 private final RabbitTemplate rabbitTemplate; /*** * 对象实例化完成（对象创建和属性注入）后调用此方法 */ @PostConstruct public void init() {
//必须设置 rabbitTemplate.setMandatory(true); //设置发布确认信息RabbitTemplate.ConfirmCallback confirmCallback; rabbitTemplate.setConfirmCallback(this); } /*** * 是ConfirmCallback的抽象方法，用来确认消息是否到达exchange(交换器)，不保证消息是否可以路由到正确的queue; * 它的抽象方法机制只确认，但需要配置部分参数： * publisher-confirm-type: correlated * 对于部分Springboot老版本需要设置：publisher-confirms: true * 交换机确认回调方法（成功和失败）： * 发送消息 --> 交换机接收到消息 --> 回调 * 1：correlationData 保存回调消息的ID及相关信息 * 2：交换机收到消息 ack = true * 3：调用回调confirm方法，对应ack=true , cause=null * 发送消息 --> 交换机接收消息失败 --> 回调 * 1：correlationData 保存回调消息的ID及相关信息 * 2：交换机收到消息 ack = false * 3：调用回调confirm方法，对应ack=false , cause="异常信息" * @param correlationData 回调的相关数据 * @param ack 消息是否成功发送给交换机，true成功，false失败 * @param cause 对于ack为false时会有对应的失败原因，否则为空 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { //获取对应的ID消息，因为不确认是否有ID被传入，所以取值需要判空 String id = correlationData == null ? "" : correlationData.getId(); //校验是否成功发送 if (ack) { log.info("消息已经成功交给了交换机，对应消息ID为：{}", id); } else { log.info("消息未能成功发送给交换机，对应消息ID为：{}，异常原因：{}", id, cause); } } } Ⅱ：编写完成过后，我们需要在发送者TestProducer来设置一些消息，用来回调用处： @Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsgDelay(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //其它的一些信息，用来回调用处 CorrelationData correlationData = new CorrelationData(); //设置id信息，其实默认就是UUID，我们其实可以根据自己设置指定ID信息 //correlationData.setId(String.valueOf(UUID.randomUUID())); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE, RabbitMQConfig.CONFIRM_ROUTING_KEY, bytes, correlationData); } } Ⅲ：还差最后一步了，设置配置文件(一定要加)： spring.rabbitmq.publisher-confirm-type: correlated 取值有： none：禁用发布确认模式，是默认值 correlated：发布消息成功到交换器后会触发回调方法（推荐） simple：存在两种效果，第一种和correlated效果一样，但是也可以使用rabbitTemplate调用waitForConfirms或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker
测试调用失败和成功现状（故意写错发送到不存在的交换机）：
生产者一次性发送2个消息：
//发送消息A（可以成功发送） rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE, RabbitMQConfig.CONFIRM_ROUTING_KEY, bytes, correlationData); //发送消息B（不可以成功发送,交换机不存在） rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE+"test", RabbitMQConfig.CONFIRM_ROUTING_KEY, bytes, correlationData);
3：回退消息
这里说的是交换机路由发送到队列时，队列不存在则消息默认就没了，但是我们设置回退消息时，路由到的队列不存在则会执行个回调方法：
Mandatory参数：在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。
Ⅰ：我们得编写一个RabbitMQMyCallBack的回调类，具体如下： @Slf4j @Component @RequiredArgsConstructor public class RabbitMQMyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback { //注入rabbitTemplate对象 private final RabbitTemplate rabbitTemplate; /*** * 对象实例化完成（对象创建和属性注入）后调用此方法 */ @PostConstruct public void init() { //设置发布确认信息回调类RabbitTemplate.ConfirmCallback confirmCallback; rabbitTemplate.setConfirmCallback(this); //设置回退消息回调类ReturnsCallback.ReturnsCallback returnsCallback; rabbitTemplate.setReturnsCallback(this); //true：交换机无法将消息进行路由时，会将该消息返回给生产者；false：如果发现消息无法进行路由，则直接丢弃 rabbitTemplate.setMandatory(true); // 或使用配置 spring.rabbitmq.template.mandatory: true } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) {...} /*** * 当消息无法被路由时执行当前回调 * @param returned 被退回的消息信息 */ @Override public void returnedMessage(ReturnedMessage returned) { // 发送的消息 Message message = returned.getMessage(); // 发送到哪个交换机 String exchange = returned.getExchange(); // 交换机到队列的路由key String routingKey = returned.getRoutingKey(); // 退回原因 String replyText = returned.getReplyText(); // 退回原因状态码 int replyCode = returned.getReplyCode(); //消息打印 log.info("信息被回退，从交换机：{} 路由：{} 发送到队列失败，发送信息为：{}，退回状态码：{} 和原因：{}", exchange, routingKey, message, replyCode, replyText); //我们可以在这后面对发送失败的消息进行处理 } } Ⅱ：编写完成过后，我们需要在发送者TestProducer来设置一些消息，用来回调用处： @Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsgDelay(MessageSendDTO msg) { //省略.... //发送消息一次（不可以成功发送,路由key不存在） rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE, RabbitMQConfig.CONFIRM_ROUTING_KEY + "test", bytes, correlationData); } } 还差最后一步了，设置配置文件加红的(这也是一个比较全的配置了)： server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test # 开启消息确认模式 publisher-confirm-type: correlated # 消息发布确认（生产者->交换机是否成功） publisher-returns: true # 消息回退（交换机路由->队列是否成功） template: mandatory: true # 消息路由发送失败返回到队列中, 相当手动设置 rabbitTemplate.setMandatory(true); # 主要针对消费者的一些设置 listener: simple: acknowledge-mode: manual # 消息应答ACK方式 prefetch: 1 # 限制每次发送1条数据到队列（只可堆积1条到消费者）（预取值） concurrency: 1 # 最少需要一个消费者来监听同一队列 max-concurrency: 2 # 最大只能拥有2个消费者来监听同一队列 default-requeue-rejected: true # 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
到这里的发布确认已经完成了
十：备份交换机
有了mandatory参数和回退消息，我们就可以对那些无法被投递的消息有着回调功能，这样就可以对无法投递的消息进行处理；但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是在多台服务器上都存在生产者的时候，我们就需要手动去每台服务器上查看生产的日志文件，分析问题，其实这样很容易看花眼，而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。
在RabbitMQ中，有一种备份交换机的机制存在，可以很好的应对这个问题。备份交换机可以理解为 RabbitMQ 中交换机的“备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为Fanout，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
基本的流程图：
server: port: 8081 spring: ## 具体RabbitMQ配置请参考：org.springframework.boot.autoconfigure.amqp.RabbitProperties rabbitmq: host: 49.235.99.193 port: 5672 username: admin password: 123 virtual-host: test # 开启消息确认模式 publisher-confirm-type: correlated # 消息发布确认（生产者->交换机是否成功） publisher-returns: true # 消息回退（交换机路由->队列是否成功） template: mandatory: true # 消息路由发送失败返回到队列中, 相当手动设置 rabbitTemplate.setMandatory(true); # 主要针对消费者的一些设置 listener: simple: acknowledge-mode: manual # 消息应答ACK方式 prefetch: 1 # 限制每次发送1条数据到队列（只可堆积1条到消费者）（预取值） concurrency: 1 # 最少需要一个消费者来监听同一队列 max-concurrency: 2 # 最大只能拥有2个消费者来监听同一队列 default-requeue-rejected: true # 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-19 0:25 * 交换机和队列的配置 */ @Configuration public class RabbitMQConfig { /* 一些简单的交换机-->路由-->队列的方式配置*/ //直接交换机 public static final String CONFIRM_EXCHANGE = "confirmExchange"; //队列 public static final String CONFIRM_QUEUE = "confirmQueue"; //绑定路由key public static final String CONFIRM_ROUTING_KEY = "confirmRoutingKey"; /*针对一些备份交换机的配置*/ //备份交换机 public static final String BACKUP_EXCHANGE = "backupExchange"; //备份队列（把一些无法被路由的消息进行备份） public static final String BACKUP_QUEUE = "backupQueue"; //报警队列（报警队列用来发送报警消息，告知消费者处理，以达到让管理员知道有数据处理不了） public static final String WARNING_QUEUE = "warningQueue"; //一些简单的队列和交换机的声明 /*** * 创建交换机消息 * @return Exchange */ @Bean("confirmExchange") public Exchange confirmExchange() { //一些其它参数 Map<String, Object> arguments = new HashMap<>(); //设置备份交换机信息，将来发送的消息无法被路由，就会发送到备份交换机 arguments.put("alternate-exchange", BACKUP_EXCHANGE); return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE).durable(true).withArguments(arguments).build(); } /*** * 队列名称 * @return Queue */ @Bean("confirmQueue") public Queue confirmQueue() { return QueueBuilder.durable(CONFIRM_QUEUE).build(); } /*** * 绑定关系 * @param confirmExchange 交换机消息 * @param confirmQueue 队列消息 * @return Binding */ @Bean("delayedQueueBindDelayedExchange") public Binding delayedQueueBindDelayedExchange(@Qualifier(value = "confirmExchange") Exchange confirmExchange, @Qualifier(value = "confirmQueue") Queue confirmQueue) { return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY).noargs(); } //备份交换机的创建 /*** * 创建备份交换机消息 * @return Exchange */ @Bean("backupExchange") public Exchange backupExchange() { return ExchangeBuilder.fanoutExchange(BACKUP_EXCHANGE).durable(true).build(); } /*** * 备份队列名称 * @return Queue */ @Bean("backupQueue") public Queue backupQueue() { return QueueBuilder.durable(BACKUP_QUEUE).build(); } /*** * 报警队列名称 * @return Queue */ @Bean("warningQueue") public Queue warningQueue() { return QueueBuilder.durable(WARNING_QUEUE).build(); } /*** * 绑定关系（备份队列绑定到备份交换机上） * @param backupExchange 备份交换机消息 * @param backupQueue 备份队列消息 * @return Binding */ @Bean("backupQueueBindBackupExchange") public Binding backupQueueBindBackupExchange(@Qualifier(value = "backupExchange") Exchange backupExchange, @Qualifier(value = "backupQueue") Queue backupQueue) { return BindingBuilder.bind(backupQueue).to(backupExchange).with("").noargs(); } /*** * 绑定关系（报警队列绑定到备份交换机上） * @param backupExchange 备份交换机消息 * @param warningQueue 报警队列消息 * @return Binding */ @Bean("warningQueueBindBackupExchange") public Binding warningQueueBindBackupExchange(@Qualifier(value = "backupExchange") Exchange backupExchange, @Qualifier(value = "warningQueue") Queue warningQueue) { return BindingBuilder.bind(warningQueue).to(backupExchange).with("").noargs(); } }
@Slf4j @Component @RequiredArgsConstructor public class RabbitMQMyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback { //注入rabbitTemplate对象 private final RabbitTemplate rabbitTemplate; /*** * 对象实例化完成（对象创建和属性注入）后调用此方法 */ @PostConstruct public void init() { //设置发布确认信息回调类RabbitTemplate.ConfirmCallback confirmCallback; rabbitTemplate.setConfirmCallback(this); //设置回退消息回调类ReturnsCallback.ReturnsCallback returnsCallback; rabbitTemplate.setReturnsCallback(this); //true：交换机无法将消息进行路由时，会将该消息返回给生产者；false：如果发现消息无法进行路由，则直接丢弃 rabbitTemplate.setMandatory(true); // 或使用配置 spring.rabbitmq.template.mandatory: true } /*** * 发布确认（生产者-->交换机的确认） * @param correlationData 回调的相关数据 * @param ack 消息是否成功发送给交换机，true成功，false失败 * @param cause 对于ack为false时会有对应的失败原因，否则为空 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { //获取对应的ID消息，因为不确认是否有ID被传入，所以取值需要判空 String id = correlationData == null ? "" : correlationData.getId(); //校验是否成功发送 if (ack) { log.info("消息已经成功交给了交换机，对应消息ID为：{}", id); } else { log.info("消息未能成功发送给交换机，对应消息ID为：{}，异常原因：{}", id, cause); } } /*** * 当消息无法被路由时执行当前回调 * @param returned 被退回的消息信息 */ @Override public void returnedMessage(ReturnedMessage returned) { // 发送的消息 Message message = returned.getMessage(); // 发送到哪个交换机 String exchange = returned.getExchange(); // 交换机到队列的路由key String routingKey = returned.getRoutingKey(); // 退回原因 String replyText = returned.getReplyText(); // 退回原因状态码 int replyCode = returned.getReplyCode(); //消息打印 log.info("信息被回退，从交换机：{} 路由：{} 发送到队列失败，发送信息为：{}，退回状态码：{} 和原因：{}", exchange, routingKey, message, replyCode, replyText); //我们可以在这后面对发送失败的消息进行处理 } }
@Slf4j @Component @RequiredArgsConstructor public class TestProducer { private final RabbitTemplate rabbitTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //其它的一些信息，用来回调用处 CorrelationData correlationData = new CorrelationData(); //设置id信息，其实默认就是UUID，我们其实可以根据自己设置指定ID信息 //correlationData.setId(String.valueOf(UUID.randomUUID())); //发送消息（不可以成功发送,路由key不存在） rabbitTemplate.convertAndSend(RabbitMQConfig.CONFIRM_EXCHANGE, RabbitMQConfig.CONFIRM_ROUTING_KEY + "test", bytes, correlationData); } }
@Slf4j @Component public class ConsumerA { /*** * 消费者 */ @RabbitListener(queues = {RabbitMQConfig.CONFIRM_QUEUE}, ackMode = "MANUAL") public void dlxConsumerTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); //队列名称 String consumerQueue = message.getMessageProperties().getConsumerQueue(); log.info("消费者从队列：{} 获取消息：{}，并处理完成手动确认", consumerQueue, messageSendDTO); channel.basicAck(deliveryTag, false); } } //============================== @Slf4j @Component public class BackupConsumer { /*** * 备份队列消费者接收的消息监听 */ @RabbitListener(queues = {RabbitMQConfig.BACKUP_QUEUE}, ackMode = "MANUAL") public void backupConsumerTest(@Payload String msgData, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); log.info("备份队列：{}，监听发送过来的数据并处理：{}", message.getMessageProperties().getConsumerQueue(), messageSendDTO); //手动确认 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); } } //============================== @Slf4j @Component public class WarningConsumer { /*** * 报警队列消费者接收的消息监听 */ @RabbitListener(queues = {RabbitMQConfig.WARNING_QUEUE}, ackMode = "MANUAL") public void warningConsumerTest(@Payload String msgData, Message message, Channel channel) throws IOException { //把接收过来的JSON信息转换为对象 MessageSendDTO messageSendDTO = JSONObject.parseObject(msgData, MessageSendDTO.class); log.info("报警队列：{}，监听发送过来的数据并处理：{}", message.getMessageProperties().getConsumerQueue(), messageSendDTO); //手动确认 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } }
@Slf4j @RestController @RequestMapping("/test") @RequiredArgsConstructor public class TestController { //注入生产者对象 private final TestProducer testProducer; /*** * 基本的POST请求，用来接收消息，并把消息交给生产者，并由生产者推送到指定交换机，由交换机分发消息 * @param msg 请求消息 * @return String */ @PostMapping("/produce") public String msgSendSimple(@RequestBody MessageSendDTO msg) { log.info("Controller接收到请求并把请求的信息交由生产者发送：{}", msg); //发送消息 testProducer.producerSendMsg(msg); return "请求发送成功，并已接收"; } }
可以好奇的是，为什么上面没有出现交换机路由发送给队列出异常的回退消息呢？并且也是设置mandatory参数的（和回退消息搭配使用）；如果回退消息和备份交换机两者同时开启，经过上面结果显示答案是备份交换机优先级比回退消息的优先级高。
十一：RabbitMQ重复消费问题（幂等性）
为什么RabbitMQ会出现生产者多次投递或者消费者多次消费呢？这里其实会有很多因素的，按道理正常处理不报错一般不会出现问题，但是不能一定保证，下面我就来和大家来探讨如何解决重复消费问题：
产生问题的几种情况：
生产者：如Controller接口被调用了两次，而Controller执行时会调用生产者，在没有处理幂等性问题时，这无形中调用2次生产者。
MQ：在消费者消费完准备响应ACK确认完成时，这时MQ突然宕机，所以消费者响应ACK就无法发送给MQ了，恰巧这时MQ修复完成启动了，
但MQ以为消费者还未消费该数据，MQ也没接收到ACK确认，所以MQ会覅后会再次推送该条消息，导致重复消费。
消费者：消费者已经消费完成，正准备ACK确认时，突然网络波动导致无法确认，连接中断，或者消费者宕机了，也无法ACK确认，消费
者重启后，MQ会再次推送原来的消息给消费者
解决方式：
最好的方式使用Redis缓存来完成，简单快捷方便，而且会定期清理历史消费完成的消息记录（缓存里，不是实际消费的具体数据被清理）
注：解决方式千千万，具体看自己项目
我解决的方式是使用Redis；首先需要对传递的消息类设置一个唯一标识（业务拼接的唯一ID或者UUID），这时要确保每次传递相同的消息UUID是相同的，在生产者投递前，先把消息的UUID设置到缓存里，并设置过期时间（这里的过期时间是，投递成功后，过期时间内可能会造成二次重复投递，但是过期时间之外不会存在再次投递，主要就是清理以前的缓存记录）；投递成功就等待过期时间结束自己清理，投递失败则需要把这次设置的缓存UUID删除，方便下次投递；在消费者那边，消费时首先对UUID缓存，代表当前已经消费成功或者消费中，但也要设置过期时间，和上面的过期时间一样用途，但是消费失败异常后也需要删除缓存，方便下次继续消费设置缓存；具体图：
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-14 21:39 * 测试生产者 */ @Slf4j @Component @RequiredArgsConstructor public class TestProducer { //注入rabbitTemplate对象 private final RabbitTemplate rabbitTemplate; //注入StringRedisTemplate对象 private final StringRedisTemplate redisTemplate; /*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { ValueOperations<String, String> forValue = redisTemplate.opsForValue(); try { //防止重复提交 //若设置成功则为True，设置不成功或者设置的值已经存在则返回False //这里设置20秒代表自动过期，一旦设置这个键值，消息被成功投递则不删除（防止20秒内重复提交），但是投递失败 //以后我需要删除这个键值，方便下次继续设置投递；；具体按照实际设置过期时间 Boolean result = forValue.setIfAbsent(msg.getUUID() + "-delivery", String.valueOf(msg.getMsgID()), 20, TimeUnit.SECONDS); //判断设置成功则发送消息（否则这个消息可能多次发送给消费者） if (Boolean.TRUE.equals(result)) { //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.ORDINARY_DIRECT_EXCHANGE,
RabbitMQConfig.ROUTE_KEY, bytes); } else { log.info("消息已经由生产者发送投递了，请忽重复投递！"); } } catch (Exception e) { //若生产者投递出现问题则代表投递不成功，删除这次缓存 redisTemplate.delete(msg.getUUID() + "-delivery"); } } }
/** * @author AnHui OuYang * @version 1.0 * created at 2023-04-13 23:29 * 消费者 */ @Slf4j @Component @RequiredArgsConstructor public class QueueConsumer { //注入StringRedisTemplate对象 private final StringRedisTemplate redisTemplate; /*** * 消费者（监听队列ordinaryQueue） * @param msgData 传递的具体消息，最好是生产者发送使用什么类型，这里接收就用什么类型 * @param message 这个就类似我们原生的message * @param channel 这个就类似我们原生的channel */ @RabbitListener(queues = {RabbitMQConfig.ORDINARY_QUEUE}, ackMode = "MANUAL") public void ordinaryQueueTest(@Payload String msgData, //这个是生产者发送的JSON消息 @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, //处理消息的编号 Message message, Channel channel) throws InterruptedException, IOException { ValueOperations<String, String> forValue = redisTemplate.opsForValue(); //获取到队列消息，因为发送是JSON格式，我们要解析对象格式 MessageSendDTO msg = JSONObject.parseObject(message.getBody(), MessageSendDTO.class); try { //判断消息有没有被消费过，没有消费过则设置（代表有消费者准备消费了） Boolean result = forValue.setIfAbsent(msg.getUUID(), String.valueOf(msg.getMsgID()), 1, TimeUnit.DAYS); //判断，若设置成功代表可以消费此条消息 if (Boolean.TRUE.equals(result)) { log.info("A：消息由消费者A消费：{}，并消费完成", msg); //手动确认，注：这个deliveryTag可以通过message.getMessageProperties().getDeliveryTag()拿到 channel.basicAck(deliveryTag, false); } else { log.info("消费者当前消费的消息被别的消费者已经消费过了，或者正在消费：{}", msg); //重复发送的也得手动确认掉，但是不处理 channel.basicAck(deliveryTag, false); } } catch (Exception e) { //若消费失败则删除之前的锁定（缓存）,下次队列投递给消费者的时候可以继续消费 redisTemplate.delete(msg.getUUID()); } } }
十二：优先级队列
其实我们发送到队列的消息是按照先进先出的顺序执行的，不能说后面的消息直接先出队列了；但是在日常开发中会遇到一些问题，比如生产者的投递速度是特别快的，而消费者消费特别慢，这时生产者投递消息有几千个，假设这时生产者投递了一个需要紧急处理的消息，这没办法，只能等前面几千个消息消费完成后才能等到那个紧急消息执行，这是显然不行的，所以就引出了优先级队列的解决方案。
RabbitMQConfig配置类修改：
/*** * 创建普通队列(并且添加优先级的配置) * @return Queue */ @Bean("ordinaryQueue") public Queue ordinaryQueue() { //其它参数 Map<String, Object> arguments = new HashMap<>(); //设置优先队列的值范围，官方允许0~255，设置10代表优先级范围为10，设置过大，后面排序耗费资源和CPU //代表后期生产者在投递消息时需要设置消息的0~10的优先级，越大越先执行 arguments.put("x-max-priority", 10); //构建队列 return QueueBuilder.durable(ORDINARY_QUEUE).withArguments(arguments).build(); }
生产者代码修改：
/*** * 生产者方法 * @param msg 消息 */ public void producerSendMsg(MessageSendDTO msg) { //循环发送投递消息 for (int i = 1001; i <= 1010; i++) { msg.setMsgID(i); //消息转换为JSON格式并转为字节数组 byte[] bytes = JSONObject.toJSONString(msg).getBytes(StandardCharsets.UTF_8); //若i为1005则代表让它的消息优先级高，其它都是默认0 if (i == 1005) { //设置其它参数（如这里需要设置优先级)（2种参数方式） //AMQP.BasicProperties properties = new AMQP.BasicProperties(); //properties.builder().priority(5).build(); MessageProperties messageProperties = new MessageProperties(); messageProperties.setPriority(5); //发送消息 Message message = new Message(bytes, messageProperties); rabbitTemplate.convertAndSend(RabbitMQConfig.ORDINARY_DIRECT_EXCHANGE, RabbitMQConfig.ROUTE_KEY, message); } else { //发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.ORDINARY_DIRECT_EXCHANGE, RabbitMQConfig.ROUTE_KEY, bytes); } } }
注意：我们设置队列优先级以后，那么在发送消息则需要设置优先级参数
十三：惰性队列
RabbitMQ 从 3.6.0 版本开始引入了惰性队列的概念。惰性队列会尽可能的将消息存入磁盘中，而在消费者消费到相应的消息时才会被加载到内存中，它的一个重要的设计目标是能够支持更长的队列，即支持更多的消息存储。当消费者由于各种各样的原因(比如消费者下线、宕机亦或者是由于维护而关闭等)而致使长时间内不能消费消息造成堆积时，惰性队列就很有必要了。默认情况下，当生产者将消息发送到 RabbitMQ 的时候，队列中的消息会尽可能的存储在内存之中，这样可以更加快速的将消息发送给消费者。即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份。当 RabbitMQ 需要释放内存的时候，会将内存中的消息换页至磁盘中，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。虽然 RabbitMQ 的开发者们一直在升级相关的算法，但是效果始终不太理想，尤其是在消息量特别大的时候。
队列具备两种模式：default 和 lazy。默认的为default 模式，在3.6.0 之前的版本无需做任何变更。lazy模式即为惰性队列的模式， 可以通过调用 channel.queueDeclare 方法的时候在参数中设置，也可以通过Policy 的方式设置，如果一个队列同时使用这两种方式设置的 话，那么Policy（这不具体展开说了）的方式具备更高的优先级。如果要通过声明的方式改变已有队列的模式的话，那么只能先删除队列，然后再
重新声明一个新的。在队列声明的时候可以通过“x-queue-mode”参数来设置队列的模式，取值为“default”和“lazy”。 下面示例中演示了一个惰性队列的声明细节： @Bean("ordinaryQueue") public Queue ordinaryQueue() { //其它参数 Map<String, Object> arguments = new HashMap<>(); //设置优先队列的值范围，官方允许0~255，设置10代表优先级范围为10，设置过大，后面排序耗费资源和CPU //代表后期生产者在投递消息时需要设置消息的0~10的优先级，越大越先执行 //arguments.put("x-max-priority", 10); //设置惰性队列 arguments.put("x-queue-mode", "lazy"); //构建队列 return QueueBuilder.durable(ORDINARY_QUEUE).withArguments(arguments).build();}
.
作者：蚂蚁小哥
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须在文章页面给出原文链接，否则保留追究法律责任的权利。