客户管理后台crm系统
日志上传线程
query
返回消费组堆积量
更新落地库数据
前置机落地库建表
前置机
权限刷新线程
阿里云oss
从配置文件中加载需要对比的表
服务端server系统
华为云waf
携带统计数据
触发权限刷新时间点
创建多个子线程
心跳输出线程
用户组消息堆积量监控线程
心跳日志添加到日志队列
返回订阅的topic和表快照信息
不一致时发起程序重启标识
消费线程
携带token，日志数据
前置机落地库
生成token
机构kafka2
每隔30s打印心跳日志
返回机构中间库特定表数据
从日志队列取出日志数据
判断堆积量是否超过阈值
账号验证
携带本地配置账号信息
消费循环
数据比对
有超时情况发起程序重启标识
触发上报时间
消费线程堵塞监控线程
返回消费到的数据
更新消费线程心跳字典
携带表名信息
获取存活消费线程数
提交消费位移
主线程
返回统计结果
返回前置机落地库特定表数据
返回表快照文件
数据比对线程
数据清洗
遍历消费心跳字典
携带旧token
检查消费心跳是否超时
超过阈值时时发起程序重启标识
返回客户最新订阅的topic，表快照信息，生成新token
消费线程存活数检查线程
携带账号信息
循环检查等待
对比账号可消费权限数
返回客户订阅的topic，表快照信息，token信息
携带kafka账号信息
清洗后的数据存进前置机落地库
落地库各表数据量上报线程
