BFPS-AP 文件处理系统应用平台

一、系统概述

1.1 基本信息

项目	说明
Artifact	`cn.webank.cnc.bfpsap:bfps-ap`
父POM	`cn.webank.cnc.sdk:zcnc-sdk-parent-for-coe:3.37.0-SNAPSHOT`
版本	`pace_replacement_version-SNAPSHOT`
描述	Application Platform for File Process System（文件处理系统应用平台）
打包方式	父POM为 `pom`，子模块为 `jar`
主启动类	`cn.webank.cnc.bfpsap.BFpsApApplication`
系统ID	`U636`
数据库名	`ti_bfpsapdb`（TiDB）
框架基础	Spring Boot + ZCNC-SDK（微众银行CNC基础开发框架）

1.2 系统定位

BFPS-AP（Batch File Process System - Application Platform）是微众银行CNC（Core banking Next-generation Cloud）架构下的批量文件处理应用平台。该系统负责大文件的**按DCN拆分（Split）和按DCN合并（Merge）**两大核心业务，是银行分布式架构中跨DCN批量数据同步的关键基础设施。

1.3 核心SDK依赖

依赖	用途
`zcnc-sdk-starter-core`	核心基础框架
`zcnc-sdk-starter-db`	数据库连接（TiDB/MyBatis）
`zcnc-sdk-starter-io`	IO/FPS文件服务客户端
`zcnc-sdk-starter-rmb`	RMB消息总线（WeMQ）
`zcnc-sdk-starter-seq`	分布式序列号生成
`zcnc-sdk-starter-flowcontrol`	流量控制
`zcnc-sdk-starter-trace`	链路追踪

二、模块结构

2.1 整体模块关系

bfps-ap (parent pom)
├── bfps-ap-common    # 公共模块：DTO、枚举、工具类
├── bfps-ap-dao       # 数据访问层：Entity、DAO接口、MyBatis映射
└── bfps-ap-service   # 业务服务层：核心业务逻辑、RMB消息处理、配置

2.2 模块依赖关系

bfps-ap-service --> bfps-ap-dao --> (无内部依赖)
bfps-ap-service --> bfps-ap-common --> (无内部依赖)
bfps-ap-dao --> zcnc-sdk-starter-core
bfps-ap-common --> zcnc-sdk-starter-core

2.3 各模块说明

bfps-ap-common（公共模块）

职责：定义系统通用的DTO、枚举、工具类和基础服务接口
特点：无数据库依赖，纯Java逻辑，可被所有模块引用
关键类：BaseFileSplitDto、BaseFileMergeDTO、SplitFileService、各类枚举

bfps-ap-dao（数据访问层）

职责：数据库表实体定义、DAO接口、MyBatis映射文件
特点：使用MyBatis代码生成插件（zcnc-sdk-starter-mybatisplugin）自动生成
数据源：TiDB（兼容MySQL协议），数据库名 ti_bfpsapdb
关键类：6个Entity类、6个DAO接口、6个MyBatis映射文件

bfps-ap-service（业务服务层）

职责：包含系统启动入口、核心业务逻辑、RMB消息订阅/发布、外部系统集成
特点：Spring Boot应用，通过RMB消息总线驱动业务流程
关键类：AbstractSplitService、AbstractMergeService、SplitFileByDCNCallable、GnsService、FpsService

三、包架构（完整包树）

cn.webank.cnc.bfpsap
├── BFpsApApplication.java                    # Spring Boot启动类
│
├── common/                                    # 服务层公共包
│   ├── constant/
│   │   ├── FpsApConstants.java               # 系统常量定义
│   │   └── SeqTableName.java                 # 序列表名常量
│   └── util/
│       ├── CheckUtil.java                    # 校验工具类
│       ├── CommonUtil.java                   # 通用工具类（MD5、空文件判断）
│       ├── EnvService.java                   # 环境变量服务
│       ├── JavaTypes.java                    # Jackson JavaType常量
│       └── MapUtils.java                     # Map工具类
│
├── config/
│   ├── FpsApConfig.java                      # 系统配置（文件路径、线程池参数）
│   └── TaskExecutorConfig.java               # 线程池配置
│
├── service/
│   ├── base/
│   │   ├── FpsService.java                   # FPS文件服务封装
│   │   ├── GnsService.java                   # GNS路由服务实现
│   │   └── IGnsService.java                  # GNS服务接口
│   │
│   ├── dao/                                  # DAO服务封装层
│   │   ├── TmFpsApSplitConfigDaoService.java
│   │   ├── TmFpsApSplitResultDaoService.java
│   │   ├── TmFpsApSplitStatusDaoService.java
│   │   ├── TmFpsApMergeConfigDaoService.java
│   │   ├── TmFpsApMergeResultDaoService.java
│   │   └── TmFpsApMergeStatusDaoService.java
│   │
│   ├── notice/                               # 通知服务
│   │   ├── INotifySplitResultService.java    # 拆分通知接口
│   │   ├── INotifyMergeResultService.java    # 合并通知接口
│   │   ├── CommonNotifySplitResultService.java    # 通用拆分通知
│   │   ├── CommonNotifyMergeResultService.java    # 通用合并通知
│   │   └── NotifySplitResultToMergeService.java   # 拆分转合并通知
│   │
│   ├── process/                              # 核心业务处理
│   │   ├── ISplitService.java                # 拆分服务接口
│   │   ├── IMergeService.java                # 合并服务接口
│   │   ├── AbstractSplitService.java         # 拆分服务抽象基类（核心）
│   │   ├── AbstractMergeService.java         # 合并服务抽象基类（核心）
│   │   ├── CommonSplitServiceImpl.java       # 通用拆分实现
│   │   ├── CommonMergeServiceImpl.java       # 通用合并实现
│   │   └── SplitFileByDCNCallable.java       # 按DCN拆分多线程任务（核心）
│   │
│   └── rmb/                                  # RMB消息订阅
│       ├── SplitFileSubscriberService.java   # 拆分消息订阅者
│       └── MergeFileSubscriberService.java   # 合并消息订阅者
│
├── dao/（bfps-ap-dao模块）                     # 数据访问接口
│   ├── TmFpsApSplitConfigDao.java
│   ├── TmFpsApSplitResultDao.java
│   ├── TmFpsApSplitStatusDao.java
│   ├── TmFpsApMergeConfigDao.java
│   ├── TmFpsApMergeResultDao.java
│   └── TmFpsApMergeStatusDao.java
│
├── dto/（bfps-ap-dao模块）                     # 数据实体
│   ├── TmFpsApSplitConfig.java
│   ├── TmFpsApSplitConfigExample.java
│   ├── TmFpsApSplitResult.java
│   ├── TmFpsApSplitResultExample.java
│   ├── TmFpsApSplitStatus.java
│   ├── TmFpsApSplitStatusExample.java
│   ├── TmFpsApMergeConfig.java
│   ├── TmFpsApMergeConfigExample.java
│   ├── TmFpsApMergeConfigKey.java
│   ├── TmFpsApMergeResult.java
│   ├── TmFpsApMergeResultExample.java
│   ├── TmFpsApMergeStatus.java
│   ├── TmFpsApMergeStatusExample.java
│   └── TmFpsApMergeStatusKey.java
│
└── common/（bfps-ap-common模块）
    ├── dto/
    │   ├── BaseFileSplitDto.java             # 拆分基础DTO
    │   ├── BaseFileMergeDTO.java             # 合并基础DTO
    │   ├── CommonSplitNoticeMessage.java     # 通用拆分通知消息
    │   ├── CommonMergeNoticeMessage.java     # 通用合并通知消息
    │   ├── NoticeResultDto.java              # 通知结果DTO
    │   └── SplitNoticeMergeMessage.java      # 拆分转合并通知消息
    ├── enums/
    │   ├── AppType.java                      # 应用类型枚举
    │   ├── DcnType.java                      # DCN类型枚举
    │   ├── MergeCheckAmtEnum.java            # 合并金额校验枚举
    │   ├── MergeCheckLineCntEnum.java        # 合并行数校验枚举
    │   ├── MergeStatusEnum.java              # 合并状态枚举
    │   ├── MergeUserGrayEnum.java            # 合并灰度枚举
    │   ├── SplitDupCheckEnum.java            # 拆分重复校验枚举
    │   ├── SplitNoDcnEnum.java               # 无DCN处理策略枚举
    │   ├── SplitStatusEnum.java              # 拆分状态枚举
    │   ├── TradeStatus.java                  # 交易状态枚举
    │   ├── dcns/
    │   │   ├── DcnTypeBusinessLineMapEnum.java   # DCN类型与业务线映射
    │   │   └── IdTypeRouteTypeMapEnum.java       # ID类型与路由类型映射
    │   └── logs/
    │       └── LogKeyEnum.java               # 日志消息键枚举
    ├── service/
    │   └── SplitFileService.java             # 文件拆分通用服务
    └── util/
        └── LogUtil.java                      # 日志工具类

四、核心类详解

4.1 BFpsApApplication（启动类）

位置：bfps-ap-service/.../BFpsApApplication.java
注解：@SpringBootApplication、@EnableAspectJAutoProxy、@EnableScheduling
职责：Spring Boot应用入口，启动时加载并打印所有Bean
特点：启用了AOP代理和定时任务调度

4.2 AbstractSplitService（拆分服务抽象基类 -- 核心）

位置：bfps-ap-service/.../service/process/AbstractSplitService.java
泛型：<T extends BaseFileSplitDto>
设计模式：模板方法模式，定义拆分流程骨架，子类只需实现 getSplitDto() 方法
核心方法 execute() 的完整拆分流程：

1. getFileSplitDto(rmbMsg)         -- 从RMB消息解析拆分DTO
2. checkParams()                   -- 参数校验（noDcnFlag与validLineCount）
3. checkDuplicate()                -- 重复性校验（批次号+唯一值）
4. initDbSplitStatus()             -- 初始化拆分状态记录到数据库
5. getDcnList()                    -- 通过GNS获取目标DCN列表
6. verifyDcn()                     -- 校验DCN列表非空
7. downloadFile()                  -- 从FPS下载待拆分文件
8. checkOriFile()                  -- 校验原文件（行数、行长度）
9. physicalSplit()                 -- 物理分片（默认分5片）
10. splitFileByDCN()               -- 按DCN拆分（多线程并发）
11. mergeFileByDCN()               -- 按DCN合并临时文件
12. checkLineCountAfterSplit()     -- 校验拆分前后行数一致性
13. uploadAndNotice()              -- 上传结果文件到FPS并通知下游
    ├── uploadFile()               -- 多线程并发上传DCN文件
    ├── notice()                   -- 通过RMB发送通知消息
    └── recordResult()             -- 记录拆分结果到数据库
14. executeSuccess()               -- 更新拆分状态为成功

关键设计：
- 使用 BeanCopier 高性能复制配置属性到DTO
- uniqueNo = channel|business|batchNo|fileId 作为全局唯一标识
- 物理分片数固定为 PHYSICAL_SPLIT_NUM = 5
- 文件上传使用 CountDownLatch 等待所有DCN文件上传完成
- 拆分异常时更新状态为失败（SplitStatusEnum.SPLIT_STATUS_FAILED）

4.3 AbstractMergeService（合并服务抽象基类 -- 核心）

位置：bfps-ap-service/.../service/process/AbstractMergeService.java
泛型：<T extends BaseFileMergeDTO>
设计模式：模板方法模式
核心方法 execute() 的完整合并流程：

1. getFileMergeDto(msg)            -- 从RMB消息解析合并DTO
2. checkParam()                    -- 参数校验（必填字段、重复性、配置完整性）
3. getDcnList()                    -- 获取DCN列表（支持灰度DCN过滤）
4. checkDcnNo()                    -- 校验请求DCN是否在合法列表中
5. insertFileMergeStatus()         -- 记录合并状态（等待合并）
6. shouldDoMerge()                 -- 判断所有DCN消息是否到齐（竞争获取合并任务）
7. downloadDcnFile()               -- 多线程下载所有DCN文件
8. merge()                         -- 按DCN顺序合并文件
9. calMergeInfo()                  -- 计算合并后信息（金额/行数/MD5）
10. checkLineCountAfterMerge()     -- 校验合并前后行数一致性
11. uploadFile()                   -- 上传合并后文件到FPS
12. noticeMergeResult()            -- 通知下游
13. recordResult()                 -- 记录合并结果到数据库
14. updateMergeStatus()            -- 更新合并状态

关键设计：
- 消息汇聚模式：每个DCN发送一条合并消息，系统需要等待所有DCN消息到齐后才执行合并
- 竞争合并：多线程环境下通过数据库 competitionGetMergeTaskByBatchNo() 原子竞争合并任务
- 灰度支持：isGray=1 时，仅合并灰度DCN列表中的DCN文件
- 金额校验：支持整数金额（checkAmt=1）和BigDecimal金额（checkAmt=2）两种校验模式

4.4 SplitFileByDCNCallable（按DCN拆分多线程任务 -- 核心）

位置：bfps-ap-service/.../service/process/SplitFileByDCNCallable.java
作用域：@Scope(PROTOTYPE) 多例模式，每次使用创建新实例
实现：Callable<Long>，返回处理行数
核心逻辑 call()：

1. 逐行读取分片文件
2. 每100行批量处理（batchProcess）
3. 批量处理流程：
   a) 提取每行的用户标识（userNo）
   b) 构建 GNS 查询请求列表
   c) 批量调用GNS查询用户所在DCN
   d) 根据DCN结果将行数据分发到对应DCN文件
   e) 追加写入目标DCN临时文件

GNS限流与重试机制：
- 使用 FlowController 进行流量控制
- 最大重试次数：10000次
- 告警阈值：1000次
- 重试等待时间使用对数函数控制：sleepTime = log(retryTimes) * 100
- 最大重试等待时间约3700秒
两种拆分模式：
1. 按用户标识拆分（idType != "dcn"）：通过GNS查询用户所在DCN进行路由
2. 按行内容DCN字段拆分（idType == "dcn"）：直接从文件行中提取DCN编号

4.5 GnsService（GNS路由服务）

位置：bfps-ap-service/.../service/base/GnsService.java
实现：IGnsService 接口
核心方法：
- getDcnListByDcnType(businessIdOrDcnType) -- 根据DCN类型或业务线获取DCN列表，最多重试5次，每次间隔1秒
- getDcnByRouteType(businessId, gnsQueryRequestList) -- 批量查询用户所在DCN，最多重试5次
- getAdmDcnNo() -- 获取管理DCN编号
底层调用：GnsResolver（来自zcnc-sdk-rmb）

4.6 FpsService（FPS文件服务）

位置：bfps-ap-service/.../service/base/FpsService.java
Bean名称：fpsApService
核心方法：
- get(fileId, localFilePath, hashValues) -- 从FPS下载文件到本地
- put(fileTypeEnum, localFilePath) -- 上传本地文件到FPS，返回 FileHashInfo（含fileId和hashValues）
底层调用：FpsClientUtil（来自zcnc-sdk-io）

4.7 SplitFileSubscriberService（拆分消息订阅者）

位置：bfps-ap-service/.../service/rmb/SplitFileSubscriberService.java
实现：MessageListener 接口
注解：@RmbMap(serviceId = SERVICE_ID_SPLIT_FILE_SUBSCRIBE, scenario = "01")
消息路由：从消息体中解析 serviceCode，路由到 SPLIT_SERVICE_{serviceCode} 实现类
默认实现：SPLIT_SERVICE_COMMON -> CommonSplitServiceImpl

4.8 MergeFileSubscriberService（合并消息订阅者）

位置：bfps-ap-service/.../service/rmb/MergeFileSubscriberService.java
实现：MessageListener 接口
注解：@RmbMap(serviceId = SERVICE_ID_MERGE_FILE_SUBSCRIBE, scenario = "01")
消息路由：从消息体中解析 serviceCode，路由到 MERGE_SERVICE_{serviceCode} 实现类
默认实现：MERGE_SERVICE_COMMON -> CommonMergeServiceImpl

4.9 FpsApConfig（系统配置类）

位置：bfps-ap-service/.../config/FpsApConfig.java
职责：管理文件操作相关的本地目录路径
配置目录：
- localDirPath -- 拆分本地根目录
- originalFileDirPath -- 拆分原始文件目录
- splitFileDirPath -- 拆分分片文件目录
- tmpFileDirPath -- 拆分临时文件目录（按DCN拆分后的中间文件）
- resultFileDirPath -- 拆分结果文件目录（按DCN合并后的最终文件）
- merge_originalFileDirPath -- 合并原始文件目录
- merge_resultFileDirPath -- 合并结果文件目录
初始化：@PostConstruct 中基于 APPHOME 环境变量创建所有目录

4.10 TaskExecutorConfig（线程池配置）

位置：bfps-ap-service/.../config/TaskExecutorConfig.java
定义三个线程池：

线程池Bean名	线程数	队列容量	线程名前缀	用途
`splitFileTaskExecutor`	300（可配）	1000	SPLIT-FILE-TASK-EXECUTOR-	拆分任务执行
`commonExecutor`	30（可配）	1000	COMMON-EXECUTOR-	通用任务执行
`fpsOperateExecutor`	128（可配）	10000	FPS-OPERATE-EXECUTOR-	FPS文件操作（上传/下载）

五、数据模型

5.1 数据库表总览

系统共使用 6张数据库表，存储于TiDB数据库 ti_bfpsapdb 中：

表名	说明	主键	用途
`TM_FPS_AP_SPLIT_CONFIG`	拆分配置表	`mapKey` (channel-business)	按channel+business存储拆分参数配置
`TM_FPS_AP_SPLIT_STATUS`	拆分状态表	`fileId`	跟踪每次拆分任务的处理状态
`TM_FPS_AP_SPLIT_RESULT`	拆分结果表	`id` (序列号)	记录拆分后每个DCN文件的结果
`TM_FPS_AP_MERGE_CONFIG`	合并配置表	`(channel, business)`	按channel+business存储合并参数配置
`TM_FPS_AP_MERGE_STATUS`	合并状态表	`(channel, business, batchNo, fromDcnNo, toDcnNo)`	跟踪每个DCN的合并消息接收状态
`TM_FPS_AP_MERGE_RESULT`	合并结果表	`id` (序列号)	记录合并后的结果信息

5.2 拆分配置表（TM_FPS_AP_SPLIT_CONFIG）

字段	类型	说明
`mapKey`	String	主键，格式为 `channel-business`
`channel`	String	渠道标识
`business`	String	业务类型
`enabled`	String	是否启用（Y/N）
`notifyServiceId`	String	通知目标服务ID
`notifyScenario`	String	通知场景
`notifyFlag`	String	通知标识（COMMON / TO_MERGE）
`noDcnFlag`	Integer	找不到DCN时的处理策略（1:忽略, 3:拒绝, 4:强校验）
`fileDesc`	String	文件描述
`dupCheckFlag`	Integer	重复校验标识（1:批次号, 2:唯一值, 3:两者都校验）
`lineLenCheck`	Integer	行长度校验（1:校验, 2:不校验）
`separateType`	Integer	字段分隔方式（1:固定分隔符, 2:指定字段长度, 3:JSON Key）
`idType`	String	用户标识类型（BankCardNo/Wechat/QQ/UserID/BankAccount/EcifNo/IdNo/WeXinOpenId/QQOpenId/SGSystemId/UIF_NUM/dcn）
`theSeparator`	String	字段分隔符
`idFieldNum`	Integer	用户标识字段位置（分隔符模式）
`idFieldStartPos`	Integer	用户标识起始字节位置（长度模式）
`idFieldLen`	Integer	用户标识长度（长度模式）
`startLine`	Integer	有效记录起始行号
`isKeepHeadInfo`	Integer	是否保留文件头信息
`isTranscode`	String	是否转换文件编码
`originalCode`	String	原始文件编码
`destinationCode`	String	目标文件编码
`dcnFieldNum`	Integer	DCN字段位置
`dcnFieldStartPos`	Integer	DCN起始字节位置
`dcnFieldLen`	Integer	DCN字段长度
`isIgnoreSplitNull`	String	是否忽略拆分空值
`toDcn`	String	目标DCN类型
`jsonKey`	String	JSON拆分Key（separateType=3时使用）
`createdDatetime`	Date	创建时间
`lastModifiedDatetime`	Date	修改时间

5.3 拆分状态表（TM_FPS_AP_SPLIT_STATUS）

字段	类型	说明
`fileId`	String	主键，文件ID
`serviceCode`	String	服务编码
`hashValue`	String	文件Hash值
`uniqueValue`	String	唯一值
`channel`	String	渠道标识
`business`	String	业务类型
`batchNo`	String	批次号
`status`	Integer	状态（1:处理中, 99:成功, -1:失败）
`taskIp`	String	执行任务的机器IP
`validLineCount`	Long	有效行数
`noDcnFlag`	Integer	无DCN处理策略
`msg`	String	处理消息
`transParams`	String	透传参数
`dupCheckFlag`	Integer	重复校验标识
`dtoJson`	String	原始DTO的JSON序列化
`startTime`	Date	开始时间
`endTime`	Date	结束时间
性能字段	Long	`physicSplitCost`、`splitByDcnCost`、`mergeByDcnCost`

5.4 拆分结果表（TM_FPS_AP_SPLIT_RESULT）

字段	类型	说明
`id`	String	主键，序列号
`channel`	String	渠道标识
`business`	String	业务类型
`batchNo`	String	批次号
`toDcn`	String	目标DCN编号
`originalFileId`	String	原始文件ID
`originalHashValues`	String	原始文件Hash值
`fileId`	String	拆分后文件ID
`hashValues`	String	拆分后文件Hash值
`noticeBizSeqNo`	String	通知业务流水号
`noticeContent`	String	通知内容
`originalLineCount`	Long	原始文件行数
`lineCount`	Long	拆分后行数

5.5 合并配置表（TM_FPS_AP_MERGE_CONFIG）

字段	类型	说明
`channel` + `business`	联合主键	渠道+业务
`enabled`	String	是否启用
`notifyServiceId`	String	通知服务ID
`notifyScenario`	String	通知场景
`notifyFlag`	String	通知标识
`checkLineCount`	Integer	行数校验（1:校验）
`startLine`	Integer	有效行起始位置
`checkAmt`	Integer	金额校验（1:整数, 2:BigDecimal）
`fileDesc`	String	文件描述
`fromDcn`	String	来源DCN类型
`isGray`	Integer	是否启用灰度（1:启用）
`grayDcnNo`	String	灰度DCN列表（分号分隔）
`fromCrossDcn`	String	跨DCN来源
`separateType` / `theSeparator` / `idFieldNum` / `idFieldStartPos` / `idFieldLen`		金额字段解析配置

5.6 合并状态表（TM_FPS_AP_MERGE_STATUS）

字段	类型	说明
`channel` + `business` + `batchNo` + `fromDcnNo` + `toDcnNo`	联合主键	唯一标识一条合并消息
`serviceCode`	String	服务编码
`fileId`	String	待合并文件ID
`hashValuesJson`	String	Hash值JSON
`transParams`	String	透传参数
`checkLineCount`	Integer	行数校验配置
`validLineCount`	Long	有效行数
`status`	Integer	状态（1:等待合并, 2:开始合并, 3:下载完成, 99:成功, -1:失败）
`taskIp`	String	执行机器IP
`msg`	String	处理消息
`notifyServiceId` / `notifyScenario` / `notifyFlag` / `checkAmt`		通知和校验配置
`startLine` / `isKeepHeadInfo` / `separateType` / `separatorStr` / `idFieldNum` / `idFieldStartPos` / `idFieldLen`		文件处理配置
`dtoJson`	String	原始DTO JSON

5.7 合并结果表（TM_FPS_AP_MERGE_RESULT）

字段	类型	说明
`id`	String	主键，序列号
`channel`	String	渠道标识
`business`	String	业务类型
`batchNo`	String	批次号
`toDcnNo`	String	目标DCN
`fileId`	String	合并后文件ID
`hashValues`	String	合并后文件Hash值
`noticeBizSeqNo`	String	通知业务流水号
`noticeContent`	String	通知内容

5.8 通用DTO

BaseFileSplitDto（拆分基础DTO）

RMB消息解析后的拆分请求对象，包含渠道、业务、文件ID、批次号、Hash值、分隔方式、用户标识配置、通知配置、DCN配置等40+个字段。

BaseFileMergeDTO（合并基础DTO）

合并请求对象，包含渠道、业务、批次号、DCN信息、文件ID、Hash值、通知配置、行数校验、金额校验、灰度配置等字段。

CommonSplitNoticeMessage / CommonMergeNoticeMessage

通知下游的消息体，包含渠道、业务、批次号、文件ID、Hash值、有效行数、透传参数等。

SplitNoticeMergeMessage

拆分结果转合并场景的消息体，继承 CommonSplitNoticeMessage，增加 originalDcnNo 和 toDcnNo 字段。

NoticeResultDto

通知结果DTO，包含业务流水号、成功标志、消息内容。

六、Service 层

6.1 拆分流程详细说明

6.1.1 消息接收

SplitFileSubscriberService 通过RMB @RmbMap 注解订阅拆分消息。根据消息体中的 serviceCode 字段路由到对应的 ISplitService 实现。

6.1.2 配置加载

attachSplitConfig() 方法从 TM_FPS_AP_SPLIT_CONFIG 表中加载 channel-business 对应的拆分配置，使用 BeanCopier 高效复制到DTO对象。

6.1.3 参数校验与去重

参数校验：noDcnFlag=3 时要求 validLineCount 必须传入
批次号去重：根据 channel+business+batchNo 查询状态表判断是否已处理
唯一值去重：根据 channel+business+uniqueValue 判断是否已处理

6.1.4 文件下载

通过 FpsService.get() 方法从FPS（File Process Service）下载文件到本地 original_file 目录。

6.1.5 原文件校验

checkOriFile() 方法执行以下校验：

统计文件有效行数和总行数
noDcnFlag=3 时校验传入行数与实际行数一致
lineLenCheck=1 时校验所有有效行的字节长度一致

6.1.6 物理分片

physicalSplit() 将原文件按行数平均分为5个分片文件，写入 split_file 目录。分片策略为：

跳过文件头（startLine 之前的行）
有效记录按行数均分，余数部分分配到前面的分片

6.1.7 按DCN拆分

splitFileByDCN() 使用 splitFileTaskExecutor 线程池并发处理每个分片文件。每个分片文件由一个 SplitFileByDCNCallable 实例处理：

每100行批量调用GNS查询用户所在DCN
将行数据追加写入对应的DCN临时文件（tmp_file 目录）
汇总所有分片的总处理行数

6.1.8 按DCN合并

mergeFileByDCN() 将 tmp_file 目录中同一DCN的多个分片临时文件合并为一个结果文件：

在 result_file 目录创建以DCN编号结尾的结果文件
如果配置保留文件头，将原始文件头写入每个结果文件
将各分片的DCN临时文件追加写入对应的结果文件
统计各DCN结果文件的行数到 dcnFileRowCntMap

6.1.9 拆分后行数校验

当 noDcnFlag=3 或 noDcnFlag=4 时，校验拆分后的总行数是否等于原文件有效行数。

6.1.10 上传与通知

上传：使用 fpsOperateExecutor 线程池并发上传各DCN结果文件到FPS
通知：根据 notifyFlag 路由到对应的通知实现：
- NOTICE_SPLIT_RESULT_COMMON -> CommonNotifySplitResultService（通用拆分通知）
- NOTICE_SPLIT_RESULT_TO_MERGE -> NotifySplitResultToMergeService（拆分转合并通知）

6.2 合并流程详细说明

6.2.1 消息接收与汇聚

MergeFileSubscriberService 订阅合并消息。每个DCN会发送一条合并消息（包含该DCN的文件信息）。系统需要等待所有DCN的消息都到达后才能开始合并。

6.2.2 消息到齐判断

shouldDoMerge() 方法：

查询数据库中该批次状态为"等待合并"的记录数
判断记录数是否等于DCN列表总数
如果相等，通过数据库更新操作竞争获取合并任务（只有一个线程能成功获取）
获取成功的线程负责执行合并

6.2.3 文件下载

downloadDcnFile() 使用 fpsOperateExecutor 线程池并发下载各DCN文件到 merge_original_file 目录。每个文件下载后：

如果 checkLineCount=1，校验文件实际有效行数与消息传入行数一致
更新该DCN的状态为"下载完成"

6.2.4 文件合并

merge() 方法按DCN列表排序顺序依次合并文件：

第一个文件可选择保留文件头
后续文件从 startLine 开始写入
合并结果写入 merge_result_file 目录

6.2.5 合并后计算

calMergeInfo() 根据配置执行不同的校验计算：

checkAmt=1：计算整数金额总和 + 行数
checkAmt=2：计算BigDecimal金额总和 + 行数
checkLineCount=1：仅统计有效行数
计算合并后文件的MD5值

6.2.6 通知下游

合并成功后根据 notifyFlag 路由到 NOTICE_MERGE_RESULT_COMMON -> CommonNotifyMergeResultService，通过RMB发送通知消息给下游系统。

6.3 通知服务

Bean名称	实现类	用途
`NOTICE_SPLIT_RESULT_COMMON`	`CommonNotifySplitResultService`	拆分完成后通用通知下游
`NOTICE_SPLIT_RESULT_TO_MERGE`	`NotifySplitResultToMergeService`	拆分完成后通知合并接口（拆分转合并场景）
`NOTICE_MERGE_RESULT_COMMON`	`CommonNotifyMergeResultService`	合并完成后通用通知下游

通知方式支持同步（sendRequest）和异步（publish），根据 serviceId 格式自动判断。

七、RMB 消息处理

7.1 消息订阅

系统通过RMB（Reliable Message Bus）消息总线接收业务请求，使用 @RmbMap 注解声明式订阅：

订阅者	ServiceId	Scenario	用途
`SplitFileSubscriberService`	`SERVICE_ID_SPLIT_FILE_SUBSCRIBE`	`01`	接收文件拆分请求
`MergeFileSubscriberService`	`SERVICE_ID_MERGE_FILE_SUBSCRIBE`	`01`	接收文件合并请求

7.2 消息格式

RMB消息体为JSON格式，包含以下核心字段：

拆分消息：

json

{
  "channel": "渠道标识",
  "business": "业务类型",
  "fileId": "文件ID",
  "batchNo": "批次号",
  "hashValues": ["文件Hash值列表"],
  "serviceCode": "COMMON",
  "validLineCount": 有效行数,
  "uniqueValue": "唯一值",
  "transParams": "透传参数"
}

合并消息：

json

{
  "channel": "渠道标识",
  "business": "业务类型",
  "batchNo": "批次号",
  "originalDcnNo": "来源DCN",
  "toDcnNo": "目标DCN",
  "fileId": "文件ID",
  "hashValues": ["文件Hash值列表"],
  "serviceCode": "COMMON",
  "validLineCount": 有效行数,
  "transParams": "透传参数"
}

7.3 消息发布

系统在处理完成后通过RMB发布通知消息给下游系统：

CommonNotifySplitResultService：发布拆分结果通知
NotifySplitResultToMergeService：发布拆分转合并通知
CommonNotifyMergeResultService：发布合并结果通知

7.4 RMB配置

系统ID：U636
线程池配置：核心72线程，最大288线程，KeepAlive 60秒
消费线程：最小24，最大72
消费队列：10000

八、文件处理流程

8.1 文件拆分完整流程图

上游系统
   │
   ▼ (RMB消息)
SplitFileSubscriberService.onMessage()
   │
   ▼
CommonSplitServiceImpl.execute()
   │
   ├──① 解析消息 → BaseFileSplitDto
   ├──② 加载配置 → TM_FPS_AP_SPLIT_CONFIG
   ├──③ 参数校验 & 去重检查
   ├──④ 记录状态 → TM_FPS_AP_SPLIT_STATUS (status=1)
   ├──⑤ 获取DCN列表 ← GNS
   ├──⑥ 下载文件 ← FPS → original_file/
   ├──⑦ 校验原文件
   ├──⑧ 物理分片 → split_file/ (5个分片)
   │
   ├──⑨ 按DCN拆分（并发） → tmp_file/
   │     └── SplitFileByDCNCallable (每分片一个线程)
   │           ├── 提取用户标识
   │           ├── 批量查询GNS获取DCN
   │           └── 写入对应DCN临时文件
   │
   ├──⑩ 按DCN合并 → result_file/
   │     ├── 创建DCN结果文件
   │     ├── 添加文件头（可选）
   │     └── 合并各分片DCN临时文件
   │
   ├──⑪ 行数校验
   ├──⑫ 上传文件 → FPS（并发）
   ├──⑬ 通知下游 → RMB
   ├──⑭ 记录结果 → TM_FPS_AP_SPLIT_RESULT
   └──⑮ 更新状态 → TM_FPS_AP_SPLIT_STATUS (status=99)

8.2 文件合并完整流程图

上游系统（多个DCN，每个DCN一条消息）
   │
   ▼ (RMB消息 × N条)
MergeFileSubscriberService.onMessage()
   │
   ▼
CommonMergeServiceImpl.execute()
   │
   ├──① 解析消息 → BaseFileMergeDTO
   ├──② 加载配置 → TM_FPS_AP_MERGE_CONFIG
   ├──③ 参数校验
   ├──④ 获取DCN列表 ← GNS（支持灰度过滤）
   ├──⑤ 校验DCN合法性
   ├──⑥ 记录状态 → TM_FPS_AP_MERGE_STATUS (status=1, 等待合并)
   │
   ├──⑦ 等待所有DCN消息到齐
   │     ├── 查询等待合并的记录数 == DCN列表数？
   │     └── 竞争获取合并任务（数据库原子更新）
   │
   ├──⑧ 下载所有DCN文件（并发） ← FPS → merge_original_file/
   ├──⑨ 按DCN顺序合并文件 → merge_result_file/
   │     ├── 第一个文件可保留文件头
   │     └── 后续文件跳过文件头
   │
   ├──⑩ 计算合并信息（金额/行数/MD5）
   ├──⑪ 行数校验
   ├──⑫ 上传合并文件 → FPS
   ├──⑬ 通知下游 → RMB
   ├──⑭ 记录结果 → TM_FPS_AP_MERGE_RESULT
   └──⑮ 更新状态 → TM_FPS_AP_MERGE_STATUS (status=99)

8.3 拆分转合并场景

在某些业务场景中，拆分后的文件不需要直接通知下游业务系统，而是需要再次触发合并流程。此时配置 notifyFlag=TO_MERGE，拆分完成后通过 NotifySplitResultToMergeService 发送合并消息，触发 MergeFileSubscriberService 处理。

8.4 文件路径命名规则

场景	路径模板	示例
原始文件	`{originalFileDirPath}/{channel}_{business}_{batchNo}_{fileId}_{timestamp}`	`original_file/CH01_BIZ01_BATCH001_FILE001_1690000000000`
分片文件	`{splitFileDirPath}/{原文件名}_{分片序号}`	`split_file/CH01_BIZ01_BATCH001_FILE001_1690000000000_1`
DCN临时文件	`{tmpFileDirPath}/{原文件名}_{分片序号}_{dcnNo}`	`tmp_file/CH01_..._1_dcn001`
DCN结果文件	`{resultFileDirPath}/{原文件名}_{dcnNo}`	`result_file/CH01_..._dcn001`
合并下载文件	`{merge_originalFileDirPath}/{channel}_{business}_{batchNo}_{originalDcnNo}`	`merge_original_file/CH01_BIZ01_BATCH001_dcn001`
合并结果文件	`{merge_resultFileDirPath}/{channel}_{business}_{batchNo}_{timestamp}`	`merge_result_file/CH01_BIZ01_BATCH001_1690000000000`

九、配置与部署

9.1 配置文件结构

conf/
└── bfps-ap-service/
    ├── bfps-ap-service.properties          # 主配置文件
    ├── bfps-ap-service-env.properties      # 环境变量配置（数据库、FPS连接）
    ├── gns-api.properties                  # GNS API配置
    ├── rmb-client.properties               # RMB客户端配置
    ├── service-code-path.properties        # 服务编码路径配置
    ├── log4j2.xml                          # 日志配置
    └── weservice.properties                # Web服务配置

conf-local/                                  # 本地开发配置
└── fps-ap-service/
    ├── application.properties
    ├── gns-api.properties
    ├── log4j2.xml
    ├── rmb-client.properties
    ├── service-code-path.properties
    └── weservice.properties

9.2 关键配置项

应用配置（bfps-ap-service.properties）

配置项	默认值	说明
`systemId`	`U636`	系统标识
`instanceType`	`FPSAP`	实例类型
`splitThreadSize`	`600`	拆分线程数
`localDirPath_env`	`local/fpsap/split/`	拆分本地目录
`merge_originalFileDirPath_env`	`local/fpsap/merge/original_file`	合并原始文件目录
`merge_resultFileDirPath_env`	`local/fpsap/merge/result_file`	合并结果文件目录
`flow.control.enabled`	`true`	流控开关
`flow.control.ruleSourceType`	`DB`	流控规则来源

数据库配置（bfps-ap-service-env.properties）

驱动：com.mysql.cj.jdbc.Driver
连接池最大连接数：20
密码已加密（decrypt=true）
MyBatis Mapper位置：classpath*:/mybatis-mappers/sqlmap-mapper-*-basic.xml

FPS文件服务配置

文件保存时间：7天
密码已解密（passwordDecrypted=true）

RMB消息总线配置

系统ID：U636
线程池：核心72，最大288
消费线程：最小24，最大72

9.3 环境变量依赖

环境变量	说明
`APPHOME`	应用主目录（必须配置）
`hostIp`	本机IP地址
`dcn`	当前DCN编号
`IDC_CURRENT_ENV_NAME`	环境名称（PROD表示生产）

9.4 MyBatis代码生成

bfps-ap-dao 模块使用 zcnc-sdk-starter-mybatisplugin 插件根据DDL自动生成：

Entity类（DTO）
Example查询类
DAO接口
MyBatis映射文件

生成的表包括：TmFpsApSplitConfig、TmFpsApSplitStatus、TmFpsApSplitResult、TmFpsApMergeConfig、TmFpsApMergeStatus、TmFpsApMergeResult

十、外部系统集成

10.1 GNS（Global Naming Service）全局命名服务

集成方式：通过 zcnc-sdk-rmb 提供的 GnsResolver 类调用

功能：

获取DCN列表：GnsResolver.queryByDcnTypeOrBusinessId() -- 根据DCN类型或业务线编号获取所有DCN列表
查询用户DCN：GnsResolver.query() -- 根据用户标识（QQ、银行卡号、微信号等）批量查询用户所在的DCN
获取管理DCN：GnsResolver.getAdminDCN() -- 获取ADM区DCN编号

DCN类型与业务线映射：

业务线编号	DCN类型	说明
`100001`	RDCN	零售贷款产品业务处理区
`100002`	DDCN	零售存款产品业务处理区
`100003`	CDCN	信用卡产品业务处理区
`100014`	BDCN	对公存贷产品业务处理区

用户标识类型与路由映射：

ID类型	RouteType	说明
QQ	QQ_ACCOUNT	QQ账号
BankCardNo	BANK_CARD_NUM	银行卡号
EcifNo	CIF_NUM	CIF编号
BankAccount	BANK_ACCOUNT	银行账户
Wechat	WECHAT_ACCOUNT	微信账号
IdNo	ECIF_CERTIFICATE	身份证件号
WeXinOpenId	WECHAT_OPENID	微信OpenID
QQOpenId	QQ_OPENID	QQ OpenID
SGSystemId	COOP_ACCOUNT	合作方账户
UIF_NUM	UIF_NUM	UIF编号

可靠性：

所有GNS调用最多重试5次
批量查询支持流量控制和限流重试（最大10000次）

10.2 FPS（File Process Service）文件处理服务

集成方式：通过 zcnc-sdk-io 提供的 FpsClientUtil 类调用

功能：

下载文件：FpsClientUtil.get(fileId, hashValues, localFilePath) -- 根据文件ID和Hash值下载文件
上传文件：FpsClientUtil.putFileOneWeek(localFilePath, fileTypeEnum) -- 上传文件，保留一周

使用场景：

拆分前从FPS下载原始文件
拆分后将各DCN结果文件上传到FPS
合并前从FPS下载各DCN文件
合并后将结果文件上传到FPS

10.3 RMB（Reliable Message Bus）可靠消息总线

集成方式：通过 zcnc-sdk-starter-rmb 框架，使用注解和API调用

功能：

消息订阅：@RmbMap 注解声明式订阅，实现 MessageListener 接口
消息发布：
- 异步：RmbClientUtil.publish() -- 发布异步消息
- 同步：RmbClientUtil.sendRequest() -- 发送同步请求
消息序列化：JSON格式

RMB消息头信息：

bizSeqNo -- 业务流水号
consumerSeqNo -- 消费者流水号
consumerDCN -- 消费者所在DCN
orgSysId -- 原始系统ID

10.4 其他外部依赖

服务	SDK模块	用途
序列号生成	`zcnc-sdk-starter-seq` (`SeqNo`)	生成唯一序列号（结果表ID、业务流水号）
流量控制	`zcnc-sdk-starter-flowcontrol` (`FlowController`)	GNS批量查询时的限流控制
链路追踪	`zcnc-sdk-starter-trace` (`TradeLogUtil`)	记录GNS调用耗时和结果
GB18030编码	`zcnc-sdk-nativee` (`CncGB18030StringUtils`)	处理中文字符的编码转换

附录：枚举值速查

拆分状态（SplitStatusEnum）

值	含义
1	处理中
99	处理成功
-1	处理失败

合并状态（MergeStatusEnum）

值	含义
1	消息已接收等待合并
2	批次消息完整开始合并
3	下载完成
99	处理成功
-1	处理失败

字段分隔方式（separateType）

值	含义
1	固定分隔符
2	指定字段长度
3	JSON Key

无DCN处理策略（SplitNoDcnEnum）

值	含义
1	忽略找不到DCN的行
3	拒绝执行（需要校验行数）
4	拒绝执行（无需传入行数但仍强校验）

重复校验标识（SplitDupCheckEnum）

值	含义
1	仅校验批次号
2	仅校验唯一值
3	批次号+唯一值都校验

合并金额校验（MergeCheckAmtEnum）

值	含义
1	校验整数金额
2	校验BigDecimal金额

页面解读

01-公共业务

02-账户查询

03-账户管理

04-账务处理

05-内部账户业务

06-账户特殊业务

专题知识

BFPS-AP 文件处理系统应用平台 ​

一、系统概述 ​

1.1 基本信息 ​

1.2 系统定位 ​

1.3 核心SDK依赖 ​

二、模块结构 ​

2.1 整体模块关系 ​

2.2 模块依赖关系 ​

2.3 各模块说明 ​

bfps-ap-common（公共模块） ​

bfps-ap-dao（数据访问层） ​

bfps-ap-service（业务服务层） ​

三、包架构（完整包树） ​

四、核心类详解 ​

4.1 BFpsApApplication（启动类） ​

4.2 AbstractSplitService（拆分服务抽象基类 -- 核心） ​

4.3 AbstractMergeService（合并服务抽象基类 -- 核心） ​

4.4 SplitFileByDCNCallable（按DCN拆分多线程任务 -- 核心） ​

4.5 GnsService（GNS路由服务） ​

4.6 FpsService（FPS文件服务） ​

4.7 SplitFileSubscriberService（拆分消息订阅者） ​

4.8 MergeFileSubscriberService（合并消息订阅者） ​

4.9 FpsApConfig（系统配置类） ​

4.10 TaskExecutorConfig（线程池配置） ​

五、数据模型 ​

5.1 数据库表总览 ​

5.2 拆分配置表（TM_FPS_AP_SPLIT_CONFIG） ​

5.3 拆分状态表（TM_FPS_AP_SPLIT_STATUS） ​

5.4 拆分结果表（TM_FPS_AP_SPLIT_RESULT） ​

5.5 合并配置表（TM_FPS_AP_MERGE_CONFIG） ​

5.6 合并状态表（TM_FPS_AP_MERGE_STATUS） ​

5.7 合并结果表（TM_FPS_AP_MERGE_RESULT） ​

5.8 通用DTO ​

BaseFileSplitDto（拆分基础DTO） ​

BaseFileMergeDTO（合并基础DTO） ​

CommonSplitNoticeMessage / CommonMergeNoticeMessage ​

SplitNoticeMergeMessage ​

NoticeResultDto ​

六、Service 层 ​

6.1 拆分流程详细说明 ​

6.1.1 消息接收 ​

6.1.2 配置加载 ​

6.1.3 参数校验与去重 ​

6.1.4 文件下载 ​

6.1.5 原文件校验 ​

6.1.6 物理分片 ​

6.1.7 按DCN拆分 ​

6.1.8 按DCN合并 ​

6.1.9 拆分后行数校验 ​

6.1.10 上传与通知 ​

6.2 合并流程详细说明 ​

6.2.1 消息接收与汇聚 ​

6.2.2 消息到齐判断 ​

6.2.3 文件下载 ​

6.2.4 文件合并 ​

6.2.5 合并后计算 ​

6.2.6 通知下游 ​

6.3 通知服务 ​

七、RMB 消息处理 ​

7.1 消息订阅 ​

7.2 消息格式 ​

7.3 消息发布 ​

7.4 RMB配置 ​

八、文件处理流程 ​

8.1 文件拆分完整流程图 ​

8.2 文件合并完整流程图 ​

8.3 拆分转合并场景 ​

8.4 文件路径命名规则 ​

九、配置与部署 ​

9.1 配置文件结构 ​

9.2 关键配置项 ​

应用配置（bfps-ap-service.properties） ​