对账流程
对账流程节点：数据抓取、数据清洗、对账、结果汇总4个操作节点；
对账结果：上游单边（订单上游存在，下游不存在）、下游单边、金额不一致（上下游订单都存在，但是金额不同）；
对账操作节点功能简介
- 数据抓取：拉取待对账业务数据，生成原始凭证数据；
- 数据清洗：原始凭证数据清洗，生成对账凭证数据；
- 对账：对账凭证对账，记录对账结果；
- 汇总：对账结果汇总；
系统设计需要处理的问题
- 数据量较大时，要分组处理；
- 数据分组时，避免相同单号的分到不同的组，导致数据单边；
- 应用多实例部署，避免多实例并发冲突，避免数据重复处理或数据遗漏；
- 对账流程，可以个性化配置，不同业务需要的对账流程可以不同；
- 对账链路中，任意节点的数据，都可以沿着对账链路往下执行；
系统实现方案
- 数据分表存储，避免后期数据量大了之后再分表时的数据迁移问题，并对数据进行了分组；
- 使用task处理方式，结合分布式定时任务组件xxljob，避免多实例冲突及数据重复处理问题（按表维度创建task任务）；
- task处理流程进行模板化，模板方法抽象task处理的公共流程；
使用到的设计模式
- 责任链模式；
- 模板模式；
责任链模式
- 对于对账流程，目前的对账分为抓取、清洗、对账、汇总4个功能节点；
- 责任链模式灵活组装、增减功能节点的流转顺序和个数；
- 任意节点开始，进入链路，都可以继续链路的流转（常规代码实现的责任链不支持该需求）；
模板模式
- 依赖task触发数据处理；
- task控制任务的并发处理；
代码结构
模板模式代码继承关系
- BaseTask<T>：基础模板抽象类，描述了Task任务处理的流程（execute方法），及流程处理过程中的抽象方法；
- task解锁方法：unlockTask();
- task获取方法：getTasks();
- task分片方法：currentPartition();
- task执行方法：process();
- task执行前处理方法：beforeProcess();
- task执行后处理方法：afterProcess();
- BaseExeTask：继承了基础抽象类BaseTask，并具体实现了task流程处理过程中的抽象方法（除process()之外的方法）；
- BaseReconcileTask：对账操作基础抽象类，其中放置对账的基础操作（目前的对账只一种，所以该类为空）；
- ReconcileDefaultTask：具体对账实现类，实现process()，对账逻辑在这里实现；
图中是对账模块的继承关系
image
责任链各节点核心处理流程介绍
- 数据抓取
- 提供目标业务数据分组功能，确定数据抓取的范围，并创建对应数量的task来分组抓取业务数据；
- 提供目标业务数据按组抓取功能，每个task可以按照上面的分组信息抓取指定范围的业务数据；
- 针对多个对账目标字段的情况，组装目标字段为json，便于数据清洗时拆分；
- 数据清洗
- 按照物理表维度创建清洗task，每个task只处理一个物理分表中的数据；
- 完成字段调整映射，统一同一对账字段名称，统一方式、渠道、状态等属性值；
- 对于多个对账目标字段的情况，拆分一条原始凭证数据为多条对账凭证；
- 对账
- 按照物理表维度创建对账task，每个task只处理一个物理分表中的数据；
- 借助Redis中Set结构的命令：Sdiffstore、Sinterstore函数实现对账操作；
- 对账汇总
- 按照物理表维度创建汇总task，每个task只处理一个物理分表中的数据；
- 每个task统计完该物理表中的汇总数据后，获取redis锁，获取成功后将单个表的汇总结果插入或更新到对账汇总表中；
责任链task节点创建及链路流转过程介绍
- 链路节点及节点的顺序信息存储在数据库中，可以针对每个对账类型创建一个链路，实现对账流程的高度配置话和自由度；
- 每次对账执行，创建批次信息，批次信息中记录当前对账流转到的节点及该节点下创建了多少个task；
- 任务开始执行，创建批次信息，task读取到批次信息后，读取对应的链路配置信息，开始创建第一个节点的task数据，并在批次中记录节点类型及对应task数量；
- 各个节点对应的task定时扫描表中对应的task，执行各自的任务（按照目标方法中规定的流程执行）；
- 批次对应的task任务出发批次执行，校验该批次当前节点下所有的任务是否都已经执行完成，完成后创建下个节点的任务，未完成则根据规则重试；
对账流程图
image
