消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能,成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件,如老牌的ActiveMQ、RabbitMQ,炙手可热的Kafka,阿里巴巴自主开发RocketMQ等。
使用消息队列可用6字概括:解耦、异步、削峰
解耦:将消息写入消息队列,需要消息的时候自己从消息队列中订阅,从而原系统不需要做任何修改。
异步:将消息写入消息队列,非必要的业务逻辑以异步的方式运行,加快响应速度
削峰:原系统慢慢的按照数据库能处理的并发量,从消息队列中慢慢拉取消息。在生产中,这个短暂的高峰期积压是允许的。
应用解耦、流量削峰、异步处理、消息通讯、远程调用
系统解耦
交互系统之间没有直接的调用关系,只是通过消息传输,故系统侵入性不强,耦合度低。
提高系统响应时间
例如原来的一套逻辑,完成支付可能涉及先修改订单状态、计算会员积分、通知物流配送几个逻辑才能完成;通过MQ架构设计,就可将紧急重要(需要立刻响应)的业务放到该调用方法中,响应要求不高的使用消息队列,放到MQ队列中,供消费者处理。
为大数据处理架构提供服务
通过消息作为整合,大数据的背景下,消息队列还与实时处理架构整合,为数据处理提供性能支持。
Java消息服务——JMS
Java消息服务(Java Message Service,JMS)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。
JMS中的P2P和Pub/Sub消息模式:点对点(point to point, queue)与发布订阅(publish/subscribe,topic)最初是由JMS定义的。这两种模式主要区别或解决的问题就是发送到队列的消息能否重复消费(多订阅)。
一致性 幂等性
MQ引入成本
增加引入复杂度 、只能保证最终一致性
系统可用性降低
系统引入的外部依赖越多,系统稳定性越差。一旦 MQ 宕机,就会对业务造成影响。如何保证MQ的高可用?
系统复杂度提高
MQ 的加入大大增加了系统的复杂度,以前系统间是同步的远程调用,现在是通过 MQ 进行异步调用。如何
保证消息没有被重复消费?怎么处理消息丢失情况?那么保证消息传递的顺序性?
一致性问题
A 系统处理完业务,通过 MQ 给B、C、D三个系统发消息数据,如果 B 系统、C 系统处理成功,D 系统处理
失败。如何保证消息数据处理的一致性?
1.点对点
PTP点对点:使用queue作为通信载体
说明:
消息生产者生产消息发送到queue中,然后消息消费者从queue中取出并且消费消息。
消息被消费以后,queue中不再存储,所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者,但是对一个消息而言,只会有一个消费者可以消费
![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/b57d97f84fe040bca039ca3cc1ac4412.png
2 发布/订阅
Pub/Sub发布订阅(广播):使用topic作为通信载体
说明:
消息生产者(发布)将消息发布到topic中,同时有多个消息消费者(订阅)消费该消息。和点对点方式不同,发布到topic的消息会被所有订阅者消费。
queue实现了负载均衡,将producer生产的消息发送到消息队列中,由多个消费者消费。但一个消息只能被一个消费者接受,当没有消费者可用时,这个消息会被保存直到有一个可用的消费者。
topic实现了发布和订阅,当你发布一个消息,所有订阅这个topic的服务都能得到这个消息,所以从1到N个订阅者都能得到一个消息的拷贝。
1、异步处理
将一个请求链路中的非核心流程,拆分出来,异步处理,减少主流程链路的处理逻辑,缩短RT,提升吞吐量。
有些业务不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。
如:注册新用户发短信通知
2、削峰填谷
避免流量暴涨,打垮下游系统,前面会加个消息队列,平滑流量冲击。比如:秒杀活动。生活中像电源适配器也是这个原理。
3、应用解耦
降低工程间的强依赖程度,针对异构系统进行适配。在项目启动之初来预测将来项目会碰到什么需求,是极其困难的。通过消息系统在处理过程中间插入了一个隐含的、基于数据的接口层,两边的处理过程都要实现这一接口,当应用发生变化时,可以独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
两个应用,通过消息系统间接建立关系,避免一个系统宕机后对另一个系统的影响,提升系统的可用性。如:下单异步扣减库存
4、消息通讯
内置了高效的通信机制,可用于消息通讯。如:点对点消息队列、聊天室
5. 冗余
有些情况下,处理数据的过程会失败。除非数据被持久化,否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理,通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中,在把一个消息从队列中删除之前,需要你的处理系统明确的指出该消息已经被处理完毕,从而确保你的数据被安全的保存直到你使用完毕。
6扩展性
因为消息队列解耦了你的处理过程,所以增大消息入队和处理的频率是很容易的,只要另外增加处理过程即可。不需要改变代码、不需要调节参数。便于分布式扩容。
7 过载保护
在访问量剧增的情况下,应用仍然需要继续发挥作用,但是这样的突发流量无法提取预知;如果以为了能处理这类瞬间峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力,而不会因为突发的超负荷的请求而完全崩溃。
8可恢复性
系统的一部分组件失效时,不会影响到整个系统。消息队列降低了进程间的耦合度,所以即使一个处理消息的进程挂掉,加入队列中的消息仍然可以在系统恢复后被处理。
9 顺序保证
在大多使用场景下,数据处理的顺序都很重要。大部分消息队列本来就是排序的,并且能保证数据会按照特定的顺序来处理。
10 缓冲
在任何重要的系统中,都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行,该缓冲有助于控制和优化数据流经过系统的速度。以调节系统响应时间。
11 数据流处理
分布式系统产生的海量数据流,如:业务日志、监控数据、用户行为等,针对这些数据流进行实时或批量采集汇总,然后进行大数据分析是当前互联网的必备技术,通过消息队列完成此类数据收集是最好的选择
AMQP协议
AMQP即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。
优点:可靠、通用
MQTT协议
MQTT(Message Queuing Telemetry Transport,消息队列遥测传输)是IBM开发的一个即时通讯协议,有可能成为物联网的重要组成部分。该协议支持所有平台,几乎可以把所有联网物品和外部连接起来,被用来当做传感器和致动器(比如通过Twitter让房屋联网)的通信协议。
优点:格式简洁、占用带宽小、移动端通信、PUSH、嵌入式系统
STOMP协议
STOMP(Streaming Text Orientated Message Protocol)是流文本定向消息协议,是一种为MOM(Message Oriented Middleware,面向消息的中间件)设计的简单文本协议。STOMP提供一个可互操作的连接格式,允许客户端与任意STOMP消息代理(Broker)进行交互。
优点:命令模式(非topic\queue模式)
XMPP协议
XMPP(可扩展消息处理现场协议,Extensible Messaging and Presence Protocol)是基于可扩展标记语言(XML)的协议,多用于即时消息(IM)以及在线现场探测。适用于服务器之间的准即时操作。核心是基于XML流传输,这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息,即使其操作系统和浏览器不同。
优点:通用公开、兼容性强、可扩展、安全性高,但XML编码格式占用带宽大
其他基于TCP/IP自定义的协议
有些特殊框架(如:redis、kafka、zeroMq等)根据自身需要未严格遵循MQ规范,而是基于TCP\IP自行封装了一套协议,通过网络socket接口进行传输,实现了MQ的功能。
Broker
消息服务器,作为server提供消息核心服务
Producer
消息生产者,业务的发起方,负责生产消息传输给broker,
Consumer
消息消费者,业务的处理方,负责从broker获取消息并进行业务逻辑处理
Topic
主题,发布订阅模式下的消息统一汇集地,不同生产者向topic发送消息,由MQ服务器分发到不同的订阅者,实现消息的 广播
Queue
队列,PTP模式下,特定生产者向特定queue发送消息,消费者订阅特定的queue完成指定消息的接收
Message
消息体,根据不同通信协议定义的固定格式进行编码的数据包,来封装业务数据,实现消息的传输
1、消息的批量处理
2、消息压缩,节省传输带宽和存储空间
3、零拷贝
4、磁盘的顺序写入
5、page cache 页缓存,由操作系统异步将缓存中的数据刷到磁盘,以及高效的内存读取
6、分区设计,一个逻辑topic下面挂载N个分区,每个分区可以对应不同的机器消费消息,并发设计。
1.产生背景: 生产者投递消息的速率与我们消费者消费的速率完全不匹配。
2.生产者投递消息的速率>消费者消费的速率
导致我们消息会堆积在我们 mq 服务器端中,没有及时的被消费者消费 所以就会产生消息堆积的问题
3.注意的是:rabbitmq 消费者我们的消息消费如果成功的话 消息会被立即删除(自动ack)
kafka 或者 rocketmq 消息消费如果成功的话,消息是不会立即被删除。
4.解决办法:
A.提高消费者消费的速率;(对我们的消费者实现集群)
B.消费者应该批量形式获取消息 减少网络传输的次数;
主要是消息的消费速度跟不上生产速度,从而导致消息堆积。解决思路:
1.可能是刚上线的业务,或者大促活动,流量评估不到位,这时需要增加消费组的机器数量,提升整体消费能力
2.也可能是消费端的问题,正常情况,一条消息处理需要10ms,但是优化不到位或者线上bug,现在要500ms,那么消费端的整体处理速度会下降50倍。这时,我们就要针对性的排查业务代码。例如数据库的一条sql没有命中索引,导致单条消息处理耗时拉长,进而导致消息堆积
如果 bug 导致几百万消息持续积压几小时。有如何处理呢? 需要解决bug,临时紧急扩容,大概思路如下:
1.先修复 consumer 消费者慢问题,以确保其恢复消费速度,然后将现有consumer 都停掉。
2.新建一个 topic,partition 原来 10 倍,临时立好原先 10 倍 queue数量。
3.然后写一个临时分发数据consumer 程序,这个程序部署上去消费积压数据,消费之后不做耗时处理,直接均匀轮询写入临时立好10 倍数量queue。
4.接着临时征用 10 倍机器来部署 consumer,每一批 consumer 消费一个临时queue 数据。这种做法相当于临时将 queue 资源和 consumer 资源扩大10 倍,以正常 10 倍速度来消费数据。
5.等快速消费完积压数据之后,得恢复原先部署架构,重新用原先consumer 机器来消费消息。
消息积压处理办法:临时紧急扩容先修复 consumer 的问题,确保其恢复消费速度,然后将现有 cnosumer 都停掉。新建一个 topic,partition 是原来的 10 倍,临时建立好原先 10 倍的 queue 数量。然后写一个临时的分发数据的 consumer 程序,这个程序部署上去消费积压的数据,消费之后不做耗时的处理,直接均匀轮询写入临时建立好的 10 倍数量的 queue。接着临时征用 10 倍的机器来部署 consumer,每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍,以正常的 10 倍速度来消费数据。等快速消费完积压数据之后,得恢复原先部署的架构,重新用原先的 consumer 机器来消费消息。 MQ中消息失效:假设你用的是 RabbitMQ,RabbtiMQ 是可以设置过期时间的,也就是 TTL。如果消息在 queue 中积压超过一定的时间就会被 RabbitMQ 给清理掉,这个数据就没了。那这就是第二个坑了。 这就不是说数据会大量积压在 mq 里,而是大量的数据会直接搞丢。我们可以采取一个方案,就是批量重导,这个我们之前线上也有类似的场景干过。就是大量积压的时候,我们当时就直接丢弃数据了,然后等过了高峰期以后,比如大家一起喝咖啡熬夜到晚上12点以后,用户都睡觉了。这个时候我们就开始写程序,将丢失的那批数据,写个临时程序,一点一点的查出来,然后重新灌入 mq 里面去,把白天丢的数据给他补回来。也只能是这样了。假设 1 万个订单积压 在 mq 里面,没有处理,其中 1000 个订单都丢了,你只能手动写程序把那 1000 个订单给查出来,手动发到 mq 里去再补一次。 mq消息队列块满了:如果消息积压在 mq 里,你很长时间都没有处理掉,此时导致 mq 都快写满了,咋办?这个还有别的办法吗?没有,谁让你第一个方案执行的太慢了,你临时写程序,接入数据来消费,消费一个丢弃一个,都不要了,快速消费掉所有的消息。然后走第二个方案,到了晚上再补数据吧。
ActiveMQ、RabbitMQ、Kafka、RocketMQ 这几种, 在架构技术选型的时候一般根据业务的需求选择合适的中间件: 比如中小型公司,低吞吐量的一般用 ActiveMQ、RabbitMQ 较为合适, 大数据高吞吐量的大型公司一般选用 Kafka 和 RocketMQ。
一般的业务系统要引入 MQ,最早大家都用 ActiveMQ,但是现在确实大家用的不多了,没经过大规模吞吐量场景的验证,社区也不是很活跃,所以大家还是算了吧,我个人不推荐用这个了;
后来大家开始用 RabbitMQ,但是确实 erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它,对公司而言,几乎处于不可控的状态,但是确实人家是开源的,比较稳定的支持,活跃度也高;
不过现在确实越来越多的公司会去用 RocketMQ,确实很不错,毕竟是阿里出品,但社区可能有突然黄掉的风险(目前RocketMQ 已捐给 Apache,但 GitHub 上的活跃度其实不算高)对自己公司技术实力有绝对自信的,推荐RocketMQ,否则回去老老实实用 RabbitMQ 吧,人家有活跃的开源社区,绝对不会黄
所以中小型公司,技术实力较为一般,技术挑战不是特别高,用 RabbitMQ 是不错的选择;大型公司,基础架构研发实力较强,用 RocketMQ 是很好的选择。
如果是大数据领域的实时计算、日志采集等场景,用 Kafka 是业内标准的,绝对没问题,社区活跃度很高,绝对不会黄,何况几乎是全世界这个领域的事实性规范。
ActiveMQ、RabbitMQ、Kafka、RocketMQ 这几种, 在架构技术选型的时候一般根据业务的需求选择合适的中间件: 比如中小型公司,低吞吐量的一般用 ActiveMQ、RabbitMQ 较为合适, 大数据高吞吐量的大型公司一般选用 Kafka 和 RocketMQ。
对于吞吐量来说kafka和RocketMQ支撑高吞吐,ActiveMQ和RabbitMQ比他们低一个数量级。对于延迟量来说RabbitMQ是最低的。
性能
消息中间件的性能主要衡量吞吐量,Kafka 的吞吐量比 RabbitMQ 要高出 1~2 个数级,
RabbitMQ 的单机 QPS 在万级别,Kafka 的单机 QPS 能够达到百万级别。RocketMQ 单机 写入 TPS 单实例约 7 万条/秒,单机部署 3 个 Broker,可以跑到最高 12 万条/秒,消息大小 10 个字节,Kafka 如果开启幂等、事务等功能,性能也会有所降低。
数据可靠性
Kafka 与 RabbitMQ 都具备多副本机制,数据可靠性较高。RocketMQ 支持异步实时刷盘, 同步刷盘,同步 Replication,异步 Replication。
服务可用性
Kafka 采用集群部署,分区与多副本的设计,使得单节点宕机对服务无影响,且支持消息容量 的线性提升。RabbitMQ 支持集群部署,集群节点数量有多种规格。RocketMQ 是分布式架 构,可用性高。
功能
Kafka 与 RabbitMQ 都是比较主流的两款消息中间件,具备消息传递的基本功能,但在一些 特殊的功能方面存在差异,RocketMQ 在阿里集团内部有大量的应用在使用。
优先级队列;队列设置最大的优先级,之后每条消息设置对应的优先级,队列根据消息优 先级进行消费,(在有可能队列堆积的情况才有意义);应用场景:不同业务消息推送。
RabbitMQ 是一个消息代理中间件,而 Apache Kafka 是一个分布式流处理平台。这种差异可能看起来只是语义上的,但它会带来严重的影响,影响我们方便地实现各种系统功能。
例如 Kafka 最适合处理流数据,在同一主题同一分区内保证消息顺序,而 RabbitMQ 对流中消息的顺序只提供基本的保证。
不过 RabbitMQ 内置了对重试逻辑和死信交换的支持,而 Kafka 将此类逻辑实现留给了用户。
使用C语言开发的一个Key-Value的NoSQL数据库,开发维护很活跃,虽然它是一个Key-Value数据库存储系统,但它本身支持MQ功能,所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作,各执行100万次,每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明:入队时,当数据比较小时Redis的性能要高于RabbitMQ,而如果数据大小超过了10K,Redis则慢的无法忍受;出队时,无论数据大小,Redis都表现出非常好的性能,而RabbitMQ的出队性能则远低于Redis。
号称最快的消息队列系统,专门为高吞吐量/低延迟的场景开发,在金融界的应用中经常使用,偏重于实时数据通信场景。ZMQ能够实现RabbitMQ不擅长的高级/复杂的队列,但是开发人员需要自己组合多种技术框架,开发成本高。因此ZeroMQ具有一个独特的非中间件的模式,更像一个socket library,你不需要安装和运行一个消息服务器或中间件,因为你的应用程序本身就是使用ZeroMQ API完成逻辑服务的角色。但是ZeroMQ仅提供非持久性的队列,如果down机,数据将会丢失。如:Twitter的Storm中使用ZeroMQ作为数据流的传输。
ZeroMQ套接字是与传输层无关的:ZeroMQ套接字对所有传输层协议定义了统一的API接口。默认支持 进程内(inproc) ,进程间(IPC) ,多播,TCP协议,在不同的协议之间切换只要简单的改变连接字符串的前缀。可以在任何时候以最小的代价从进程间的本地通信切换到分布式下的TCP通信。ZeroMQ在背后处理连接建立,断开和重连逻辑。
特性:
● 无锁的队列模型:对于跨线程间的交互(用户端和session)之间的数据交换通道pipe,采用无锁的队列算法CAS;在pipe的两端注册有异步事件,在读或者写消息到pipe的时,会自动触发读写事件。
● 批量处理的算法:对于批量的消息,进行了适应性的优化,可以批量的接收和发送消息。
● 多核下的线程绑定,无须CPU切换:区别于传统的多线程并发模式,信号量或者临界区,zeroMQ充分利用多核的优势,每个核绑定运行一个工作者线程,避免多线程之间的CPU切换开销。
设计消息队列中间件时需要考虑以下因素: