Notify

1 有什么用

作用 ： 解耦，异步，并行

举个栗子

假设我们有这么一个应用场景，为了完成一个用户注册淘宝的操作，可能需要将用户信息写入到用户库中，然后通知给红包中心给用户发新手红包，然后还需要通知支付宝给用户准备对应的支付宝账号，进行合法性验证，告知sns系统导入新的用户等10步操作。
那么针对这个场景，一个最简单的设计方法就是串行的执行整个流程，如图1所示：

图1 ： 用户注册流程 - 串行流程
这种方式的最大问题是，随着后端流程越来越多，每步流程都需要额外的耗费很多时间，从而会导致用户更长的等待延迟。自然的，我们可以采用并行的方式来完成业务，能够极大的减少延迟，如图2所示。
![image.png](https://img-blog.csdnimg.cn/img_convert/c35b77ac5a3257aa77beb70637747de5.png#clientId=ue811b460-
图2 ：用户注册流程 - 并行流程
但并行以后又会有一个新的问题出现了，在用户注册这一步，系统并行的发起了4个请求，那么这四个请求中，如果通知SNS这一步需要的时间很长，比如需要10秒钟的话，那么就算是发新手包，准备支付宝账号，进行合法性验证这几个步骤的速度再快，用户也仍然需要等待10秒以后才能完成用户注册过程。因为只有当所有的后续操作全部完成的时候，用户的注册过程才算真正的“完成”了。用户的信息状态才是完整的。而如果这时候发生了更严重的事故，比如发新手红包的所有服务器因为业务逻辑bug导致down机，那么因为用户的注册过程还没有完全完成，业务流程也就是失败的了。这样明显是不符合实际的需要的，随着下游步骤的逐渐增多，那么用户等待的时间就会越来越长，并且更加严重的是，随着下游系统越来越多，整个系统出错的概率也就越来越大。
通过业务分析我们能够得知，用户的实际的核心流程其实只有一个，就是用户注册。而后续的准备支付宝，通知sns等操作虽然必须要完成，但却是不需要让用户等待的。 这种模式有个专业的名词，就叫最终一致。为了达到最终一致，我们引入了MQ系统。业务流程如下：

图3 ：主流程

图4 ： 异步流程

整体流程
解耦 ： 消息发送方和消息消费方解耦
异步&并行 ： 消息消费之间异步&并行

2 是什么

核心原理

Notify的设计理念和传统的MQ有些不同，Notify的核心理念是

1. 为了消息堆积而设计系统

• 为了消息堆积而设计系统在市面上的大部分MQ产品，大部分的核心场景就是点对点的消息传输通道，然后非常激进的使用内存来提升整体的系统性能，这样做虽然标称的tps都能达到很高，但这种设计的思路是很难符合大规模分布式场景的实际需要的。
• 在实际的分布式场景中，这样的系统会存在着较大的应用场景瓶颈，在后端有大量消费者的前提下，消费者出现问题是个非常常见的情况，而消息系统则必须能够在后端消费不稳定的情况下，仍然能够保证用户写入的正常并且TPS不降，是个非常考验消息系统能力的实际场景。
• 也因为如此，在Notify的整体设计中，我们最优先考虑的就是消息堆积问题，在目前的设计中我们使用了持久化磁盘的方式，在每次用户发消息到Notify的时候都将消息先落盘，然后再异步的进行消息投递，而没有采用激进的使用内存的方案来加快投递速度。
• 这种方式，虽然系统性能在峰值时比目前市面的MQ效率要差一些，但是作为整个业务逻辑的核心单元，稳定，安全可靠是系统的核心诉求。

2. 无单点，可自由扩展点设计

Notify领域模型

notify消息领域对象的设计如下图所示，每条消息记录包含messageId（这个就是key）、topic、body(二进制)、header（用户自定义kv集合）、commited(事务状态，是否提交)、Fail target(记录失败的消费者id，消息重试要用到)。

Notify启动流程

1. Notify Server服务器设置一个固定的订阅关系（终端或客户端），例如 A ： B
2. 消息消费 Server初始化中设置参数为B，那么消息消费 Server启动时，会携带服务器信息发送到Notify Server上，在Notify Server中，B就关联了当前消费服务器
3. 消息发送Server初始化时设置参数A，那么消息发送Server初始化时在Notify Server上通过A可以获得对应B关联到服务器
4. 消息发送服务器发送消息到Notify Server，Notify根据订阅关系将当前消息分发到不同的消息消费服务器

Notify普通消息处理流程

1. 当notify收到消息后，先进行持久化，在kv存储增加一条记录，然后ack给生产者；
2. 持久化后执行exchange，这一步会对消费者Subscription的属性过滤规则进行计算，得到目标消费者id集合，找到消费者channel直接push。push完后，内存中维护callback等待消费者ack（成功或者失败），如果没有ack，callback默认等待5秒，作为消费超时处理。最多等待5秒后，callback收集到所有ack，如果全部成功，那么直接删除消息记录；如果部分失败，对消息记录进行修改，主要是修改fail target字段，记录失败消费者id。
3. notify会有后台任务扫描未删除记录，重新投递给失败的消费者id，再次执行4、5、6步骤。

Notify事务消息处理流程

** 对于事务消息，区别之处在于消息生产环节**。生产者先发送half消息，当notify收到half消息后，先进行持久化，在kv存储增加一条half（即commited字段为false）记录，然后ack给生产者，此时并不会触发推送消费者事件。生产者half发送成功后，执行本地事务，事务执行成功后，发送异步commit请求。notify收到commit请求后，将消息的commited字段更新为true，然后就立马执行推送消费者的流程，消费相关逻辑和普通消息无异。

注意点 ： notify事务消息，第5步commit是异步的吗？ commit之后第6步 notify才写KV，如果写失败了，生产者的本地事务还是会提交的。这样的话怎么保证消息会发出去呢？是因为会定期扫描没有commit的数据，然后回调生产者的接口询问数据是否需要comit吗？
1 第5步commit是异步的
2 会扫描未commit的消息回查生产者，达到最终一致

存储服务选择

从07年到现在，notify采用的kv存储模块经历过多次改造。最早notify的消息存储是采用本地文件存储的，参考了ActiveMQ的kaha实现了单机kv存储引擎，在07年12月上线，支撑直冲业务，日均500w消息。然而考虑到交易消息重要性，消息要保证不能丢失，第一个版本的单机存储引擎没有经过多年生产环境的验证，不够成熟，另外也没有多副本机制，无法满足交易对消息可靠性的严苛需求。到了第二个版本，notify开始采用当时最稳定的关系型数据库oracle来做消息存储，在08年3月份发布，支撑核心交易链路，日均1000w消息。随后集团开始启动去O战略，数据库存储需要全面从oracle迁到mysql，为了弥补mysql和oracle的性能和稳定性差异，notify实现了多点高可用mysql存储集群，全部迁移到mysql存储，10年4月完成。由于多点mysql在存储层具备了无状态横向扩容的能力，在稳定性和性能方面许久未遇到瓶颈，所以一直使用了很多年。

图5 ： Notify系统组成结构

• 消息发送集群：主要是业务方的机器，这些机器上是没有任何状态信息的，可以随着用户的请求量增加而增加或减少业务发送方的机器数量，从而扩大或缩小集群能力
• 配置服务集群（Config Server） ： 这个集群的主要目的是动态感知应用集群，消息集群机器上线与下线，并且及时广播给其他集群。如当前业务接受消息的机器下线时，config server会感知到机器下线，从而将该机器从目标用户组踢出，并通知notify server，notify server在获取到通知之后，就可以将已经下线的机器从自己的投递目标列表中删除，这样就可以实现机器的自动上下线扩容
• 消息服务器集群（Notify Server） ： 也就是真正承载消息发送和消息接受的服务器，也是一个集群，应用发送消息时可以随机选择一台机器进行消息发送，任意一台server挂掉，系统都可以正常运行。当需要增加出林能力时，只需要增加notify Server就可以了
• 存储（Storage） ： Notify存储集群有多种不同的实现方式，以满足不同的应用的实际需求。针对消息安全性要求高的应用，我们会选择使用多份落盘的方式存储消息数据，而对于要求吞吐量而不要求消息安全的场景，我们则可以使用内存存储模型的存储。所以，所有的存储也被设计为随机无状态写入存储模型以保障可以自由扩展。
• 消息接收集群 ： 业务方用于处理消息的服务器集群，上下线机器时也能动态的被config server感知，从而实现机器的自动扩展。