消息总线(Message Queue),后文称MQ,是一种跨进程的通信机制,用于上下游传递消息。
MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。
结论:调用方实时依赖执行结果的业务场景,请使用调用,而不是MQ。
1. 任务时间有先后顺序 2. 任务会依赖前面任务处理的结果 方案是,采用MQ解耦: 1)task1准时开始,结束后发一个“task1 done”的消息 2)task2订阅“task1 done”的消息,收到消息后第一时间启动执行,结束后发一个“task2 done”的消息 3)task3同理 特别说明:MQ只用来传递上游任务执行完成的消息,并不用于传递真正的输入输出数据。 另外:这种方式也可以用工作流引擎来处理。
上游需要关注执行结果时要用“调用”,上游不关注执行结果时,就可以使用MQ了。 举例: 智联招聘 用户 更改简历,会 通知分析系统重新分析简历,通知关注你的人你已经做跟了简历更新可以联系你了 等等。 做法1:在更新简历的时候 调用 其他方法。 优点:实现简单 不足: A:更新简历的时间增加了。 -- 后边的通知并不是现在就要通知到的,可以有一定的延迟;本质是 实时调用应该完成的只是需要实时处理的任务,后续可延迟的任务后续处理。 B:如果后面的没有通知到,会影响更新简历的功能。上下游逻辑+物理依赖严重 C:如果后边增加 一个关注该消息的功能,那么就需要修改这个方法。属于架构设计中典型的依赖倒转(本应该B依赖A的处理消息的,但现在是A依赖B的处理结果) 做法2:采用MQ解耦,更新成功后发一个消息即可,订阅消息这自己处理。 优点是: 1)上游执行时间短 2)上下游逻辑+物理解耦,除了与MQ有物理连接,模块之间都不相互依赖 3)新增一个下游消息关注方,上游不需要修改任何代码
有时候上游需要关注执行结果,但执行结果时间很长(典型的是调用离线处理,或者跨公网调用),也经常使用回调网关+MQ来解耦。
借用一个例子:
说明:
1. 上游应该说的就是调用方
2. MQ应该是微信提供的接口
3. 对接的时候,在调用支付后,还需要订阅这个消息,这样应用服务就是微信MQ的订阅方了,在支付成功后就可以通知你。
场景:公示3天后,没人有异议,就算通过。 最Low方法:做一个定时任务轮训,可以cron也可以是java中的timer,如果轮训间隔设置长了则粒度大,时效性不好,设置太短,则效率低。 1.Java中java.util.concurrent.DelayQueue 优点:JDK自身实现,使用方便,量小适用 缺点:队列消息处于jvm内存,不支持分布式运行和消息持久化 2.Rocketmq延时队列 优点:消息持久化,分布式 缺点:不支持任意时间精度,只支持特定level的延时消息 3.Rabbitmq延时队列(TTL+DLX实现) 优点:消息持久化,分布式 缺点:延时相同的消息必须扔在同一个队列 4.Redis实现 参考:https://www.cnblogs.com/lylife/p/7881950.html 5.当然可以自己写: 思路: A:创建一个3600(S)的循环链表 B:启动一个Timer,每秒移动一个下标 C:下标每到一个Node就判断该node是否有需要执行的JOB,有则发搜一个消息去执行 D:添加任务时,需要 delayTime/3600 作为循环的次数,每循环一次需要--;delayTime%3600作为偏移量。 RocketMQ消息产生后,生产者希望在间隔一段时间后被消费的场景可以使用定时消息,RocketMQ目前不支持自定义延迟时间,但可以指定延迟等级,可以选择18个延迟等级,分别是对应延迟时间是1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。
MQ要想尽量消息必达,架构上有两个核心设计点:
(1)消息落地
(2)消息超时、重传、确认
投递消息:MQ持久化后返回给 消息的生产者。
消费消息:只有当消息的消费者 确认 收到消息才删除消息。
为了降低消息丢失的概率,MQ需要进行超时和重传。
投递消息:如果MQ一直没有确认,MQ Client就会重发,直到MQ确认收到,如果超时就会回调 投递失败。
消费消息:如果消费者一直没有回复,MQ Server会重发消息,采用指数间隔时间(1s 2s 4s …),这是消费者会收到多条一样的数据。
消费者拿到了重复的消息,就需要去重,实现幂等性。
幂等性设计至关重要 投递消息的幂等性是由MQ来完成,逻辑是:对每条消息,MQ系统内部必须生成一个inner-msg-id,作为去重和幂等的依据,这个内部消息ID的特性是: (1)全局唯一 (2)MQ生成,具备业务无关性,对消息发送方和消息接收方屏蔽 消费消息的幂等性是由消费者保证,为了保证业务幂等性,业务消息体中,必须有一个biz-id,作为去重和幂等的依据,这个业务ID的特性是: (1)对于同一个业务场景,全局唯一 (2)由业务消息发送方生成,业务相关,对MQ透明 (3)由业务消息消费方负责判重,以保证幂等
削峰填谷:顾名思义,是将流量的高峰 削去,保证系统可用性。在系统稍微空闲(谷)的时候处理。 原因是:一个处理链路,各个节点处理能力不同。一般情况,上游的业务系统都可以支撑过高并发,下游的业务系统,处理逻辑相对复杂,单TPS处理时间长。 如果让下游业务系统跟上游处理能力相同,这个费用相当大。 举个栗子,秒杀业务: 上游发起下单操作 下游完成秒杀业务逻辑(库存检查,库存冻结,余额检查,余额冻结,订单生成,余额扣减,库存扣减,生成流水,余额解冻,库存解冻) 上游下单业务简单,每秒发起了10000个请求,下游秒杀业务复杂,每秒只能处理2000个请求,很有可能上游不限速的下单,导致下游系统被压垮,引发雪崩。 解决思路: 1. 上游发慢点 -- 用户下单处理不及时,脑子有问题 2. 下游按照正常容量来处理 -- 这个可以,别采用订阅模式,使用MQ的拉模式 下游改动: 1. 定时任务:用定时任务去拉 2. 批处理:为提高处理效率,一次拉一批,处理也按照批处理方式
end
联系客服