RabbitMQ高级用法

embedded/2024/10/18 7:53:04/

💥 该系列属于【SpringBoot基础】专栏，如您需查看其他SpringBoot相关文章，请您点击左边的连接

目录

一、发送者的可靠性

1. 生产者重试机制

2. 生产者确认机制【return和confirm机制】

（1）开启生产者确认

（2）定义ReturnCallback

（3）定义ConfirmCallback

二、MQ的可靠性

1. 数据持久化

（1）交换机持久化

（2）队列持久化

（3）消息持久化

2. LazyQueue

（1）控制台配置Lazy模式

（2）代码配置Lazy模式

三、消费者的可靠性

1. 消费者确认机制

2. 失败重试机制

（1）开启重试机制

（2）失败处理策略

3. 业务幂等性

（1）唯一消息ID

（2）业务判断

四、延迟消息

1. 死信交换机

2. DelayExchange插件

（1）下载和安装

（2）声明延迟交换机

（3）发送延迟消息

考虑消息丢失的可能性有哪些。消息从发送者发送消息，到消费者处理消息，需要经过的流程是这样的：

消息从生产者到消费者的每一步都可能导致消息丢失：

  • 发送消息时丢失：

    • 生产者发送消息时连接MQ失败

    • 生产者发送消息到达MQ后未找到Exchange

    • 生产者发送消息到达MQ的Exchange后，未找到合适的Queue

    • 消息到达MQ后，处理消息的进程发生异常

  • MQ导致消息丢失：

    • 消息到达MQ，保存到队列后，尚未消费就突然宕机

  • 消费者处理消息时：

    • 消息接收后尚未处理突然宕机

    • 消息接收后处理过程中抛出异常

综上，我们要解决消息丢失问题，保证MQ的可靠性，就必须从3个方面入手：

  • 确保生产者一定把消息发送到MQ

  • 确保MQ不会将消息弄丢

  • 确保消费者一定要处理消息

一、发送者的可靠性

1. 生产者重试机制

生产者发送消息时，出现了网络故障，导致与MQ的连接中断。为了解决这个问题，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后，多次重试。修改publisher模块的application.yaml文件，添加下面的内容：

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multipliermax-attempts: 3 # 最大重试次数

2. 生产者确认机制【return和confirm机制】

RabbitMQ提供了生产者消息确认机制，包括Publisher Confirm和Publisher Return两种。在开启确认机制的情况下，当生产者发送消息给MQ后，MQ会根据消息处理的情况返回不同的回执。

具体如图所示：

总结如下：

  • 当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ACK的确认信息，代表投递成功

  • 临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功

  • 持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功

  • 其它情况都会返回NACK，告知投递失败

其中ack和nack属于Publisher Confirm机制，ack是投递成功；nack是投递失败。而return则属于Publisher Return机制。

（1）开启生产者确认

在publisher模块的application.yaml中添加配置：

spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型publisher-returns: true # 开启publisher return机制

（2）定义ReturnCallback

每个RabbitTemplate只能配置一个ReturnCallback，因此我们可以在配置类中统一设置。在publisher模块定义一个配置类：

@Slf4j
@AllArgsConstructor
@Configuration
public class MqConfig {private final RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {@Overridepublic void returnedMessage(ReturnedMessage returned) {log.error("触发return callback,");log.debug("exchange: {}", returned.getExchange());log.debug("routingKey: {}", returned.getRoutingKey());log.debug("message: {}", returned.getMessage());log.debug("replyCode: {}", returned.getReplyCode());log.debug("replyText: {}", returned.getReplyText());}});}
}

（3）定义ConfirmCallback

MQ的回执就会通过这个Future来返回，我们可以提前给CorrelationData中的Future添加回调函数onFailure和onSuccess来处理消息回执。

新建一个测试，向系统自带的交换机发送消息，并且添加ConfirmCallback：

@Test
void testPublisherConfirm() {// 1.创建CorrelationDataCorrelationData cd = new CorrelationData();// 2.给Future添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {// 2.1.Future发生异常时的处理逻辑，基本不会触发log.error("send message fail", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {// 2.2.Future接收到回执的处理逻辑，参数中的result就是回执内容if(result.isAck()){ // result.isAck()，boolean类型，true代表ack回执，false 代表 nack回执log.debug("发送消息成功，收到 ack!");}else{ // result.getReason()，String类型，返回nack时的异常描述log.error("发送消息失败，收到 nack, reason : {}", result.getReason());}}});// 3.发送消息rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd);
}

二、MQ的可靠性

消息到达MQ以后，如果MQ不能及时保存，也会导致消息丢失，所以MQ的可靠性也非常重要。

默认情况下，RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题：一旦MQ宕机，内存中的消息会丢失；内存空间有限，当消费者故障或处理过慢时，会导致消息积压，引发MQ阻塞。

所谓的持久就是把数据都写在磁盘而不是仅仅在内存里面，重启MQ后依然可以查找到数据。

1. 数据持久化

为了提升性能，默认情况下MQ的数据都是在内存存储的临时数据，重启MQ后就会消失。为了保证数据的可靠性，必须配置数据持久化，包括：

  • 交换机持久化

  • 队列持久化

  • 消息持久化

（1）交换机持久化

（2）队列持久化

（3）消息持久化

控制台：

出现的问题：消息丢失，MQ阻塞

而换成持久化数据时：

2. LazyQueue

RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的模式，也就是惰性队列。惰性队列的特征如下：

  • 接收到消息后直接存入磁盘而非内存

  • 消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）

  • 支持数百万条的消息存储

 而在3.12版本之后，LazyQueue已经成为所有队列的默认格式。

（1）控制台配置Lazy模式

在添加队列的时候，添加x-queue-mod=lazy参数即可设置队列为Lazy模式：

（2）代码配置Lazy模式

在利用SpringAMQP声明队列的时候，添加x-queue-mod=lazy参数也可设置队列为Lazy模式：

@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy() // 开启Lazy模式.build();
}

基于注解来声明队列并设置为Lazy模式：

@RabbitListener(queuesToDeclare = @Queue(name = "lazy.queue",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){log.info("接收到 lazy.queue的消息：{}", msg);
}

lazyQueue并发能力更强，消息处理速度更快： 

三、消费者的可靠性

1. 消费者确认机制

为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。即：当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。回执有三种可选值：

  • ack：成功处理消息，RabbitMQ从队列中删除该消息

  • nack：消息处理失败，RabbitMQ需要再次投递消息

  • reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息

由于消息回执的处理代码比较统一，因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式，有三种模式：

  • none：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用

  • manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活

  • auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：

    • 如果是业务异常，会自动返回nack；

    • 如果是消息处理或校验异常，自动返回reject;

例如：

spring:rabbitmq:listener:simple:acknowledge-mode: auto # 或none或manual

确认机制修改为auto，模拟一个消息处理的异常：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {log.info("spring 消费者接收到消息：【" + msg + "】");if (true) {throw new MessageConversionException("故意的");}log.info("消息处理完成");
}

此时消息状态为unacked（未确定状态）：

放行以后，由于抛出的是消息转换异常，因此Spring会自动返回reject，所以消息依然会被删除：

将异常改为RuntimeException类型：

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {log.info("spring 消费者接收到消息：【" + msg + "】");if (true) {throw new RuntimeException("故意的");}log.info("消息处理完成");
}

在异常位置打断点，然后再次发送消息测试，程序卡在断点时，可以发现此时消息状态为unacked，放行以后，由于抛出的是业务异常，所以Spring返回nack，最终消息恢复至Ready状态，并且没有被RabbitMQ删除

2. 失败重试机制

（1）开启重试机制

SpringAMQP提供了消费者失败重试机制，在消费者出现异常时利用本地重试，而不是无限的requeue到mq。

修改consumer服务的application.yml文件，添加内容：

spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000ms # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

重启consumer服务，消费者抛出RuntimeException异常后进行测试。可以发现：

  • 消费者在失败后消息没有重新回到MQ无限重新投递，而是在本地重试了3次

  • 本地重试3次以后，抛出了AmqpRejectAndDontRequeueException异常。查看RabbitMQ控制台，发现消息被删除了，说明最后SpringAMQP返回的是reject

结论：

  • 开启本地重试时，消息处理过程中抛出异常，不会requeue到队列，而是在消费者本地重试

  • 重试达到最大次数后，Spring会返回reject，消息会被丢弃

（2）失败处理策略

在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。

因此Spring允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由MessageRecovery接口来定义的，它有3个不同实现：

  • RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式

  • ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队

  • RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer，失败后将消息投递到一个指定的，专门存放异常消息的队列，后续由人工集中处理。

在consumer服务中定义处理失败消息的交换机和队列，然后定义一个RepublishMessageRecoverer，关联队列和交换机

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

3. 业务幂等性

幂等是一个数学概念，用函数表达式来描述是这样的：f(x)=f(f(x)）。在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。

为了防止消息队列给消费者发送消息的时候潜在的非幂等现象，我们必须想办法保证消息处理的幂等性。这里给出两种方案：

  • 唯一消息ID

  • 业务状态判断

（1）唯一消息ID

每一条消息都生成一个唯一的id，与消息一起投递给消费者。消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库。如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。

 如何给消息添加唯一ID呢？

SpringAMQP的MessageConverter自带了MessageID的功能，我们只要开启这个功能即可。

以Jackson的消息转换器为例，在发送消息时添加id：

@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息jjmc.setCreateMessageIds(true);return jjmc;
}

存在业务侵入问题。

（2）业务判断

业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息，不同的业务场景判断的思路也不一样。

以支付修改订单的业务为例，我们需要修改OrderServiceImpl中的markOrderPaySuccess方法：

@Override
public void markOrderPaySuccess(Long orderId) {// UPDATE `order` SET status = ? , pay_time = ? WHERE id = ? AND status = 1lambdaUpdate().set(Order::getStatus, 2).set(Order::getPayTime, LocalDateTime.now()).eq(Order::getId, orderId).eq(Order::getStatus, 1).update();
}

四、延迟消息

电商中通常的做法就是：对于超过一定时间（如30分钟）未支付的订单，应该立刻取消订单并释放占用的库存。

如何才能准确的实现在下单后第30分钟去检查支付状态呢？

像这种在一段时间以后才执行的任务，我们称之为延迟任务，而要实现延迟任务，最简单的方案就是利用MQ的延迟消息了。

在RabbitMQ中实现延迟消息也有两种方案：

  • 死信交换机+TTL

  • 延迟消息插件

1. 死信交换机

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

  • 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false

  • 消息是一个过期消息，超时无人消费

  • 要投递的队列消息满了，无法投递

如果一个队列中的消息已经成为死信，并且这个队列通过dead-letter-exchange属性指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为死信交换机（Dead Letter Exchange）。而此时加入有队列与死信交换机绑定，则最终死信就会被投递到这个队列中。

死信交换机有什么作用呢？

  1. 收集那些因处理失败而被拒绝的消息

  2. 收集那些因队列满了而被拒绝的消息

  3. 收集因TTL（有效期）到期的消息

定义消费者：

定义死信交换机和队列：

 定义生产者：

2. DelayExchange插件

基于死信队列虽然可以实现延迟消息，但是太麻烦了。因此RabbitMQ社区提供了一个延迟消息插件来实现相同的效果。

（1）下载和安装

下载地址：rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ (github.com)

安装：

先查看RabbitMQ的插件目录对应的数据卷。

docker volume inspect mq-plugins

插件目录被挂载到了/var/lib/docker/volumes/mq-plugins/_data这个目录，我们上传插件到该目录下。

接下来执行命令，安装插件：

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

（2）声明延迟交换机

基于注解方式：delayed = "true"

@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);
}

基于@Bean的方式：delayed()

@Slf4j
@Configuration
public class DelayExchangeConfig {@Beanpublic DirectExchange delayExchange(){return ExchangeBuilder.directExchange("delay.direct") // 指定交换机类型和名称.delayed() // 设置delay的属性为true.durable(true) // 持久化.build();}@Beanpublic Queue delayedQueue(){return new Queue("delay.queue");}@Beanpublic Binding delayQueueBinding(){return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");}
}

（3）发送延迟消息

发送消息时，必须通过x-delay属性设定延迟时间：

@Test
void testPublisherDelayMessage() {// 1.创建消息String message = "hello, delayed message";// 2.发送消息，利用消息后置处理器添加消息头rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 添加延迟消息属性message.getMessageProperties().setDelay(5000);return message;}});
}


http://www.ppmy.cn/embedded/100914.html

相关文章

Python策略模式：灵活应对多变的业务逻辑

Python策略模式：灵活应对多变的业务逻辑

在软件开发中&#xff0c;我们经常遇到需要根据不同情况执行不同算法或行为的情况。这些场景下&#xff0c;如果直接在代码中嵌入大量的条件判断语句&#xff08;如if-else或switch-case&#xff09;&#xff0c;不仅会使代码变得难以维护&#xff0c;还会降低其扩展性和可复用…
阅读更多...
Qt第十六章 多媒体Multimedia

Qt第十六章 多媒体Multimedia

文章目录 多媒体音频播放音频录制音频低延迟音效低级音频播放和录制推送和拉取解码压缩音频到内存与音频处理相关的类 视频播放视频处理低级视频帧录制视频与视频处理相关的类 支持的媒体格式 多媒体 cmakelist 添加Multimedia模块 设备信息查询 #include <QAudioDevice>…
阅读更多...
ELK

ELK

ELK elk介绍前期准备1、修改主机名2、配置/ect/hosts3、检查防火墙selinux是否关闭4、时钟同步 elasticsearch部署介绍1、安装JAVA包2、解压安装包&#xff0c;修改配置文件 elasticsearch集群部署elaticsearch基础API操作1、RestFul API 格式2、查看节点信息3、查看索引信息和…
阅读更多...
【Material-UI】深入了解Radio Group中的useRadioGroup Hook

【Material-UI】深入了解Radio Group中的useRadioGroup Hook

文章目录 一、什么是useRadioGroup&#xff1f;1.1 Hook的返回值 二、useRadioGroup的基本用法2.1 代码示例2.2 代码解析 三、useRadioGroup的应用场景3.1 动态样式调整3.2 高级交互逻辑 四、使用useRadioGroup的最佳实践4.1 保持代码简洁4.2 结合主题定制4.3 注意无障碍设计 五…
阅读更多...
【Pyhthon读取 PDF文件表格 ，转为 CSV/TSV/JSON文件】

【Pyhthon读取 PDF文件表格 ，转为 CSV/TSV/JSON文件】

tabula-py tabula-py 是一个将 PDF 表格转换为 pandas DataFrame 的工具。 tabula-py 是 tabula-java 的包装器&#xff0c;需要您的机器上有 java。 tabula-py 还允许您将 PDF 中的表格转换为 CSV/TSV 文件。 tabula-py 的 PDF 提取准确度与 tabula-java 或 tabula app 相…
阅读更多...
（十五）Flink 内存管理机制

（十五）Flink 内存管理机制

在大数据领域，很多开源框架（Hadoop、Spark、Storm）都是基于 JVM 运行，但是 JVM 的内存管理机制往往存在着诸多类似 OutOfMemoryError 的问题，主要是因为创建大量的实例，超过 JVM 的最大堆内存限制，没有被有效的回收。这在很大程度上影响了系统的稳定性，因此很多框架都实…
阅读更多...
学习大数据DAY43 Sqoop 安装，配置环境和使用

学习大数据DAY43 Sqoop 安装，配置环境和使用

目录 sqoop 安装 配置 mysql sqoop 安装 sqoop 指令集 sqoop 使用 sqoop 创建 hive 表 sqoop 全量导入表 sqoop 增量导入表 sqoop 全量导出表 sqoop 分区表导入表 sqoop 分区表导出表 上机练习 sqoop 安装 配置 mysql create database test DEFAULT CHARACTER S…
阅读更多...
深度学习学习经验——长短期记忆网络（LSTM）

深度学习学习经验——长短期记忆网络（LSTM）

长短期记忆网络&#xff08;LSTM&#xff09; 长短期记忆网络&#xff08;LSTM&#xff0c;Long Short-Term Memory&#xff09;是一种特殊的循环神经网络&#xff08;RNN&#xff09;&#xff0c;专为解决 RNN 中长期依赖问题而设计。LSTM 引入了三个门和一个细胞状态&#x…
阅读更多...
最新文章

Copyright @ 2022~2023