# rabbitmq-note
**Repository Path**: zy0912/rabbitmq-note
## Basic Information
- **Project Name**: rabbitmq-note
- **Description**: rabbitmq学习笔记
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 0
- **Created**: 2023-10-27
- **Last Updated**: 2023-11-07
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
# rabbitmq-note
## 快速入门
### 需求
在RabbitMQD控制台完成下列操作：
- 新建队列hello.queue1和hello.queue2
- 向默认的amp.fanout交换机发送一条消息
消息到达交换机，交换机把消息路由到队列，消费者监听队列
绑定路由
发送路由
- 查看消息是否到达hello.queue1和hello.queue2
Get Message(s) 可以查看消息，不会消费
## 数据隔离
### 需求
在RQ的控制台完成操作
- 新建一个用户
- 为zhengyue用户创建一个virtual host
- 测试不同virtual host之间的数据隔离现象
## Java客户端
### 快速入门
#### 需求
- 利用控制台创建simple.queue
- 在pulisher服务中，利用SpringAMQP直接向simple.queue发送消息
引入依赖
```xml
org.springframework.boot
spring-boot-starter-amqp
```
配置RQ服务端信息
```yml
spring:
rabbitmq:
host: 192.168.0.208 # 主机名
port: 5672 # 端口
virtual-host: /yueyue # 虚拟主机
username: zhengyue # 用户名
password: yAPNBngF # 密码
```
利用RabbitTemplate发送消息
```java
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void testSendMsg2Queue() {
String queueName = "simple.queue";
String msg = "hello, amqp!";
this.rabbitTemplate.convertAndSend(queueName, msg);
}
```
- 在consumer服务中，利用SpringAMQP编写消费者，监听simple.queue队列
利用@RabblitListener监听队列使消费者接受消息
```java
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
System.out.println("消费者收到了simple.queue的消息：【" + msg + "】");
}
```
### WorkQueue
#### 案例
模拟WorkQueue，实现一个队列绑定多个消费者
#### 基本思路
1. 在rq的控制台创建一个队列，名为work.queue
2. 在publisher服务中定义测试方法，在1s内产生50条消息，发送至work.queue
```java
@Test
void testWorkQueue() throws InterruptedException {
String queueName = "work.queue";
for (int i = 0; i < 50; i++) {
String msg = "hello, amqp, message_" + i;
this.rabbitTemplate.convertAndSend(queueName, msg);
Thread.sleep(20);
}
}
```
3. 在consumer服务中定义两个消息消费者，都监听work.queue队列
```java
@RabbitListener(queues = "work.queue")
public void listenSimpleQueue1(String msg) {
System.out.println("消费者1 收到了work.queue的消息：【" + msg + "】");
}
@RabbitListener(queues = "work.queue")
public void listenSimpleQueue2(String msg) {
System.err.println("消费者2 收到了work.queue的消息：【" + msg + "】");
}
```
消息默认平均的分配给了两个消费者
4. 消费者1每秒处理50条消息，消费者2每秒处理5条消息
```java
@RabbitListener(queues = "work.queue")
public void listenSimpleQueue1(String msg) throws InterruptedException {
System.out.println("消费者1 收到了work.queue的消息：【" + msg + "】");
Thread.sleep(20);
}
@RabbitListener(queues = "work.queue")
public void listenSimpleQueue2(String msg) throws InterruptedException {
System.err.println("消费者2 收到了work.queue的消息：【" + msg + "】");
Thread.sleep(200);
}
```
消息还是平均的分配给了两个消费者
5. 修改消费者消息推送限制
默认情况下，rq会将消息轮询投递给绑定在队列上的每一个消费者。但没有考虑消费者是否已经处理完消息，可能出现消息堆积。
因此需要修改application.yml，设置preFetch值为1，确保同一时刻最多投递给消费者1条消息
```yml
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成后才能获取下一个消息
```
已根据不同消费者的处理效率进行分配
#### Work模型的使用
- 多个消费者绑定到一个队列，可以加快消息处理速度
- 同一条消息只会被一个消费者处理
- 通过设置prefetch来控制消费者预取的消息数量，处理完一条再处理下一条，实现能者多劳
### 交换机的作用
- 接收publisher发送的消息
- 将消息按照规则路由到与之绑定的队列
- FanoutExcahang的会将消息路由到每个绑定的队列
### Fanout（广播）交换机
Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的队列，也叫广播模式。
#### 案例
使用SpringAMQP演示FanoutExchange的使用
#### 实现思路
1. 在rq控制台中，声明队列fanout.queue1 和 fanout.queue2
2. 在rq控制台中，声明交换机 zheng.fanout，将两个队列与其绑定
3. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1 和 fanout.queue2
```java
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
System.out.println("消费者1 收到了fanout.queue的消息：【" + msg + "】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
System.err.println("消费者2 收到了fanout.queue的消息：【" + msg + "】");
}
```
4. 在publisher中编写测试方法，向zheng.fanout发送消息
```java
@Test
void testSendFanout() {
String exchangeName = "zheng.fanout";
String msg = "hello, everyone";
this.rabbitTemplate.convertAndSend(exchangeName, null, msg);
}
```
两个队列都收到了广播的消息
### Direct（定向）交换机
Direct Exchange 会将接收到的消息根据规则路由到指定的queue，因此称为 定向路由
- 每一个queue都与exchange 设置一个bindingKey
- 发布者发送消息时，指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
#### 案例
使用SpringAMQP演示FanoutExchange的使用
1. 在rq控制台中，声明队列direct.queue1和direct.queue2
2. 在rq控制台中，声明队列zheng.direct，将两个队列与其绑定
3. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
```java
@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {
System.out.println("消费者1 direct.queue的消息：【" + msg + "】");
}
@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {
System.err.println("消费者2 direct.queue的消息：【" + msg + "】");
}
```
4. 在publisher中编写测试方法，利用不同的RoutingKey向zheng.direct发送消息
```java
@Test
void testSendDirect() {
String exchangeName = "zheng.direct";
String msg = "红色警报！";
this.rabbitTemplate.convertAndSend(exchangeName, "red", msg);
String msg = "蓝色警报！";
this.rabbitTemplate.convertAndSend(exchangeName, "blue", msg);
}
```
#### 
### Topic（话题）交换机
Topic Exchange 与 Direct Exchange 类似，区别在于 routingKey 可以是多个单词的列表，并且以 . 分割
Queue 与 Exchange 指定 BindingKey时可以使用通配符：
- #：代指0个或多个**单词**
- *：代指一个**单词**
#### 案例
1. 在rq的控制台中，声明队列 topic.queue1 和 topic.queue2
2. 在rq的控制台中，声明交换机zheng.topic，将两个队列与其绑定
3. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1 和 topic.queue2
```java
@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg) {
System.err.println("消费者1 topic.queue的消息：【" + msg + "】");
}
@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg) {
System.err.println("消费者2 topic.queue的消息：【" + msg + "】");
}
```
4. 在publisher 中编写测试方法，利用不同的RoutingKey向zheng.topic发送消息
```java
@Test
void testSendTopic() {
String exchangeName = "zheng.topic";
String msg = "我是新闻！";
this.rabbitTemplate.convertAndSend(exchangeName, "china.news", msg);
String msg2 = "tz新闻！";
this.rabbitTemplate.convertAndSend(exchangeName, "tz.news", msg2);
}
```
#### 效果
### Direct 和 Topic 的差异
- Topic交换机接收的的消息RoutingKey可以是多个单词，以 . 分割
- Topic交换机与队列绑定时的bingingKey可以指定通配符
- #：代表0个或多个词
- *：代表1个词
### Java客户端 声明队列和交换机
#### 方案一：Bean
声明队列、交换机、绑定关系的Bean
- Queue：用于声明队列，可以用工厂类QueueBuilder构建
- Exchange：用于声明交换机，可以用工厂类ExchangeBuilder构建
- Binding：用于声明队列和交换机的绑定关系，可以用工厂类BindingBuilder构建
##### Direct类型
```java
@Configuration
public class DirectConfiguration {
@Bean
public DirectExchange directExchange() {
// 方案1
// return ExchangeBuilder.directExchange("apitest.direct").build();
// 方案2
return new DirectExchange("apitest.direct");
}
@Bean
public Queue directQueue1() {
// 方案 1
// return QueueBuilder.durable("apitest.direct"); // 持久化
// 方案 2 默认持久化
return new Queue("apitest.direct.queue1");
}
@Bean
public Binding DirectBinding(Queue directQueue1, DirectExchange directExchange) {
return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
}
@Bean
public Queue directQueue2() {
// 方案 1
// return QueueBuilder.durable("apitest.direct"); // 持久化
// 方案 2 默认持久化
return new Queue("apitest.direct.queue2");
}
@Bean
public Binding DirectBinding2() {
return BindingBuilder.bind(directQueue2()).to(directExchange()).with("red");
}
}
```
##### Fanout类型
```java
@Configuration
public class FanoutConfiguration {
@Bean
public FanoutExchange fanoutExchange() {
// 方案1
// return ExchangeBuilder.fanoutExchange("apitest.fanout").build();
// 方案2
return new FanoutExchange("apitest.fanout");
}
@Bean
public Queue fanoutQueue1() {
// 方案 1
// return QueueBuilder.durable("apitest.fanout"); // 持久化
// 方案 2 默认持久化
return new Queue("apitest.fanout");
}
@Bean
public Binding fanoutBinding(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
@Bean
public Queue fanoutQueue2() {
// 方案 1
// return QueueBuilder.durable("apitest.fanout"); // 持久化
// 方案 2 默认持久化
return new Queue("apitest.fanout2");
}
@Bean
public Binding fanoutBinding2() {
return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
}
}
```
#### 方案二：注解
基于@RabbitListener注解声明队列和交换机
- @Queue
- @Exchange
```java
/**
* SpringAMQP 还提供了基于@RabbitListener注解来声明队列和交换机的方式
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1", durable = "true"),
exchange = @Exchange(name = "zheng.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void annoDirectQueue(String msg) {
System.out.println("消费者1 direct.queue的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2", durable = "true"),
exchange = @Exchange(name = "zheng.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void annoDirectQueue2(String msg) {
System.err.println("消费者2 direct.queue的消息：【" + msg + "】");
}
```
### 消息转换器
#### 需求
测试利用SpringAMQP发送对象类型的消息
1. 声明一个队列，名为object.queue
2. 编写单元测试，向队列中直接发送一条消息，消息类型为Map
```java
@Test
void testSendObject() {
Map msg = new HashMap<>();
msg.put("name", "jack");
msg.put("age", 21);
this.rabbitTemplate.convertAndSend("object.queue", msg);
}
```
3. 在控制台查看消息
#### 问题分析
Spring 的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。
存在下拉问题：
- JDK的序列化有安全风险
- JDK序列化的消息太大
- JDK序列化的消息可读性差
#### 处理方案
建议采用JSON序列化代替默认的JDK序列化，要做两件事情
在publisher和consumer中都要引入jackson依赖：
```xml
com.fasterxml.jackson.core
jackson-databind
```
在publisher和consumer中都要配置MessageConverter：
```java
@Bean
public MessageConverter jacksonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
```
效果
### 实际应用
## MQ高级
### 发送者的可靠性
#### 生产者重连
有的时候由于网络波动，可能会出现客户端连接MQ失败的情况。通过配置我没可以开启链接失败后的重连机制：
```xml
spring:
rabbitmq:
connection-timeout: 1s # 设置MQ的连接超时时间
template:
retry:
enabled: true # 开启超时重试机制
initial-interval: 1000ms # 失败后初始等待时间
multiplier: 1 # 失败后下次等待时长的倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数
```
当网络不稳定的时候，利用重试机制可以有效提高消息发送的成功率。不过SpringAMQP提供的重试机制是**阻塞式**的重试，也就是说多次重试等待的过程中，当前线程是被阻塞的，会影响业务性能
如果对于业务性能有要求，建议**禁用**重试机制。如果一定要使用，请合理配置等待时长和重试次数，当然也可以考虑使用**异步**线程来执行发送消息的代码。
#### 生产者确认
rq有Publisher Confirm 和 Publisher Return 两种确认机制。开启确认机制后，在RQ成功收到消息后会返回确认消息给生产者。返回结果有以下情况：
- 消息投递了MQ，但是路由失败。此时会通过PublisherReturn返回路由异常原因，然后返回ACK，告知投递成功
- 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
- 持久消息投递到了MQ，并且入队完成持久化，返回ACK，告知投递成功
- 其它情况都会返回NACK，告知投递失败
1. 在publisher这个微服务的application.yml中添加配置
```yml
spring:
rabbitmq:
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启 publisher return 机制 路由失败问题一般由开发者导致，故一般不配置
```
配置说明：
- 这里publisher-confirm-type有三种模式可选
- none：关闭confirm机制
- simple：同步阻塞等待MQ的回执消息
- correlated：MQ异步回调方式返回回执消息
2. 每个RabbitTemplate只能配置一个ReturnCallback，因此需要在项目启动过程中配置
```java
@Slf4j
@Configuration
public class MqConfirmConfig implements ApplicationContextAware {
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
// 获取RabbitTemplate
RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
// 配置回调
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{}，消息{}",
returnedMessage.getReplyCode(), returnedMessage.getReplyText(),
returnedMessage.getExchange(), returnedMessage.getRoutingKey(),
returnedMessage.getMessage());
}
});
}
}
```
3. 发送消息，指定消息ID、消息ConfirmCallback
```java
@Test
void testPublisherConfirm() throws InterruptedException {
// 创建CorrelationData
CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
// 给Future添加ConfirmCallback
cd.getFuture().addCallback(new ListenableFutureCallback() {
@Override
public void onFailure(Throwable ex) {
// Future 发送异常时的处理逻辑，基本不会触发
log.error("消息回调失败", ex);
}
@Override
public void onSuccess(CorrelationData.Confirm result) {
log.debug("收到confirm callback 回执");
// Future 接收到回执的处理逻辑，参数中的result就是回执内容
if(result.isAck()) { // resulat.isAck, boolean, true 代表ack回执，fasle代表nack回执
log.debug("发送消息成功，收到ack！");
} else {// result.getReason string类型
log.error("发送消息失败，收到nack，reason：{}", result.getReason());
}
}
});
this.rabbitTemplate.convertAndSend("zheng.direct", "red", "hello", cd);
Thread.sleep(2*1000);
}
```
##### 如何处理生产者的确认消息
- 生产者确认需要额外的网络和系统资源开销，尽量不要使用
- 如果一定要使用，无需开启Publisher-Return机制，因为一般路由失败是自己业务问题
- 对于nack消息可以有限次数重试，依然失败则记录异常消息
### MQ的可靠性
在默认情况下，MQ会将收到的消息保存在内存中以降低消息收发的延迟。会导致两个问题
- 一旦MQ宕机，内存中的消息会丢失
- 内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发MQ阻塞
#### 数据持久化
MQ实现数据持久化包括3个方面：
- 交换机持久化
- 队列持久化
- 消息持久化
```java
@Test
void testPageOut() {
Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8))
.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
rabbitTemplate.convertAndSend("simple.queue", message);
}
```
#### LazyQueue
**3.6版本开始，增加Lazy Queue模式**
**3.12版本后，所有队列默认Lazy Queue模式，无法修改**
惰性队列的特征：
- 接收到消息后直接存入磁盘而非内存（内存中只保留最近的消息，默认2048条）
- 消费者要消费消息时才会从磁盘中读取并加载到内存
- 支持数百万的消息存储
##### 控制台增加惰性队列
##### Java代码-声明Bean
```java
@Bean
public Queue lazyQueue() {
return QueueBuilder
.durable("lazy.queue")
.lazy() // 开启lazy模式
.build();
}
```
##### Java代码-注解
```java
@RabbitListener(queuesToDeclare = @Queue(name = "lazy.queue", durable = "true",
arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg) {
System.out.println("消费者 lazy.queue的消息：【" + msg + "】");
}
```
直接写入磁盘
##### 总结
RabbitMQ如何保证消息的可靠性？
- 首先通过配置可以让交换机、队列、以及发送的消息都持久化。这样队列中的消息会持久化到磁盘，MQ重启消息依然存在。
- MQ在3.6版本引入LazyQueue，并且在3.12后成为队列的默认模式。LazyQueue会将所有消息持久化。
- 开启持久化和生产者确认时，MQ只要在消息持久化完成后才会给生产者返回ACK回执。
### 消费者的可靠性
#### 消费者确认
为了确认消费者是否成功处理消息，MQ提供了消费者确认机制（Consumer Acknowledgement）。当消费者处理消息结束后，应该向MQ发送一个回执，告知MQ自己消息处理状态。
- ack：成功处理消息，MQ从队列中删除该消息，有三种方式
- none：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用
- manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但是更灵活
- auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack。当业务出现异常时，根据异常判断返回不同结果：
- 如果是业务异常，会自动返回nack
- 如果是消息处理或校验异常，自动返回reject
```yml
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: auto # 消费者确认机制
```
```java
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {
System.out.println("消费者收到了simple.queue的消息：【" + msg + "】");
throw new MessageConversionException("手动异常"); // 消息处理异常，返回reject
}
```
- nack：消息处理失败，MQ需要再次投递消息
- reject：消息处理失败并拒绝该消息，MQ从队列中删除该消息 （消息本身有问题）
#### 失败重试机制
当消费者出现异常后，消息会不断requeue（重新入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力。
我们可以利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。
```yml
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成后才能获取下一个消息
retry:
enabled: true # 开启消费者重试
initial-interval: 1000ms # 失败后初始等待时间
multiplier: 1 # 失败后下次等待时长的倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数
stateless: true # true无状态，false有状态。如果业务中包含事务，这里改为false
```
重试次数耗尽才会报错，且reject，队列丢弃消息
在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现：
- RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
- ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
- **RepublishMessageRecoverer：重试耗尽后，将消息投递到指定的交换机**
##### 失败消息处理策略
将失败策略改为RepublishMessageRecoverer：
1. 定义接收失败消息的交换机、队列及其绑定关系
2. 定义RepublishMessageRecoverer
```java
@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}
```
##### 实例代码
```java
@Configuration
/**
* 前置条件是enable为true配置生效
*/
@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry", name = "enabled", havingValue = "true")
public class ErrorConfiguration {
@Bean
public DirectExchange errorExchange() {
return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue() {
return new Queue("error.queue");
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorExchange) {
return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
}
@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}
}
```
##### 总结
消费者如何保证消息一定被消费？
- 开启消费者确认机制为auto，由spring确认消息处理成功后返回ack，异常返回nack
- 开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理
#### 业务幂等性
幂等是一个数学概念，用函数表达式来描述是这样的：f(x) = f(f(x))。在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。
##### 业务令牌
在表单提交前由后端生成一个唯一标识保存到redis，并发到前端，等表单提交时携带标识执行提交并在redis中删除
##### 唯一消息id
是给每个消息都设置一个唯一id，利用id区分是否是重复消息（业务入侵大）
1. 每一条消息都生成一个唯一的id，与消息一起投递给消费者。
2. 消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库
3. 如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理
```java
@Bean
public MessageConverter jacksonMessageConverter() {
// 定义消息转换器
Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
// 配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jjmc.setCreateMessageIds(true);
return jjmc;
}
```
##### 业务判断
是结合业务逻辑，基于业务本身做判断。以我们的业务为例：我们要在支付后修改订单状态为已支付，应该在修改订单状态前先查询订单状态，判断状态是否是未支付。只有未支付订单才需要修改，其它状态不做处理
##### 总结
如何保证支付服务与交易服务之间的订单状态一致性？
- 首先，支付服务会在用户支付成功以后利用MQ消息通知交易服务，完成订单状态同步
- 其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制、消费者确认、消费者失败重试等策略，确保消息投递和处理的可靠性。同时也开启了MQ的持久化，避免因服务宕机导致消息丢失。
- 最后，我们还在交易服务更新订单状态时做了业务幂等判断，避免因消息重复消费导致订单状态异常
如果交易服务消息处理失败，有没有什么兜底方案？
- 我们可以在交易服务设置定时任务，定期查询订单支付状态。这样即便MQ通知失败，还可以利用定时任务佐伟兜底方案，确保订单支付状态的最终一致性。
### 延迟消息
延迟消息：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。
延迟任务：设置在一定时间之后才执行的任务
#### 死信交换机
当一个队列中的消息满足下列情况之一时，就会成为死信（dead letter）：
- 消费者使用basic.reject或basic.nack声明消息失败，并且消息的requeue参数设置为false
- 消息是一个过期消息（达到了队列或消息本身设置的过期时间），超时无人消费
- 要投递的队列消息堆积满了，最早的消息可能成为死信
如果队列通过dead-letter-exchange属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机（DLX）。
创建队列时指定死信交换机
发送消息
```java
@Test
void testTTLMessage() {
rabbitTemplate.convertAndSend("simple.direct", "hi", "hello",
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setExpiration("10000"); // 10秒
return message;
}
});
log.info("消息发送成功");
}
```
监听死信队列
```java
@RabbitListener(queues = "dlx.queue")
public void listenDlxQueue(String msg) {
log.info("dlx.queue的消息：【" + msg + "】");
}
```
#### 延迟消息插件
MQ官方也推出了一个插件，原生支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列
插件安装地址：[GitHub - rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ](https://github.com/rabbitmq/rabbitmq-delayed-message-exchange)
将文件拷贝到MQ的plugins目录下，执行命令 (文件无需加ez后缀名)
```bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
```
##### JavaBean 创建
```java
@Bean
public DirectExchange delayExchange() {
return ExchangeBuilder
.directExchange("delay.direct")
.delayed() // 设置delay的属性为true
.durable(true) // 持久化
.build();
}
```
##### 注解 创建
```java
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
exchange = @Exchange(name = "delay.direct", delayed="true"),
key = "delay"
))
public void listenDelayMessage(String msg) {
log.info("接收到 delay.queue的延迟消息消息：【" + msg + "】");
}
```
##### 发送消息
```java
@Test
void testPublisherDelayMessage() {
rabbitTemplate.convertAndSend("delay.direct", "delay", "hello delay",
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(10000); // 10秒
return message;
}
});
log.info("消息发送成功");
}
```
##### 总结
延迟消息会产生CPU压力，所以适合较短延时的业务场景
#### 取消超时订单
设置30分钟后检测订单支付状态实现存在的问题：
- 如果并发较高，30分钟可能堆积消息过多，对MQ压力很大
- 大多数订单在下单后1分钟内就会支付，但是却需要在MQ内等待30分钟，浪费资源
