SpringBoot教程（十五） | SpringBoot集成RabbitMq（问题处理版）

RabbitMQ常见问题解决方案
- 问题一：消息丢失的解决方案
- - （1）生成者丢失消息
  - - 丢失的情景
    - 解决方案1：发送方确认机制（推荐，最常用）
    - 解决方案2：事务（不推荐，因为性能差）
  - （2）MQ丢失消息
  - - 丢失的情景
    - 解决方案：开启RabbitMQ的持久化+开启镜像队列
  - （3）消费者丢失消息
  - - 丢失的情景 1
    - 解决方案：无需解决
    - 丢失的情景 2
    - 扩展：重试机制
    - 解决方案：消费者方确认机制（推荐，最常用）
- 问题二：消息重复的解决方案
- - 什么时候会重复消费
  - 如何解决
- 问题三：保证消息顺序的解决方案
- - 单一队列和单一消费者模式（RabbitMQ）
- 问题四：消息堆积的解决方案
- - 消息堆积原因
  - 预防措施
  - 已出事故的解决措施

RabbitMQ常见问题解决方案

问题一：消息丢失的解决方案

首先明确一条消息的传送流程：生产者->MQ->消费者
所以这三个节点都可能丢失数据

（1）生成者丢失消息

丢失的情景

发送消息过程中出现网络问题：生产者以为发送成功，但RabbitMQ server没有收到

解决方案1：发送方确认机制（推荐，最常用）

发送方确认机制最大的好处在于它是异步的，等信道返回ark确认的同时继续发送下一条消息（不会堵塞其他消息的发送）

（一）修改application.properties配置

# 确认消息已发送到交换机(Exchange)
spring.rabbitmq.publisher-confirms=true #旧版本 
spring.rabbitmq.publisher-confirm-type=correlated #新版本 
# 确认消息已发送到队列(Queue)
spring.rabbitmq.publisher-returns=true

springBoot 2.2.0.RELEASE版本之前是使用 spring.rabbitmq.publisher-confirms=true
在2.2.0及之后使用spring.rabbitmq.publisher-confirm-type=correlated 属性配置代替

（二）新建配置文件RabbitTemplate

对于发送确认写法有多种方式，以下的是其中的一种方式

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitTemplateConfig {@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//setMandatory设置表示：消息在没有被队列接收时是否应该被退回给生产者（true:退回;false:丢弃）。//通常与yml配置文件中的publisher-returns配合一起使用，若不配置该项，setReutrnCallback将不会有消息返回rabbitTemplate.setMandatory(true);//帮助生产者判断 确认消息是否成功发送到RabbitMQ//ack 为true表示已发送成功 false表示发送失败rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);System.out.println("ConfirmCallback:     "+"确认情况："+ack);System.out.println("ConfirmCallback:     "+"原因："+cause);});//当消息无法 放到队列里面时 返回的提醒rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {System.out.println("ReturnCallback:     "+"消息："+message);System.out.println("ReturnCallback:     "+"回应码："+replyCode);System.out.println("ReturnCallback:     "+"回应信息："+replyText);System.out.println("ReturnCallback:     "+"交换机："+exchange);System.out.println("ReturnCallback:     "+"路由键："+routingKey);});return rabbitTemplate;}
}

解决方案2：事务（不推荐，因为性能差）

RabbitMQ提供的事务功能，在生产者发送数据之前开启RabbitMQ事务

（2）MQ丢失消息

丢失的情景

RabbitMQ服务端接收到消息后由于服务器宕机或重启等原因（消息默认存在内存中）导致消息丢失；

解决方案：开启RabbitMQ的持久化+开启镜像队列

RabbitMQ的持久化分为三个部分：交换器的持久化、队列的持久化、消息的持久化
三者都持久化才能保证 RabbitMQ服务重启之后，消息才能存在且能发出去

交换机持久化
交换机持久化描述的是当这个交换机上没有注册队列时，这个交换机是否删除。
如果要打开持久化的话也很简单（上面列子都是有体现的）

//定义直接交换机
@Bean
public DirectExchange directExchange() {//第一个参数:定义交换机的名称，第二个参数:是否持久化，第三个参数:是否自动删除return new DirectExchange("directExchange", true, false);
}

队列持久化
队列持久化描述的是当这个队列没有消费者在监听时，是否进行删除。
持久化做法：

//定义队列
@Bean
public Queue directQueue() {//第一个参数:队列的名称，第二个参数:是否持久化return new Queue("directQueue", true);
}

消息持久化

关键配置持久化（MessageDeliveryMode.PERSISTENT）

@Test
public void testDurableMessage() {// 1.准备消息Message message = MessageBuilder.withBody("hello, rabbitmq".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();// 2.消息ID，封装到CorrelationData中CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 3.发送消息rabbitTemplate.convertAndSend("simple.queue", message, correlationData);log.info("发送消息成功");
}

（3）消费者丢失消息

丢失的情景 1

RabbitMQ服务端向消费者发送完消息之后，网络断了，消息并没有到达消费者

解决方案：无需解决

无需解决。因为此情景下服务端收不到确认消息，会再次发送的。

丢失的情景 2

启用了重试机制，重试指定次数之后，还没成功，但消息被确认。

扩展：重试机制

重试机制的三大前提

重试模式已启用：通过配置 spring.rabbitmq.listener.simple.retry.enabled=true 来启用重试模式。
抛出了异常：在 @RabbitListener 标注的方法中抛出了异常，通常是 RuntimeException 或 Error。
Spring AMQP 会捕获这些异常并根据配置的重试策略来重试消息。
未达到最大重试次数：消息的重试次数尚未达到配置的最大值（spring.rabbitmq.listener.simple.retry.maxAttempts）。

配置以下即可实现重试操作

# 是否支持重试
spring.rabbitmq.listener.simple.retry.enabled=true
# 重试次数（默认3次）
spring.rabbitmq.listener.simple.retry.max-attempts=5

重点！！！：重试机制和是否能让队列删除消息没有任何关系

解决方案：消费者方确认机制（推荐，最常用）

默认情况下(自动确认模式)，消费者第一次从队列中接收到消息时，RabbitMQ会立即将该消息标记为已确认并从队列中删除（不管其是否存在业务逻辑的报错），这个消息就从队列上被删除了。

改成手动后就可以实现 “先操作业务逻辑（数据库操作）后，再手动从队列上删除这个消息” 的动作
其中“从队列上删除这个消息“这个动作体现就是使用 channel.basicAck 去完成的。
切记改成手动后，这个channel.basicAck方法一定要写。

（一）修改application.properties配置

# 设置消费端手动 ack
spring.rabbitmq.listener.simple.acknowledge-mode=manual

（二）修改Service接收信息项

当消息在进入 emailProcess、smsProcess（被@RabbitListener注解）方法时就已经被视为“接收到了”，但是需要你执行 channel.basicAck(手动确认)

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Service;import java.io.IOException;@Service
public class DirectReceiver {@RabbitHandler@RabbitListener(queues = "emailQueue")  //监听的队列名称public void emailProcess(Channel channel, Message testMessage) throws IOException {try{System.out.println(new String(testMessage.getBody(),"UTF-8"));//TODO 具体业务.......//你使用手动消息确认模式时，basicAck 一定要执行，不然会导致会保留在队列中，无法被消费//第1个参数表示消息投递序号//第2个参数false只确认当前一个消息收到（大多数情况下都设置为false），true确认所有consumer获得的消息channel.basicAck(testMessage.getMessageProperties().getDeliveryTag(),false);} catch (Exception e) {//若是消息没有成功接收，第二个参数设置为true的话，代表重新放回队列中，false则为丢弃，在此也可以做成放置死信队列的操作channel.basicReject(testMessage.getMessageProperties().getDeliveryTag(),false);}}}

确认和拒绝消息：

basicAck: 这个方法用于确认消息已被成功处理。
第一个参数是消息的delivery tag（用于标识消息），
第二个参数指定是否批量确认（false表示只确认当前消息）。
basicReject: 这个方法用于拒绝消息。
第一个参数同样是delivery tag，
第二个参数指定是否将消息重新放回队列（false表示不重新放回，即丢弃消息）。

方法解释：

emailProcess: 这个方法监听emailQueue队列。
当队列中有消息时，它会打印出消息的内容，并尝试确认消息。
如果处理过程中发生异常，它会拒绝消息，但不会重新放回队列（第二个参数为false）。

问题二：消息重复的解决方案

什么时候会重复消费

1.自动提交模式时

消费者收到消息后，要自动提交，但提交后，网络出故障，RabbitMQ服务器没收到提交消息，那么此消息会被重新放入队列，会再次发给消费者。

2.手动提交模式时

情景1：网络故障问题，同上。
情景2：接收到消息并处理结束了，此时消费者挂了，没有手动提交消息。

总体来说就是：网络不可达、消费端宕机。

如何解决

消费端处理消息的业务逻辑保持幂等性

比如你拿个数据要写库，先根据主键查一下，如果这数据有了，就别插入了，update 一下。
比如你是写 Redis，那没问题了，反正每次都是 set，天然幂等性。

问题三：保证消息顺序的解决方案

单一队列和单一消费者模式（RabbitMQ）

在RabbitMQ中，可以确保一个队列只被一个消费者消费，这样可以保证消息按照发送的顺序被处理。因为队列本身就是一个先进先出的结构。

适用场景：RabbitMQ用户且对消息顺序有严格要求的场景。
优点：实现简单，易于管理。
缺点：可能成为性能瓶颈，在处理大量消息时需要考虑扩展性。

问题四：消息堆积的解决方案

消息堆积原因

消息堆积即消息没及时被消费，是生产者生产消息速度快于消费者消费的速度导致的。
消费者消费慢可能是因为：本身逻辑耗费时间较长、阻塞了。

预防措施

生产者
1.减少发布频率
3.考虑使用队列最大长度限制
消费者
1.优化代码

已出事故的解决措施

情况1：堆积的消息还需要使用

方案1：简单修复

修复消费者（consumer）的问题，让他恢复消费速度，然后等待几个小时消费完毕

方案2：复杂修复

单队列消费转变为多队列并行消费

也是需要先修复消费者（consumer）的问题，再进行下面的步骤

步骤 1: 队列和路由设置
1.创建新队列：在RabbitMQ中创建10个新队列，每个队列分配一个独特的名称。
2. 设置交换机：定义一个直连型（Direct）交换机。
3. 绑定路由键：将每个新队列通过唯一的路由键绑定到直连型交换机上。

伪代码例子：

// 假设这是配置类的一部分  
@Bean  
Queue queue1() {  return new Queue("queue1", false);  
}  
@Bean  
Queue queue2() { return new Queue("queue2", false);  
}
// 以此类推，为其他9个队列创建Bean  
.........
@Bean  
DirectExchange exchange() {  return new DirectExchange("myExchange");  
}  
@Bean  
Binding binding1(Queue queue1, DirectExchange exchange) {  return BindingBuilder.bind(queue1).to(exchange).with("routingKey1");  
}  
@Bean  
Binding binding2(Queue queue2, DirectExchange exchange) { return BindingBuilder.bind(queue2).to(exchange).with("routingKey2");  
}
// 以此类推，为其他队列和路由键创建绑定  
......

步骤 2: 消息分发
1.接收堆积数据：现有消费者（或分发者）接收从发送者处堆积的数据。
2.分发到新队列：实现分发逻辑，将接收到的消息根据路由键分发到相应的10个新队列中。

伪代码例子：

@RabbitListener(queues = "oldQueue")  
public void emailProcess(Message testMessage, Channel channel) throws IOException {  try {  // 生成1-10之间的顺序数  SequentialRandom sequentialRandom = new SequentialRandom()String key = sequentialRandom.getNextSequentialRandom();// 重新发送消息到交换机，交换机将根据routingKey将消息路由到正确的队列  rabbitTemplate.convertAndSend("myExchange", "routingKey"+key, new String(testMessage.getBody(),"UTF-8")); // 确认原始队列中的消息（如果您想要的话）  channel.basicAck(testMessage.getMessageProperties().getDeliveryTag(), false);  } catch (Exception e) {  // 处理异常，可能包括记录日志、发送警报等  channel.basicReject(testMessage.getMessageProperties().getDeliveryTag(), false);  }  
}

public class SequentialRandom {  private int currentIndex = 1; // 初始索引为1  /**  * 获取下一个顺序数* @return 下一个数字，从1到10循环  */  public int getNextSequentialRandom() {  int next = currentIndex;  currentIndex = (currentIndex % 10) + 1; // 使用模运算实现循环，并更新索引  return next;  }  
}

步骤 3: 并行消费
1.开发新消费端：编写新的消费端程序，该程序能够同时监听并处理来自10个新队列的消息。
2. 部署并启动：将新消费端程序部署到服务器，并启动它以开始并行消费。

伪代码例子：

@Component  
public class ParallelConsumer {  @RabbitListener(queues = {"queue1", "queue2", ..., "queue10"}) // 列出所有要监听的队列  public void receiveMessage(Message message) {  // 处理消息  processMessage(message);  }  private void processMessage(Message message) {  // 实现消息处理逻辑  }  
}