作 者 | 杜圣埃卢瓦(悬衡)
编者按：导入最新消息堆栈能协助我们解耦销售业务方法论，提高操控性，让主信道更为明晰。但是最新消息信道的标识符堕落和连续性问题也给销售业务带来了很多困扰，责任编辑阐述了钉钉审核最新消息信道解构的结构设计和解决方案。注：Metaq 是阿里 RocketMQ 最新消息堆栈的内部网版本。
概述
导入最新消息堆栈能协助我们解耦销售业务方法论，提高操控性，让主信道更为明晰。Metaq 也确实可靠，重传监督机制能保障足够的连续性。
钉钉审核将审核中的关键事件，比如说审核单发动，各项任务开始，各项任务完结以及审核单完结之类作为最新消息发布出去，将审核的主体流程和邻近销售业务明晰地分开。
但是历经数年的商品插值，邻近销售业务愈来愈多，最新消息信道愈来愈繁杂（参见 “最新消息信道的幸福虚幻与残暴现实生活” 段落）：
1.不同的销售业务方法论拼凑在一同互相影响，单个最新消息处理方法就能有上绝情标识符。
2.因为方法论太繁杂无法同时实现幂等，只能放弃重传，不一致问题严重。每一销售业务都必须附加开发繁杂的收款，补偿监督机制才能防止Q1517A。
钉钉审核每秒钟单厢数十万条最新消息的收到，从监视能窥见平均每秒钟单厢有余条最新消息失利（有的是时候会有百条），失利率大约 0.5%。
通过责任编辑的最佳课堂教学，将能同时实现：
1.将拼凑在一同的方法论分拆成一个个销售业务 Listener 类（参见 “朴素的分拆想法” 段落）；
2.布季谢销售业务 Listener 失利时，能同时实现失利销售业务精确重传，而不是蛮横地全部重传（参见 “精确重传” 段落）；
3.高操控性地构建 Listener 的标准化语句，降低读扩散，并且防止其随著插值堕落（简述 “标准化语句” 段落）；
4.最终，责任编辑的课堂教学不需要附加的存储，也不需要建立附加的 Consumer，原来的基础设施能直接F83E43Se；
通过右图监视能窥见，历经责任编辑方案的治理，最新消息信道每晚多于非常零星的失利（约几十个），而审核每晚要发送一亿以上的最新消息，失利率多于约 0.000005%。
图中多于顶点处存有失利，其他都是 Sunfire 的自动连线，不存有失利。
最新消息信道的幸福虚幻与残暴现实生活
幸福虚幻
最新消息堆栈在结构设计之初就给销售业务规划好了一条成功之路：
主销售业务信道作为 Producer 收到最新消息
十多个甚至更多 Consumer 订户该最新消息，分别执行自己快速且幂等的原子方法论
在这种结构设计下，每一订户者的方法论相互隔离，互不影响；能独立地重传，保证自己的连续性。
残暴现实生活
现实生活并不雷米雷蒙县上面虚幻的那样，随著销售业务发展多半会出现一些 “大泥球” 顾客。
以钉钉审核为例，为了优化商品体验，在审核单发动后，还要消费审核单发动最新消息，触发器做一系列的事，比如说发最新消息通知，同步浏览器，更新提示红点之类，这些功能随著商品插值只会越叠越多，最终成为一个同时做蒂卢事的 “大泥球” 顾客。
大泥球 Consumer Vertus和用户都有巨大的损害：
方法论相互影响，修改风险高；
信道脆弱，容易中断，一个调用失利，后续所有方法论将不会执行；
没有重传：大泥球无法做到原子和幂等，整体重传代价太大，所以直接触发器执行放弃重传
最新消息堆栈引以为傲的重传功能反而会成为故障的温床，导致雪崩。
为什么不直接把大泥球分拆成前面的多个 Consumer 呢？这确实也是一种方案，但是对于大泥球 Consumer，可能会拆出几十个 Consumer，这会导致非常严重的读扩散。举个例子，审核单发动的最新消息中只含有审核单的 id，内容需要从数据库反查，原本在“大泥球”中，只需要查询一次就F83E43Se，而分拆后可能要多查几十次。这还只是众多扩散问题的其中一个，如果为了治理大泥球，却加重了扩散问题，就得不偿失了。
朴素的分拆想法
最简单的想法就是把大泥球中相关的方法论聚合到一同，不相关的方法论隔离，拆成一个个独立的销售业务处理器，俗称 “高内聚低耦合”。
在一个审核单的生命周期中会有很多种类的最新消息收到，比如说审核单发动，审核单完结，审核各项任务生成，审核各项任务完成，审核抄送之类，我们能将这些将这些最新消息的处理方法论聚合到一个销售业务处理器中，只需要同时实现这个接口的 onXxx 方法，就能将方法论嵌入整个审核的生命周期中。这个销售业务处理器接口在审核中就是 ProcessEventListener:
public interface ProcessEventListener extends EventListener {
/*** 审核单发动事件*/void onProcessInstanceStart(InstanceEventContext instanceEventContext);
/*** 审核单完结事件*/void onProcessInstanceFinish(InstanceEventContext instanceEventContext);
/*** 审核各项任务生成事件*/void onTaskActivated(TaskEventContext taskEventContext);
// 省略其他事件// ...}
通过在框架层面 catch 异常就能防止信道因为单个销售业务的失利而折断，用伪标识符表述如下：
// 接到审核单发动最新消息InstanceEventContext instanceEventContext = buildContext();for (handler : handlers) {try {handler.onProcessInstanceStart(instanceEventContext);} catch (Exception e) {// 打印监视日志之类// ...}}
现在虽然方法论内聚了，信道更为健壮了，但是还有很多技术上的问题没有解决：
某个处理器因为网络超时失利了，如何重传？：我们仅仅是将方法论拆开了，执行的时候还是一串 “大泥球”，如果仅仅依靠最新消息堆栈本身监督机制，要重传只能一同重传，这显然无法满足诉求；
如何高操控性地构建庞大的标准化语句（即
Context
参数）：为了满足众多处理器对数据查询的诉求，需要提供庞大的语句，除了操控性风险外，也是标识符腐败的温床
我们先来看看第一个问题。
精确重传
当处理器因为超时等意外情况失利时，如果销售业务重要的话，都需要补偿重传。但是不同处理器对重传的诉求可能不同，为此我们在上一节朴素想法的基础上，还同时实现了多种重传策略，并且支持单独指定最大重传次数：
Ignore：非重要销售业务，失利就算了，不需要重传；
Concurrent：在另外线程池中并发执行，也不会重传。适合一些容易影响后续执行的长耗时的处理器；
Retry Now：立即重传。会将各项任务放到本地的一个延迟堆栈中，100~500ms 后重传。适合时效性比较强的处理器；
Retry Later：重投最新消息，精确重传失利的处理器，遵从 Metaq 的重投延迟，前三次重传分别是
1s 5s 10s
，因此不适合时效性强的处理器；
为了方便使用，我们将这些策略做成了注解的形式，比如说审核抄送时效性没这么强，能使用 Retry Later 策略：
public class CcListener implements ProcessEventListener {// Retry Later 策略, 最多重传两次@Policy(value = PolicyType.RETRY_LATER, retry = 2)@Overridepublic void onProcessInstanceStart(InstanceEventContext instanceEventContext) {//... 方法论省略}}
再比如说审核待办同步，是时效性比较强的，间隔太久容易有时效性问题，能使用 Retry Now 策略：
public class SyncTodoTaskListener implements ProcessEventListener {// Retry Now 策略, 最多重传三次@Policy(value = PolicyType.RETRY_NOW, retry = 3)@Overridepublic void onProcessInstanceStart(InstanceEventContext instanceEventContext) {//... 方法论省略}}
前三种策略都比较好理解，就不多说了。下文重点讨论最终一种策略是如何同时实现的。
要想同时实现精确重传，只需要记下每一处理器的执行状态，比如说重传到第几次，是否成功之类。这种流水数据使用关系数据库记会比较重，因此我们利用 Metaq 提供的 UserProperty，将各个销售业务处理器的执行状态以 json 的形式存储在 RETRY_STORE 这个 UserProperty 中，然后重投到另一个专门的重传 topic 中。
处理器的执行状态存储格式如下：
{// 总体重传次数, 第一次重传(第 0 次代表正常执行)"globalCnt": 1,// 每一处理器的执行状态"cntMap": {// handler1 第 1 次重传, 读取该属操控性判断 handler1 是否还有重传机会"handler1": 1,// -1 表示 handler2 已经执行成功, 不需要再执行"handler2": -1,// -2 表示 handler3 已经彻底执行失利(一般是超过了设置的最大重传次数), 不需要再执行"handler3": -2}}
正常执行完结后，如果存有失利的处理器，并且没有达到最大重传次数，会生成上面这种格式的执行调查报告，存放到 RETRY_STORE 这个 UserProperty 中再重投最新消息，简化标识符如下：
Message message = new Message();// 重投到专门的重传主题message.setTopic("my-retry-topic");// 最新消息体保持不变message.setBody(preBody);// nextCnt 是重传的次数// 设置 DelayTimeLevel 能让重投有一定的延时message.setDelayTimeLevel(nextCnt);// 将本次执行状态存储到 user property 中message.putUserProperty("RETRY_STORE", "{\"globalCnt\":1,\"cntMap\":{\"handler1\":1,\"handler2\":-1,\"handler3\":-2}}");// 发送最新消息mqPublishService.send(message);
虽然我们另起了一个专门重传用的 topic，但是顾客的方法论跟原 topic 是完全一样的，理论上直接重投原来的 topic 也是能的，但还是分离开更为安全。
总体思路如图：
图中的重投都是指往 topic 再发一遍最新消息，而不是 Metaq 自身的重投监督机制。
其中有一些细节问题需要注意，比如说某次发布中上线了一个新的处理器，而发布机器刚好接到了一个重传最新消息，如何才能防止新处理器被意外执行呢？我们之前记录的执行状态 cntMap 中的所有 key 就相当于首次执行时的处理器快照，我们重传时只执行 cntMap 中存有的 key 就好了。
另外 Metaq 的 UserProperty 大小也不是没有限制，它最多能存储 30KB 的数据，粗略计算一下大约能存储 1500 个处理器的状态，这对大多销售业务来说都是绰绰有余的了。
标准化语句
另一个问题则是由于我们的编码结构导致的，每一处理器都接收相同的语句参数（XxxContext）处理销售业务，这种 "语句 + 处理器" 的结构在销售业务系统上很常见，但是它存有几个问题：
为了满足所有处理器的需求，语句往往会很庞大，因此构建操控性差。
外部无法感知处理器内部需要使用语句的哪些字段，只能一股脑地将所有字段都填充好，传递进去，而且内部很有可能一个字段都不使用，白白损耗了操控性。
语句中存有一些幽灵字段，在某个处理器中设置进去，又在某几个处理器中读取，也就是它有时候为 null，有时候又有值，维护难度巨大，从中取个值都要战战兢兢。
读扩散问题：每一处理器都去读相同的数据，导致信道数十倍的读扩散。
懒加载
前两个问题能通过导入懒加载监督机制解决，对于语句中操控性损耗比较大的字段，在读取时再进行加载。这样语句的构建就非常快了 ，也不会去附加加载处理器中用不到的字段。
懒加载的思路非常简单，就是第一次读取字段时再通过远程调用，SQL 等操控性消耗较大的方式获取字段值，获取之后在内存缓存起来，后续直接从内存返回。但是在编码结构设计上有一些小技巧，能让它用起来跟普通字段一样，为此我们开发了一个 Lazy 框架，用来将这部分懒加载方法论封装起来，举个例子，我们将 User 实体的部门属性懒加载化：
public class User {// 用户 idprivate Long uid;// 用户的部门，为了保持示例简单，这里就用普通的字符串// 需要远程调用 通讯录系统 获得private final Lazy department;
public User(Long uid, Lazy department) {this.uid = uid;this.department = department;}
public Long getUid() {return uid;}
public String getDepartment() {return department.get();}}
// 构建 User 实体Long uid = 1L;User user = new User(uid, Lazy.of(() -> departmentService.getDepartment(uid)));
// 使用 User 实体，部门属性用起来和普通属性一样user.getDepartment();
Lazy 框架的具体同时实现能参考另一篇文章 利用 。
朴素的懒加载思想在依赖服务正常的情况下能大大减少读扩散和提高操控性，但是在依赖服务大规模异常时，还是会造成读扩散，进一步加大依赖服务的压力：
因此我们对懒加载监督机制进行了优化，支持设置最大调用次数。超过调用次数，直接从内存返回失利，防止在依赖服务大规模失利时还对其制造压力。
我们给 Lazy 框架添加一个状态机监督机制，每次 get 调用，Lazy 单厢根据当前状态进行一次状态转移，在 “已加载” 状态时直接从内存返回数据，处于 “失利” 状态时，则直接从内存返回失利：
在新的监督机制下，如果依赖服务大规模异常，则能防止大量的读扩散：
最终，我们能通过一个懒加载字段计算出另一个懒加载字段，比如说用户实体中，部门是懒加载的，而用户的主管又是要通过部门计算出的，产生了 用户 -> 部门 -> 主管 这样的层次加载结构，如下：
// 通过用户获得部门Lazy departmentLazy = Lazy.of(() -> departmentService.getDepartment(uid));// 通过部门获得主管// department -> supervisorLazy supervisorLazy = departmentLazy.map(department -> SupervisorService.getSupervisor(department));
在层次加载结构中，一旦某个节点被求值，路径上所有的是属性单厢被缓存，因为这种对象能自动地优化操控性。以 这篇文章中 User 实体为例：
下文中我们会发现，审核的销售业务特点就是层次结构鲜明，因而非常适合这种结构设计。
基于销售业务的语句结构设计
我们常常出于技术操控性而不是销售业务去结构设计语句中的字段，这是语句堕落的根本原因。
什么叫 “基于技术” 结构设计字段？举个例子，之前审核也有类似的 语句 + 处理器 的架构。但是因为不知道处理器中会用哪些字段，为了优化语句的操控性，在语句中只放置了一个审核单 id，所有数据都需要在处理器中附加查询，造成了严重的读扩散。为了优化操控性还容易导入幽灵字段，幽灵字段在语句中的初始值为 null，处理器在需要时才将其设置进去。
通过字段懒加载监督机制，销售业务不再需要妥协于技术，能将字段放在它销售业务上 “应该在” 的地方。字段放在实体中的原因，不是为了方便取用，也不是为优化操控性，而是销售业务上它就应该在那里。这也是 DDD 的核心思想。
审核销售业务的特点就是层次鲜明，一个审核单由多个活动组成，而一个活动由多个各项任务组成。
根据这个特点，我们针对不同的审核事件最新消息结构设计了三种语句：
审核单实例语句
审核单发动最新消息
审核单完结最新消息
审核单撤销最新消息
活动语句
活动开始最新消息
活动完结最新消息
各项任务语句
各项任务开始最新消息
各项任务完结最新消息
各项任务取消最新消息
语句全部结构设计成不可变的，不允许一个处理器设置字段，另一个处理器又去读取字段的情况，如果实在需要，说明这两个处理器是耦合的，那么合并成一个处理器更为合适。
至此，我们将字段都放在销售业务上应该在地方，开发者只要根据自己对销售业务的理解就能一层层地找到字段，并且肯定能获取到，不会有的是时候存有，有的是时候又不存有。
如何防止雪崩
在最新消息信道执行这么一大串处理器，如果发生雪崩还是挺危险的。好在本方案不依赖 Metaq 自身的重传监督机制，直接不处理任何重复/重传的最新消息，就能规避无限重传的问题。
根据踩坑的经验，Metaq 发生雪崩的主要原因在于特定场景下的无限重传：
rebalance 导致重传次数归 0；
顾客执行超时，重传次数从 3 开始继续重投；
所以我们只要不处理重复最新消息，也不处理重传最新消息，同时做好 Metaq优雅上下线，就能防止雪崩。
不处理重复最新消息：通过 ${msgId}_${reconsumeTimes} 设置一个 tair 分布式锁，遇到重复最新消息就直接返回，不进行处理。根据最新消息量设置过期时间长短，太长可能会占用很多的 tair 空间，但是防止雪崩的效果会更好。这个地方也能使用布隆过滤器进行优化。
private boolean isDuplicateMessage(Message msg) {try {Integer consumedCount = ltairManager.incr("dingflow_mq_consume_" + msg.getMsgId() + "_" + msg.getReconsumeTimes(), 1, 30);if (consumedCount != null && consumedCount > 1) {return true;}} catch (Throwable throwable) {// 打印错误日志// ...}
return false;}
不处理重传最新消息：reconsumeTimes 大于 0 直接返回，不处理重传最新消息。因为我们不依赖 Metaq 层面的重传监督机制。
Show Me The Code: 框架编码结构设计
框架的整体结构设计遵循 中的原则，将副作用和核心方法论完全抽离。框架的主体部分完全与最新消息堆栈无关，而是抽象出了一个通用的 RemoteRetryStore，用于持久化地存储处理器重传状态。这样我们不仅仅能基于最新消息堆栈重传，也能基于数据库进行重传，只要传递不同的 RemoteRetryStore 即可。框架在本地就能模拟各种异常情况进行单元测试，文中就不大段罗列标识符了。
监视与感知
有了上面的课堂教学，监视与感知方式基本是不言自明的，只需要在框架层面 catch 异常，在有剩余重传机会时打印重传日志，无重传机会时打印失利日志。结合 Sunfire 的 Top 监视，即可看到每一处理器的失利情况。
如图，OpenEventCallbackListener 处理器因为 NPE 问题，失利量突然大幅度增加。
总结
虽然这篇文章写了很多，但是总体思路非常简单，概括起来多于三点：
将最新消息信道分拆成多个处理器；
利用 Metaq 的 UserProperty 存储每次处理器的执行状态，精确重传失利的处理器；
用懒加载监督机制构建标准化语句，提高构建操控性，降低读扩散，最终结构设计出最符合销售业务的语句；
我一直坚信好的结构设计一定是简单而深刻的，编码过程中必然会遇到大量结构设计中没有的是细节问题，因此多于简单的结构设计才能起到指导编码的作用，不然要么难以编码，要么一堆 bug; 也多于对问题的深刻理解，才能找到隐藏在背后的简洁模型。这和我另一篇 并不矛盾，它批判的是“多于简单”，而“没有对问题深刻理解”的结构设计，比如说前文中的“大泥球 Consumer” ，那就是结构设计不足了。
还木有评论哦，快来抢沙发吧~