本文引用的文件
治理模块 govern 是调查场景中的数据治理中枢,负责三类核心职责:
该模块对外暴露两类接口:治理任务接口 /govern 和治理树查询接口 /gt。前端通过 SSE 实时获取治理进度,并在治理树中查看按机构、银行、运营商、人员维度的关联数据。
治理模块位于后端 module/govern,并与公共模型 common/model/govern、AI 清洗链路 module/aiclean、数据建模 module/dm、图谱 module/graph、时序图 module/otg 和 agent 工具链协作。
graph TB
subgraph "前端"
FE_API["治理前端 API<br/>governApi.ts"]
FE_DTO["治理类型定义<br/>dto.ts"]
end
subgraph "治理模块 govern"
CTRL_G["GovernController<br/>/govern"]
CTRL_GT["GovernTreeController<br/>/gt"]
SVC_GOVERN["GovernService<br/>治理流水线"]
SVC_TREE["GovernTreeService<br/>治理树查询"]
SVC_CONF["GovernConfService<br/>治理配置"]
CACHE["GovernCache<br/>时序图 SQL 映射"]
RS_TRANS["TransResultHandler<br/>交易结果处理器"]
RS_CALL["CallResultHandler<br/>通话结果处理器"]
end
subgraph "共享模型"
ENT_TREE["GovernTree<br/>治理树实体"]
ENT_CONF["GovernConf<br/>治理配置实体"]
end
subgraph "外部模块"
AICLEAN["AiCleanService / AiGovernService<br/>AI 清洗模板与治理树"]
DM["DmService<br/>数据建模/外部表"]
GRAPH["GraphService<br/>图谱关系"]
OTG["TimeSeriesService<br/>时序图"]
AGENT["GovernTableTool<br/>Agent 治理表工具"]
end
FE_API --> CTRL_G
FE_API --> CTRL_GT
CTRL_G --> SVC_GOVERN
CTRL_GT --> SVC_TREE
SVC_GOVERN --> SVC_CONF
SVC_GOVERN --> RS_TRANS
SVC_GOVERN --> RS_CALL
SVC_GOVERN --> AICLEAN
SVC_GOVERN --> CACHE
SVC_TREE --> SVC_CONF
SVC_GOVERN --> ENT_TREE
SVC_CONF --> ENT_CONF
SVC_GOVERN --> DM
SVC_GOVERN --> GRAPH
SVC_GOVERN --> OTG
AGENT --> SVC_TREE
图表来源
章节来源
治理模块的核心由控制器、服务、实体、映射器和结果处理器组成:
| 组件 | 职责 | 关键行为 |
|---|---|---|
| GovernController | 治理任务入口 | 启动 SSE、防重入、异步执行 calcTask |
| GovernTreeController | 治理树查询入口 | 提供树结构、各类分页查询 |
| GovernService | 治理流水线编排 | 阶段化执行清洗、时序图、分类树、关系边计算 |
| GovernTreeService | 治理树查询与聚合 | 组装树、按人员和对象维度查交易/通话 |
| GovernConfService | 治理配置访问 | 读取 useOpenInfo、duplicate、hasGovern 等开关 |
| GovernCache | 时序图 SQL 映射 | 维护多类数据源的列映射,生成 DuckDB SQL |
| TransResultHandler | 交易结果处理 | 将交易记录转为人员关系边批量写入 |
| CallResultHandler | 通话结果处理 | 将通话记录转为人员关系边批量写入 |
| GovernTree | 治理树实体 | 描述节点 id、父节点、类型、表名和数据量 |
| GovernConf | 治理配置实体 | 存储 confStr 与 confType,支持 JSON 扩展 |
章节来源
治理任务是一个分阶段、带超时、带进度推送的流水线。它不直接面向用户交互,而是作为后台任务驱动多个子系统的状态变化。
sequenceDiagram
participant FE as "前端"
participant CTRL as "GovernController"
participant GOV as "GovernService"
participant CONF as "GovernConfService"
participant TREE as "GovernTreeService"
participant MAPPER as "GovernMapper / Mapper XML"
participant SSE as "SSE 推送"
participant DM as "DmService / FileInfo"
participant GRAPH as "GraphService"
participant OTG as "TimeSeriesService"
participant AI as "AiCleanService / AiGovernService"
FE->>CTRL: GET /govern/sse
FE->>CTRL: GET /govern/calcTask
CTRL->>GOV: tryStartGovern(caseId)
CTRL->>GOV: calcTask(taskProcess, caseId, userId)
GOV->>SSE: 日志:治理任务已启动
GOV->>CONF: useOpenInfo() / getTransTableName()
GOV->>MAPPER: 第一阶段:清洗、去重、关系清理
alt 启用开户信息
GOV->>MAPPER: 清理并复制通话到 use_open_info_call_record
end
GOV->>OTG: 创建时序图 SQL 并执行
GOV->>TREE: 统计分类树、人员树、开户信息树
GOV->>AI: 统计 AI 动态表治理节点
GOV->>MAPPER: 第三阶段:通话/交易/其他关系边计算
GOV->>MAPPER: 合并关系边阶段表
GOV->>DM: 更新文件治理完成状态
GOV->>SSE: 进度 100% 并关闭连接
图表来源
治理树以 govern_tree 表为持久化载体,继承通用树节点基类,并通过 type、pid、tableNameEn、tableNameCn、dataCount 表达业务分类、父子关系、目标表和记录数。
classDiagram
class GovernTree {
+Long id
+Long pid
+String type
+String tableNameEn
+String tableNameCn
+Integer dataCount
+String aiTemplateId
}
class GovernConf {
+Long id
+String confStr
+String confType
+getOtherConfStr(key, defaultValue) String
}
class GovernTreeService {
+getTree() List~GovernTree~
+getTreeTablePage(dto) Map
+getTreeTransPersonPage(dto) Map
+getTreeTransBankPage(dto) Map
+getTreeCallPersonPage(dto) Map
+getTreeCallOperPage(dto) Map
+getTreeCallPersonUseOpenInfoPage(dto) Map
+getTreeCallOperUseOpenInfoPage(dto) Map
}
class GovernTreeController {
+getTree() Result
+treeTablePage(dto) Result
+getTreeTransPersonPage(dto) Result
+getTreeTransBankPage(dto) Result
+getTreeCallPersonPage(dto) Result
+getTreeCallOperPage(dto) Result
+getTreeCallPersonUseOpenInfoPage(dto) Result
+getTreeCallOperUseOpenInfoPage(dto) Result
}
GovernTreeService --> GovernTree : "返回/构造"
GovernTreeController --> GovernTreeService : "调用"
GovernTreeService --> GovernConfService : "决定通话/交易表"
图表来源
治理树查询的关键点包括:
dataCount > 0 的节点后按 pid 组装树,并为负 id 的 AI 动态表节点补充字符串形式的模板 id。tableName 从全局表头缓存取列定义,并支持前端传入的动态条件 SQL。章节来源
治理配置以 govern_conf 表存储,字段包含自增主键、JSON 或开关字符串 confStr,以及配置类型 confType。
| 配置类型 | 含义 | 默认行为 |
|---|---|---|
useOpenInfo |
是否启用开户信息模式 | 未配置时不使用开户信息 |
duplicate |
是否启用去重治理 | 未配置时不清理重复数据 |
hasGovern |
导入数据时是否跳过治理 | 未配置时不跳过 |
other |
其他 JSON 扩展配置 | 通过 getOtherConfStr 读取键值 |
GovernConfService 还负责根据配置选择实际使用的通话表和交易表:
call_record;启用时为 use_open_info_call_record。trans_record_all;未启用去重时为 trans_record。这些选择会缓存到案件级配置缓存中,避免每次治理都重复查库。
章节来源
GovernService.calcTask 是治理任务的主入口,负责上下文绑定、防重入、阶段编排和异常处理。
flowchart TD
Start(["开始 calcTask"]) --> BindContext["绑定案件与用户上下文"]
BindContext --> CleanConfig["清理案件治理配置缓存"]
CleanConfig --> FirstStage["第一阶段:清洗、去重、关系清理"]
FirstStage --> OpenInfo{"是否启用开户信息?"}
OpenInfo -->|是| OpenInfoTasks["清理并复制通话到 use_open_info_call_record"]
OpenInfo -->|否| SecondStage
OpenInfoTasks --> SecondStage["第二阶段:时序图与分类树统计"]
SecondStage --> ThirdStage["第三阶段:关系边计算"]
ThirdStage --> MergeRelEdge["合并关系边阶段表"]
MergeRelEdge --> Complete["完成治理任务"]
Complete --> UpdateFileStatus["更新文件状态为治理完成"]
Complete --> SendDone["推送 100% 进度并关闭 SSE"]
图表来源
call_record。GovernCache.getSql() 创建时序图相关数据。use_open_info_call_record 的人员和运营商节点。AiGovernService.calcAiTree() 补充 AI 动态表治理节点。章节来源
治理模块使用 MyBatis ResultHandler 把原始记录流式转换为关系边,达到阈值后批量插入,降低内存压力和数据库写入压力。
| 处理器 | 输入 | 输出 | 关键逻辑 |
|---|---|---|---|
TransResultHandler |
TransRecord |
PersonRelEdge |
根据借贷标志调整方向,对方为空时降级为姓名或摘要节点 |
CallResultHandler |
CallRecord |
PersonRelEdge |
根据主被叫方向交换节点,固定次数为 1,类别为通信 |
两者都维护一个缓存列表,达到 3000 条后批量插入 PersonRelEdgeMapper,并在结束时 flush 剩余数据。
flowchart TD
Read["读取一条原始记录"] --> Validate{"记录是否为空?"}
Validate -->|是| Next["跳过"]
Validate -->|否| Build["转换为 PersonRelEdge"]
Build --> Cache["加入批量缓存"]
Cache --> Threshold{"缓存 >= 3000?"}
Threshold -->|是| BatchInsert["批量插入关系边表"]
BatchInsert --> Clear["清空缓存"]
Threshold -->|否| End(["等待下一条"])
Clear --> End
图表来源
章节来源
治理树节点不是简单目录,而是业务分类与数据实体的混合拓扑:
| 字段 | 含义 | 说明 |
|---|---|---|
id |
节点主键 | 正 id 用于既有表;负 id 表示 AI 动态表 |
pid |
父节点 id | 根节点通常为 1 |
type |
节点类型 | 如 table、person、bank、yys、no_yys、call_person 等 |
tableNameEn |
目标英文表名 | 指向治理后的实际表或视图 |
tableNameCn |
展示中文名 | 可以是表中文名、人员姓名、银行名、运营商名 |
dataCount |
数据量 | 用于过滤空节点和前端排序 |
aiTemplateId |
AI 模板 id | 仅 AI 动态表节点填充,字符串形式避免 JS 精度丢失 |
治理树的状态主要由 dataCount 控制:统计完成后大于 0 的节点才进入前端树;AI 动态表节点通过负 id 与真实模板 id 的映射参与治理树。
章节来源
治理模块与多个子系统存在明确依赖:
graph LR
GOVERN["GovernService"] --> AICLEAN["AiCleanService / AiGovernService"]
GOVERN --> DM["DmService / FileInfoMapper"]
GOVERN --> GRAPH["GraphService"]
GOVERN --> OTG["TimeSeriesService"]
GOVERN --> PERSON["PersonBasicInfo / PersonLibNo"]
GOVERN --> CALL["CallRecord / CallRecordAll"]
GOVERN --> TRANS["TransRecord / TransRecordAll"]
GOVERN --> EDGE["EdgeConf"]
GOVERN --> REL["PersonRelEdge"]
GOVERN --> MAPPERS["GovernMapper / GovernTreeMapper"]
图表来源
治理模块的耦合特征如下:
EdgeConf、GovernCache 映射、AiGovernService 和自定义结果处理器扩展。章节来源
治理模块在性能和并发方面做了专门设计:
| 特性 | 实现方式 | 目的 |
|---|---|---|
| 单案件互斥 | RUNNING_CASES 原子集合 |
防止同一案件双击或双标签页互相 truncate 数据 |
| 异步执行 | ThreadUtil.execAsync + 虚拟线程执行器 |
大案件不阻塞其他用户的任务 |
| 阶段超时 | CompletableFuture.orTimeout |
防止某个阶段长期挂起 |
| 进度推送 | SlowDownProgress + SSE |
前端看到渐进式治理进度 |
| 批量写入 | 结果处理器每 3000 条批量插入 | 减少数据库写入压力 |
| 上下文绑定 | CaseContextHolder.runWith |
保证跨线程数据源路由和 SSE 接收者正确 |
| 活跃探针 | GovernActivityProbe |
告诉数据源回收器不要关闭正在治理的案件连接 |
治理任务的线程模型需要注意:
章节来源
治理树的静态结构来自已有表信息和 AI 清洗模板:
govern_tree。use_open_info_call_record 下的人员和运营商节点。AiGovernService.calcAiTree() 补充,负 id 对应 ai_clean_template.id。前端通过 /gt/tree 获取树结构,再通过 /gt/getTreeTransPersonPage、/gt/getTreeTransBankPage、/gt/getTreeCallPersonPage、/gt/getTreeCallOperPage 等接口查看具体数据。
治理规则主要通过以下机制绑定:
EdgeConf 配置其他类型数据的节点和关系边。govern_conf 控制是否启用开户信息、去重和导入时是否跳过治理。GovernCache.TABLE_MAPPINGS 定义源表到标准列的映射。前端调用流程:
GET /govern/sse。GET /govern/calcTask?taskProcess=66。后端关键约束:
治理结果体现在多个位置:
章节来源
FileInfo.fileStatus,并可能结合外部表进行建模。GovernCache 生成的 SQL 在 DuckDB 中创建时序图数据。GovernTableTool 访问治理结果表,辅助智能体进行分析。图谱:关系边是图谱关系的重要输入,治理模块负责将原始记录转化为结构化关系边。
graph TB
RESULT["治理结果"] --> SEARCH["search 模块"]
RESULT --> DM["dm 模块"]
RESULT --> OTG["otg 模块"]
RESULT --> AGENT["agent 工具"]
RESULT --> GRAPH["图谱模块"]
[此图为概念性流程图,不直接映射具体代码结构]
如果需要在治理流水线中加入新的业务规则,建议遵循以下路径:
GovernService 统计阶段新增节点插入逻辑,或在 GovernTreeMapper.xml 新增统计 SQL。calcOtherRel,通过 EdgeConf 配置节点字段、关系名称和类别。calcCallRel、calcTransRel 和对应结果处理器。ResultHandler<T> 的处理器。PersonRelEdge。govern_conf 表中新增配置项。GovernConfService 中添加读取方法,并在治理任务中使用。GovernCache.TABLE_MAPPINGS 中新增源表到标准列的映射。AiGovernService 把 AI 清洗模板纳入治理树。新增结果处理器时应注意:
SpringUtil.getBean 获取 PersonRelEdgeMapper。end() 方法处理剩余数据。章节来源
常见问题和定位建议:
| 现象 | 可能原因 | 排查建议 |
|---|---|---|
| 治理任务提示“请先打开案件” | Controller 未捕获案件上下文 | 检查请求是否携带有效案件 |
| 治理任务被拒绝 | 同案件已有治理任务在运行 | 等待前一个任务完成或检查并发触发 |
| 治理进度卡在某一阶段 | 某阶段超时或数据库执行异常 | 查看阶段日志和数据库慢查询 |
| 治理失败但前端仍显示正常 | 异常被吞掉 | 确认 awaitStage 和 exceptionally 是否正确上抛 |
| SSE 进度推给错误用户 | 异步线程未绑定 userId | 检查 CaseContextHolder.runWith 使用 |
| 治理树为空 | 所有节点 dataCount 为 0 |
检查基础表是否有数据、是否启用了正确表名 |
| AI 动态表治理节点无法翻页 | JS 数字精度丢失 | 使用前端下发的字符串 aiTemplateId |
| 开户信息模式下通话数据异常 | 开户时间范围或电话号码匹配问题 | 检查人员库号码时间范围和 use_open_info_call_record |
| 关系边数量异常 | 借贷方向、主被叫方向或对方为空逻辑处理不当 | 检查 TransResultHandler 和 CallResultHandler |
章节来源
治理模块 govern 是调查场景中连接原始数据与下游分析的枢纽。它以治理树表达业务分类和关系拓扑,以治理配置控制治理策略,以阶段化治理任务驱动清洗、时序图、分类统计和关系边构建。治理结果不仅服务于前端可视化,还被 search、dm、otg、agent 和图谱模块消费。
扩展治理能力时,应优先复用现有模式:通过配置驱动表选择和规则分支,通过结果处理器批量写入关系边,通过治理树节点表达分类维度,通过治理任务阶段管理复杂流程。这样既能保持治理链路的稳定性,又能支撑调查场景不断演进的业务需求。