网站建设分为几个时期,链接交换公司,攀枝花网站seo,企业网站定制开发一条龙全包中间件middleware 内容管理 introwhy use MQMQ实现漫谈主流消息队列QMQ IntroQMQ架构QMQ 存储模型 本文还是从理论层面分析消息队列中间件 cfeng现在处于理论分析阶段#xff0c;以中间件例子#xff0c;之前的blog对于中间件是从使用角度分享了相关的用法#xff0c;现在就…中间件middleware 内容管理 introwhy use MQMQ实现漫谈主流消息队列QMQ IntroQMQ架构QMQ 存储模型 本文还是从理论层面分析消息队列中间件 cfeng现在处于理论分析阶段以中间件例子之前的blog对于中间件是从使用角度分享了相关的用法现在就从理论层面分析中间件后面再从理论出发尝试分析中间件以及实现中间件这样我们才能更好的自定义相关的功能 intro
消息队列在如今的软件系统中扮演着重要的角色。cfeng之前work中使用RabbitMQ进行的服务通信和解耦消息队列的发布/订阅模型常常用于服务解耦众多开源实现RabbitMQ、RocketMQ、Kafka、QMQ…都是采用的分布式架构支持弹性高可用。 消息队列的概念很简单 Message Queue就是消息传播过程中保存消息的一个队列。一端连接生产者一端连接消费者。
why use MQ
为什么要使用MQ呢首先就是考虑企业中的应用场景比如实时索引更新、 异步化流程… 这些使用MQ就可以轻易解决。 MQ最主要的功能就是异步解耦、流量削峰… 异步 消息生产者将消息投递到消息队列中因为可靠性由MQ保证所以生产者可以继续处理剩余的业务逻辑提升可靠性比如注册成功之后发送通知短信… 松耦合 生产者和消费者不需要知道对方的存在只需要事先定义好消息的格式不需要知道对方的实现细节相比RPC调用强耦合被调用方异常或者响应慢可能产生回压甚至雪崩MQ可以为业务提升稳定性 cfeng之前work中各个后台系统的通信就采用的MQ进行异步通信eg 新增一个用户到系统需要在主系统中同步那么我直接在这个队列中增加一条insert的消息主系统接收即可因为是异步的松耦合
后面这样也出现了问题本身预算不足MQ集群规模小所以MQ的压力增大后面又改成了PRC调用Feign数据分发 Fanout广播不同的消费者组全量拷贝同一主体的消息队列topic并且彼此互不干扰广播模式的MQ在很多场景下可以发挥重要作用eg 对于一个旅游列表变更提供方而言那么机票、酒店、车票、火车票…等多个业务都不需要轮询旅游业务就都可以收到变更事件 流量削峰 消息队列具备积压能力。很多业务像秒杀具有潮汐效应就是流量成峰谷状如果实时性要求不高那么就可以利用Queue的积压能力进行削峰填谷就可以不用增加硬件… 可靠投递 消息队列本身需要考虑的问题就是可靠性不能够丢失消息。一般成熟的MQ产品都是两次RPC 一次转储实现整个流程。【生产者投递消息到MQMQ作为broker将消息通知给消费者】 消息队列适用场景
上下游业务解耦 订单支付完成后核心业务完成但是可能还需要给用户加积分发放优惠劵这些不是核心逻辑可靠性要求低那么就可以引入MQ来进行解耦延迟通知: 用户订单下单后可能还没有支付需要在30分钟内支付未支付那就取消订单可以使用延迟MQ实现大数据离线分析: 用户的行为日志通过实时系统进入MQ当业务处于低峰期时且资源充裕即可进行信息分析缓存同步: 类似实时价格变动事件需要刷新缓存可以通过MQ广播给消费者这样就可以不用进行数据库轮询了【轮询会有实时性误差】
MQ实现漫谈
消息队列主要就是一个中间队列具有积压消息的功能同时作为系统间的解耦利器一定是单独部署的。
NoSQL充当消息队列
如果只是需要一个地方进行消息的存取那么客户端可以直接将消息比如写入MongoDB中消费者从里面拖出来即可。没有任何的broker代理
一旦升级涉及大量的客户端【因为代码耦合严重】消费者之间的协调只能通过DB进行会涉及DB的modify等操作性能很低没有弹性 producer ------ ---Mongo DB--- ------ Consumer
引入中间server充当broker
没有server导致MQ只是一个没有弹性的容器好的MQ应该能够具有消息协调的功能的
producer ------- Broker -------- consumer| |Mongo DB引入一个broker生产者和消费者直接交互的是borker那么broker就可以起一个协调作用客户端很轻只是和broker通讯告诉borker需要发消息即可升级只需要升级borker即可。
但是这里没有binding一个topic绑定到一个mongo的表上存储粒度为一个topic【mongo这边消息的清理思路 分表-- 写了上一张表写下一张表上一张表直接drop不用一张表里面insert、delete】
再引入客户端和服务端SDK
上面的还是一个很基础的实现因为消费者和生产者和Broker的交互性很低所以borker的协调性很低那么客户端和服务端引入SDK制订一套完整的规则引入binding加上一些辅助组件就可以形成一个综合的MQ ---------------------- --- MetaServer - ----------------------| Producer | | | Consumer || HTTP Recelver | ----- Broker ---- | Http Deliver || APP | | | App |---------------------- MySQL ---------------------这个也不可能是真正的MQ的架构只是一个比较抽象的想法。MQ作为成熟的产品那么就需要具备优秀的性能。需要考虑很多方面的事情
消息的写入 消息怎么写入的更快批量?..)消息的投递的及时性 延迟怎么降低 (partition Stick… 截获代替轮询 Long Pulling…)集群管理 …
主流消息队列
之前cfeng快速的讲解过Spring中使用RabbitMQ的方式【关键就是binding、exchange和queue配置的时候一个binding就是将exchange和queue绑定生产者发送消息的时候指定exchage和bindingKey就可投放到指定的Queue消费者监听消费Queue中消息即可】 可以看到的是Kafka的可用性相比RabbitMQ是非常高的拥有成熟生态日志系统、流式系统…、活跃社区
Kafka主要是Scala实现的同时能够很好的集成到java生态中
RabbitMQ主要是Erlang实现的cfeng之前work业务量小采用的该MQ
RocketMQ是ali利用java实现的具备较高的可靠性
QMQ是qunar利用java是实现的采用的无序消费存储模型Kafka
Kafka将一个Topic分成多个Parition每一个Partition作为一个Broker的物理文件通过Apend only的方式实现文件顺序写的高性能线性提高集群单topic的吞吐量。
但是当Broker上所有的Topic的Partition总和过多时会产生随机写
Partition 0 |0|1|2|3|4|5|6|7|.. --------
Partition 1 |0|1|2|3|4|5|6|... -------|---- Writes
Partition 2 |0|1|2|3|4|5|6|7|8|... -------|Old ----------------------- new 【Kfaka写入消息】顺序访问和随机访问的性能不同 随机访问时需要小号磁头寻道和盘片旋转等待的时间
SSD使用的是半导体闪存介质随机访问和顺序访问的差异不大 硬盘/吞吐 顺序写 随机写 顺序读 随机读SATA 125M 548K 124M 466KSSD 592M 549M 404M 505M【使用fio测试工具每次访问4KB工具】测试开发机磁盘访问速度数据RocketMQ
RocketMQ吸取了Kafka中多Partition消息文件会导致随机写的教训采用的是单一消息文件 Commit Log 将所有Topic的消息在物理上全部顺序追加到Commit Log文件中。
上述操作可以能增加消息写入的吞吐量但是消费方在消费历史【操作系统Page Cache正在发生IO条件为未命中Page Cache实时消费基本不会引入IO】消息时候会引入随机读。
RocketMQ是一主多从架构主写从读只有主节点提供写操作从节点比较空闲RocketMQ将历史消息消费通过重定向到从节点 来缓解随机读 无论是Kafka还是RocketMQ都存在一个约束 一个Partition只能绑定在一个Consumer上 因此 消费者集群上限是Partition的数目Partition的均衡性可能导致消费组个别机器的负载高、积压多。
eg: 一个Tpic(cfeng.fx.kafka.example)设置了3个partition(012)如果消费组(kafka.example.group)初始化两台机器一台消费者消费一个partition另一个消费者消费两个partition 这个时候如果消费能力不够那么通过水平扩容消费者的方案 ❌ 此时Kafka | RocketMQ 只能通过 增加partition来进行Rebalance但Rebalance之后只能对新生产的消息生效 原本积压的消息不会被Rebalance 可能会破环消息的顺序性同时清理积压会对新的消息有积压耗时partition 1 | | | | | | | | | | | -------- Consumer 1
partition 2 | | | | | | | | | | | -------- Consumer 2
partition 3 | | | | | | | | | | | -------- Consumer 3生产者通过选定某个字段如tenant_id作为Partition Key来决定将消息投递到哪个Partition因此Partition Key会影响消费速度 eg: 比如一个Partition Key分布不均匀时就会出现某些Counsmer的消费速度达不到生产速度也就是消费能力不足导致消息积压partition 1 | | | | -------- Consumer 1
partition 2 | | | | | -------- Consumer 2
partition 3 | | | | | | | | | | | | -------- Consumer 3这里就发现 Consumer 3 的消费能力不足出现消息积压而Consumer 1和2则相对空闲QMQ Intro
最近cfeng了解架构的时候就经常浏览github寻找相关的资料进行辅助的study在探索消息队列的时候就在gitee上看到了qunar开源的QMQ【其和Hermes的区别就是Hermes是以MySQL作为消息持久化存储而QMQ则是以磁盘文件进行存储】
QMQ相比其他的MQ比较小众这里也就简单的探究一下这款MQ产品
事务消息 生产者消息可靠投递
一些业务比如订单类型业务对于可靠性的要求很高一些场景如业务系统宕机或网络暂时不可用时也需要确保消息可靠投递 — 如何解决
QMQ的解决方案 生产者侧引入持久化存储【MySQL、MongoDB…】发送消息之前先将消息持久化到存储中之后再异步化发送消息当Broker返回消息发送成功的结果之后将消息从持久化存储中del当生产者突然宕机那么负责补发消息的watch dog会代理消息发送的工作 QMQ同时支持事务消息依赖的是存储的本地事务实现分布式事务还可以通过两阶段提交 Two-phase Commit 但是两阶段提交的对于本地事务来说 交互过多流程复杂性能较低 并且业务系统大多依赖MySQL进行存储
延迟消息
这是很多MQ都支持的一个特性比如超时30分钟未支付订单取消就可以使用延迟MQ进行实现
定时重试
某些业务系统特定的流程也就是状态机只有当某个前置条件满足时才会消费这条消息条件队列 这个时候不能直接丢弃这种消息 QMQ可以设置定时重试的功能让业务定时重新进行消费
同机房生产与消费
生产者采用同机房投递的策略这样可以避免跨机房流量的产生 消费者默认多机房消费 消费者不用关心生产者机房部署结构核心链路的业务支持单元化只是消费本单元内的消息可以实现单元隔离
消息检索追踪
消息检索追踪是非常必要的应该实现离线任务按照时间回溯选择性重发、端到端耗时的长尾问题排查、未消费问题的排查重发、死信重发…
按照时间端筛选消息
记录每条消息的消息ID、创建时间、接收时间、broker组、详细信息支持重发等操作也就是MQ具有良好的消息治理功能同时QMQ还支持广播消息、Server随意扩容等多种特性再Spring中使用也是annotation化非常便捷
QMQ架构
QMQ服务端包含3个核心组件 Meta Server、 Broker、 Delay Meta Server 元信息管理服务 用于消息路由控制下发维持Broker和Server的心跳 还有上下线挂历、消费者进度管理… 当Meta Server检测到Broker或者Delay的心跳失联那就标记下线 Broker是QMQ存储核心 用于接收消息并持久化到磁盘文件中创建消息索引管理消费进度响应拉取请求。Broker实现HA采用的是主从模式 Master能接收读写请求 仿照的PacificA实现主从切换…当Master宕机自动选举新的Master继续服务 Delay 是接收延迟消息并持久化到本地磁盘当超过延迟时间后消息将被投递到Broker。 Delay的HA也是采用的主从模式副本保证消息的灾备。 可以看出延迟消息是剥离的单独的服务【RocketMQ是集成在Broker的逻辑中RabbitMQ也是在Broker的逻辑中但是是增加了死信交换机和死信队列】
QMQ考虑的因素
1. 延迟和实时是两种消息类型隔离可以不互相影响提高可靠性
2. 在达到延迟时间时消费者路由可能发生迁移如果逻辑耦合在Broker中那么就需要进行重定向【单一职责最好】RabbitMQ模型RabbitMQ broker--------------------------
producre 1 ---- | exchange1 --- quwuw1 |--- consumer12.| .... |
producer 2 ---- | exchange2 --- queue2 | --- consumer n-------------------------QMQ 存储模型
之前提到过Kafka和RocketMQ的存储模型中的Parition缺点QMQ采用的是独特的无序消费存储模型 同时有序模型和二者是相同的 Message Log 存储所有Topic的消息消息顺序写入此文件中 避免发生类似Kafka多partition文件造成文件随机写性能下降的问题Consumer Log 是以Topic为维度组织的Message Log的索引文件。索引记录固定长度记录这个Topic的第X条消息在Message中的物理偏移量和消息大小 【感觉类似OS中的存储】
QMQ的无序存储模型中 不存在Q与单一Consumer的绑定关系而是一个消费者组consumer group中的消费者合力消费一个Q所以消费者组是支持动态扩容的
当没有了单一的consumer和Q的绑定关系
每一个消费者的ACK和pull都是离散的所以不能通过Q的ACK和pulloffset来管理消费者的消费进度
QMQ 抽象一层Pull Log 记录Consumer在Consumer Log中的offset 当Consumer重启后读取pull Log即可定位进行消费当消费者拉取积压过久的消息没有命中Page Cache时提到就会产生读磁盘操作对于文件系统负担过重。 QMQ采用的时类似RocketMQ所有消息顺序写入Message Log索引文件对应的物理偏移基本是块离散的【一个物理块可能是多个消息Topic】 QMQ就给Message Log进行排序排序后的Message Log再增加索引文件 ----- 相同主题Topic的消息是块连续可以充分利用OS的预读特性提升效率
后面Cfeng会结合QMQ的源码进行详细的分析这里只是见到那提及QMQ整体上的架构和RabbitMQ相比可能更好的适配java生态
