在微服务架构中,MQ(Message Queue,消息队列)扮演着至关重要的角色。MQ 是一种跨进程的通信机制,通常用于应用程序间进行数据的异步传输,也被称作“消息中间件”。在微服务架构中,MQ 的主要作用包括服务间的解耦、异步通信、流量削峰和数据同步等。
安装插件并启动 根据其路径可以访问界面化控制台
publisher-->exchange-->queue-->consumer
Spring AMQP(二次封装AMQP)
1. 引入依赖spring-boot-starter-amqp
2. 配置rabbitmq服务端信息
3. 利用RabbitTemplate发送信息
4.利用@RabbitListener注解声明要监听的队列,监听消息
消费者消息推送限制
默认情况下RabbitMQ会将消息轮询投递给绑定到队列的每一个消费者,没有考虑消费者的处理速度,可能出现消息堆积
多个消费者绑定一个队列,可以加快消息处理速度
同一个消息只会被一个消费者处理
通过设置profetch来控制消费者的预读的消息数量,处理完一条再处理下一条,能者多劳
扇形交换机:将消息广播到交换价绑定的所有队列,无视消息的路由键
直连交换机:将接收到的消息根据规则路由到指定的Queue,也称为定向路由
主题交换机:使用通配符匹配路由键,允许更灵活的消息路由
bean实现
注解实现
通过 @RabbitListener 的 bindings 属性声明 Binding(若 RabbitMQ 中不存在该绑定所需要 的 Queue、Exchange、RouteKey 则自动创建,若存在则抛出异常)
消息转换器 默认用到jkd的序列化 占用内存大,可读性差(控制台) 引入json相关依赖
描述:(默认关闭)由于网络波动,可能出现发送者连接MQ失败的情况,需要重新连接
配置:
改方式提高了消息发送成功的概率。但其用的阻塞式重连,降低了性能,要用的话可以考虑异步
为了提升性能,默认情况下MQ的数据都是在内存存储的临时数据,重启后就会消失。为了保证数据的可靠性,必须配置数据持久化,包括:
一旦出现消息堆积问题,RabbitMQ的内存占用就会越来越高,直到触发内存预警上限。此时RabbitMQ会将内存消息刷到磁盘上,这个行为成为PageOut
. PageOut
会耗费一段时间,并且会阻塞队列进程。因此在这个过程中RabbitMQ不会再处理新的消息,生产者的所有请求都会被阻塞。
为了解决这个问题,从RabbitMQ的3.6.0版本开始,就增加了Lazy Queues的模式,也就是惰性队列。惰性队列的特征如下:
而在3.12版本之后,LazyQueue已经成为所有队列的默认格式。因此官方推荐升级MQ为3.12版本或者所有队列都设置为LazyQueue模式。
交换机持久化
队列持久化
消息持久化
在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。但在某些特殊情况下,这会导致消息积压,比如:
消费者宕机或出现网络故障
消息发送量激增,超过了消费者处理速度
消费者处理业务发生阻塞
接收到消息后直接存入磁盘而非内存
消费者要消费消息时才会从磁盘中读取并加载到内存(也就是懒加载)
支持数百万条的消息存储
Mq的消费者可靠性是确保消息在消费过程中不丢失、正确处理且能够应对异常情况的重要方面。以下是从几个方面详细阐述Mq消费者可靠性的内容:
RabbitMQ等消息队列提供了消费者确认机制(Consumer Acknowledgement),即当消费者处理消息结束后,应该向MQ发送一个回执,告知MQ消息处理的状态。回执有三种可选值:
在消费者处理消息的过程中,可能会遇到暂时性的错误(如网络波动、数据库连接问题等),导致消息处理失败。为了应对这种情况,MQ通常提供了消息重试机制。当消费者处理消息失败时,MQ可以根据配置将消息重新投递给消费者进行重试。
对于无法正常处理的消息(如超过最大重试次数),MQ通常会将其转移到死信队列(Dead Letter Queue,DLQ)。死信队列是一个特殊的队列,用于存放那些无法正常处理的消息。通过监控死信队列,可以及时发现并处理这些异常消息,从而确保消息系统的可靠性。
消息的重复消费问题如何解决:保证消息的唯一性,保证消息的幂等性
为了确保Mq消费者系统的可靠性,还需要实施全面的监控体系。监控内容可以包括消息发送速率、消费延迟、队列长度、服务器健康状况等关键指标。一旦发现异常指标,应立即触发报警机制,以便及时介入处理并防止问题扩大。
这一机制确保了消息在处理过程中的可靠性,防止了消息因处理失败而丢失或未被正确处理的情况。
在分布式系统中,节点故障是不可避免的。为了确保Mq消费者系统的可靠性,需要采取容错与恢复措施。例如,可以通过集群部署、主备节点切换等方式来确保系统的高可用性。当某个节点发生故障时,其他节点可以接管其工作,从而确保消息的正常处理。