175｜数据对账与补偿体系：差异发现、重放、修复、审计和闭环

// 哈希对账示意
public boolean hashReconcile(String queryA, String queryB) {
    // 对两边数据分别按 key 排序，计算 MD5
    String hashA = computeSortedHash(dataSourceA, queryA);
    String hashB = computeSortedHash(dataSourceB, queryB);
    return hashA.equals(hashB);
}

// 逐条对账示意
public List<DiffRecord> detailReconcile(List<Record> sourceA, List<Record> sourceB) {
    Map<String, Record> mapB = sourceB.stream()
        .collect(toMap(Record::getKey, Function.identity()));
    List<DiffRecord> diffs = new ArrayList<>();
    for (Record a : sourceA) {
        Record b = mapB.get(a.getKey());
        if (b == null) {
            diffs.add(new DiffRecord(a.getKey(), DiffType.MISSING_IN_B));
        } else if (!a.equals(b)) {
            diffs.add(new DiffRecord(a.getKey(), DiffType.VALUE_MISMATCH, a, b));
        }
    }
    // 检查 B 有 A 无
    Set<String> keysA = sourceA.stream().map(Record::getKey).collect(toSet());
    for (Record b : sourceB) {
        if (!keysA.contains(b.getKey())) {
            diffs.add(new DiffRecord(b.getKey(), DiffType.MISSING_IN_A));
        }
    }
    return diffs;
}

事件存储（Event Store） → 事件消费者 → 目标数据源

// Kafka 重放示意
consumer.seekToBeginning(partitions); // 重放所有历史消息
// 或指定时间戳
Map<TopicPartition, OffsetAndTimestamp> offsets = 
    consumer.offsetsForTimes(timestampMap);
offsets.forEach((tp, offset) -> consumer.seek(tp, offset.offset()));

MySQL binlog → Canal → Kafka → 消费者 → 目标数据源

// 幂等重放示意
public void replay(Event event) {
    // 1. 检查是否已处理
    if (dedupTable.exists(event.getEventId())) {
        return; // 已处理，跳过
    }
    // 2. 检查版本号
    Data current = dataStore.get(event.getKey());
    if (current != null && current.getVersion() >= event.getVersion()) {
        dedupTable.insert(event.getEventId()); // 标记已处理（不需要实际处理）
        return;
    }
    // 3. 执行重放
    dataStore.apply(event);
    // 4. 标记已处理
    dedupTable.insert(event.getEventId());
}

发现差异
├── 自动修复（差异类型明确，修复逻辑确定）
│   ├── 重新触发业务操作（推荐）
│   ├── 直接数据覆写（需谨慎）
│   └── 合并修复（冲突场景）
├── 半自动修复（差异明确但需要人工确认）
│   └── 生成修复建议 → 人工审核 → 执行
└── 人工修复（差异复杂，需人工介入）
    └── 告警通知 → 人工分析 → 手动修复

public RepairResult repairAndVerify(DiffRecord diff) {
    // 1. 执行修复
    RepairResult repair = doRepair(diff);
    if (!repair.isSuccess()) {
        return repair.withStatus(RepairStatus.FAILED);
    }
    // 2. 二次对账验证
    DiffRecord recheck = reconcile(diff.getKey());
    if (recheck != null) {
        // 修复后仍有差异，标记为修复失败
        return repair.withStatus(RepairStatus.VERIFY_FAILED)
                     .withDetail(recheck);
    }
    // 3. 记录修复审计
    auditLog.record(diff, repair, recheck);
    return repair.withStatus(RepairStatus.SUCCESS);
}

对账审计记录：
├── 对账批次号（batch_id）
├── 对账时间
├── 对账数据源（A 和 B）
├── 对账范围（时间区间、数据量）
├── 差异数量
├── 差异明细列表
└── 对账结果摘要

修复审计记录：
├── 差异 ID
├── 修复时间
├── 修复方式（自动/人工）
├── 修复操作人
├── 修复前值 / 修复后值
├── 修复结果
└── 二次验证结果

对账任务（Trace ID）→ 差异发现（Span）→ 修复执行（Span）→ 二次验证（Span）

┌─────────────────────────────────────────────────────────┐
│                    对账补偿闭环                          │
│                                                         │
│  ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐   ┌──────┐ │
│  │ 对账  │──→│ 发现  │──→│ 修复  │──→│ 验证  │──→│ 归档  │ │
│  │      │   │ 通知  │   │      │   │      │   │      │ │
│  └──────┘   └──────┘   └──────┘   └──────┘   └──────┘ │
│      ↑                                         │        │
│      └─────────────────────────────────────────┘        │
│                    下一轮对账                            │
└─────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│                        对账调度层                              │
│  ┌──────────────────────────────────────────────────────────┐ │
│  │  XXL-Job / SchedulerX / K8s CronJob                      │ │
│  │  对账任务调度、分片、失败重试、监控                         │ │
│  └──────────────────────────────────────────────────────────┘ │
├──────────────────────────────────────────────────────────────┤
│                        对账引擎层                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐ │
│  │ 数据抽取模块   │  │ 比对计算模块   │  │ 差异处理模块          │ │
│  │ - 多数据源适配 │  │ - 哈希比对    │  │ - 差异分类            │ │
│  │ - 分页拉取    │  │ - 逐条比对    │  │ - 差异去重            │ │
│  │ - 并发控制    │  │ - 抽样比对    │  │ - 差异持久化          │ │
│  └──────────────┘  └──────────────┘  └──────────────────────┘ │
├──────────────────────────────────────────────────────────────┤
│                        补偿引擎层                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────┐ │
│  │ 自动修复模块   │  │ 人工修复模块   │  │ 验证模块              │ │
│  │ - 修复策略路由 │  │ - 修复建议生成 │  │ - 二次对账            │ │
│  │ - 幂等修复    │  │ - 修复审批流  │  │ - 修复结果记录        │ │
│  │ - 批量修复    │  │ - 修复执行    │  │                      │ │
│  └──────────────┘  └──────────────┘  └──────────────────────┘ │
├──────────────────────────────────────────────────────────────┤
│                        基础设施层                              │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│  │ 审计日志  │ │ 告警通知  │ │ 监控大盘  │ │ 数据源连接池       │ │
│  └──────────┘ └──────────┘ └──────────┘ └──────────────────┘ │
└──────────────────────────────────────────────────────────────┘

// 对账任务分片示意
public class ShardingReconcileJob {
    public void execute(int shardIndex, int shardTotal) {
        // 按 key 取模分片，每个分片处理一部分数据
        // shardIndex: 当前分片编号 (0, 1, 2, ...)
        // shardTotal: 总分片数
        String sql = "SELECT * FROM orders " +
                     "WHERE MOD(CRC32(order_id), ?) = ? " +
                     "AND update_time BETWEEN ? AND ?";
        List<Order> orders = jdbcTemplate.query(sql, 
            shardTotal, shardIndex, startTime, endTime);
        // 对当前分片数据进行对账
        reconcile(orders);
    }
}

CREATE TABLE reconcile_batch (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    batch_id VARCHAR(64) NOT NULL UNIQUE COMMENT '对账批次号',
    task_type VARCHAR(64) NOT NULL COMMENT '对账类型',
    source_a VARCHAR(128) COMMENT '数据源A',
    source_b VARCHAR(128) COMMENT '数据源B',
    time_start DATETIME NOT NULL COMMENT '对账数据开始时间',
    time_end DATETIME NOT NULL COMMENT '对账数据结束时间',
    total_count INT COMMENT '对账总条数',
    diff_count INT COMMENT '差异条数',
    status VARCHAR(16) COMMENT 'PENDING/RUNNING/DONE/FAILED',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_batch_id (batch_id),
    INDEX idx_task_type_time (task_type, created_at)
);

CREATE TABLE reconcile_diff (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    batch_id VARCHAR(64) NOT NULL COMMENT '对账批次号',
    diff_key VARCHAR(256) NOT NULL COMMENT '差异记录key',
    diff_type VARCHAR(32) COMMENT 'MISSING_IN_A/MISSING_IN_B/VALUE_MISMATCH',
    field_name VARCHAR(64) COMMENT '差异字段名',
    value_a TEXT COMMENT '数据源A的值',
    value_b TEXT COMMENT '数据源B的值',
    severity VARCHAR(8) COMMENT 'P0/P1/P2/P3',
    repair_status VARCHAR(16) COMMENT 'PENDING/AUTO_REPAIRED/MANUAL/FIXED/FAILED',
    repair_detail TEXT COMMENT '修复详情',
    verify_status VARCHAR(16) COMMENT 'PENDING/PASSED/FAILED',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_batch_id (batch_id),
    INDEX idx_repair_status (repair_status),
    INDEX idx_created_at (created_at)
);