环境构建
- 拉取镜像
docker pull rabbitmq:3.8.19-management
- 创建网络
docker network create app-tier --driver bridge
- 使用docker命令启动rabbitmq
docker run \
-e RABBITMQ_DEFAULT_USER=chenmin \
-e RABBITMQ_DEFAULT_PASS=cm021035 \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
--network app-tier \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3.8.19-management
- 进入容器
docker exec -it mq /bin/bash
- 输入IP地址进入到rabbitmq界面,地址为:http://ip:15672
rabbitmq使用
保证消息的可靠性
发送者的可靠性
1.重连机制
2.生产者确认
1.publisher-confirm-type:none,simple,correlatedMQ的可靠性
1.一旦MQ宕机,内存中的消息会丢失
2.内存空间有限,当消费者故障或处理过慢时,会导致消息积压,引发MQ阻塞
解决问题:1. 数据持久化 1. 交换机持久化 2. 队列持久化 3. 消息持久化 2. 惰性队列 1. 接收到消息后直接存入磁盘而非内存(内存中只保留最近的消息,默认2048条) 2. 消费者要消费消息时才会从磁盘读取并加载到内存 3. rabbitMq如何保证消息的可靠性 1. 通过配置可以让交换机,队列,消息都持久化 2. 引入lazyQueue,lazyQueue会将所有消息都持久化 3. 开启持久化和生产者确认时,rabbitmq只有在消息持久化完成后才会给生产者返回ack回执
消费者的可靠性
消费者确认机制
为了确认消费者是否成功处理消息,rabbitmq提供了消费者确认机制,当消费者处理
消息结束后,应该向rabbitmq发送一个回执,告知rabbitmq自己消息处理状态,回执有三中可选值:
ack:成功处理消息,rabbitmq从队列中删除该消息
nack:消息处理失败,rabbitmq需要再次投递消息
reject:消息处理失败并拒绝该消息,rabbitmq从队列中删除该消息
springAMQP已经实现了消息确认功能,并允许我们公国配置文件选择ACK处理方式,有三种方式:- none:不处理,即消息投递给消费者后立即ack,消息会立刻从mq删除,非常不安全
- manual:手动模式,需要自己在业务代码中调用api,发送ack或reject,存在业务入侵
- auto:自动模式,springAMQP利用aop对我们的消息处理逻辑做了环绕增强,当业务正常执行时则自动返回ack,当业务出现异常时,根据异常判断返回不同结果:
如果是业务异常,会自动返回nack
如果是消息处理或校验异常,自动返回reject
消费失败处理
失败消息处理策略1. RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式 2. ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队 3. RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机
业务幂等性
唯一消息id
在消息转换器中设置
延迟消息
死信交换机
当一个队列中的消息满足下列情况之一,就会成为死信:1. 消费者使用basic.reject或basic.nack声明消费失败,并且消息的requeue参数设置为false 2. 消息是一个过期消息(达到队列或消息本身设置的过期时间),超时无人消费 3. 要投递的队列消息堆积满了,最早的消息可能成为死信
如果队列通过dead-letter-exchange属性指定了一个交换机,那么该队列中的死信就会投递到这个交换机中,这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)
延迟消息插件
取消超时订单
源码地址:https://gitee.com/persionalPage/mq-demo
视频地址:https://www.bilibili.com/video/BV1mN4y1Z7t9?p=1&vd_source=7ebf9410d44fe497a305dad4df81bf1b