4. 数据接入层设计：实时交易消息队列（Kafka）的接入、数据格式标准化、字段映射与校验

反洗钱在线推理链路的第一道关卡是数据接入层。该层负责从上游业务系统（如核心银行、支付网关、交易平台）实时接收原始交易事件，经过标准化、映射与校验后，送入下游的规则引擎与模型推理模块。本章聚焦于基于 Kafka 的消息队列接入方案、数据格式标准化策略、字段映射规则以及实时校验机制。

4.1 实时消息队列接入：Kafka 架构与消费策略

Kafka 作为高吞吐、低延迟的分布式消息系统，是反洗钱实时链路的首选数据总线。接入层需定义清晰的 Topic 命名规范、分区策略与消费者组配置。

4.2 数据格式标准化：从异构到统一

上游系统可能输出 JSON、XML、Protobuf 甚至定长文本。接入层必须将所有原始消息转换为统一的 AML 标准格式（推荐 Avro 或 Protobuf），以便下游规则引擎高效解析。

4.3 字段映射规则：源字段 → AML 标准字段

字段映射是数据接入层的核心逻辑。每个上游系统都有自己独特的字段命名与结构，必须通过配置化的映射表将其对齐到 AML 标准模型。

以下为典型的字段映射表示例（以 JSON 源为例）：

4.4 数据校验：实时质量门禁

标准化后的消息必须通过多层校验才能进入推理链路。校验失败的消息将被路由到死信队列（DLQ）并触发告警。

4.4.1 校验层级

1. 结构校验： 检查必填字段是否存在（transaction_id、amount、sender_account、receiver_account、transaction_time）
2. 格式校验： 金额必须为正整数（Long）、时间戳必须在合理范围内（非未来 5 分钟以上）、账户号符合 Luhn 算法
3. 业务校验： 货币代码在 ISO 4217 列表中、交易类型在 AML 枚举范围内、金额不超过单笔限额（如 1 亿）
4. 一致性校验： 发送方与接收方账户不能相同、金额与手续费之和不超过账户余额（可选，依赖上游数据）

4.5 代码示例：Kafka 消费者 + 校验处理器

以下为基于 Java 的 Kafka 消费者核心代码片段，展示消息消费、标准化、校验与死信处理的关键逻辑：

// Kafka 消费者主循环
while (true) {
    ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, byte[]> record : records) {
        try {
            // 1. 反序列化原始消息
            RawTransaction raw = deserialize(record.value(), record.topic());

            // 2. 标准化为 AML 标准格式
            AmlTransaction aml = standardize(raw);

            // 3. 执行多层校验
            ValidationResult result = validate(aml);

            if (result.isPassed()) {
                // 4. 发送到下游处理 Topic
                producer.send(new ProducerRecord<>("aml.transaction.standardized", aml));
            } else {
                // 5. 写入死信队列
                DeadLetterMessage dlq = new DeadLetterMessage(aml, result.getErrors(), System.currentTimeMillis());
                producer.send(new ProducerRecord<>("aml.deadletter.transaction", dlq));
                // 6. 触发告警（异步）
                alertService.sendAlert("VALIDATION_FAILED", dlq);
            }
            // 7. 手动提交偏移量
            consumer.commitSync();
        } catch (Exception e) {
            // 异常消息直接入死信
            producer.send(new ProducerRecord<>("aml.deadletter.transaction", record.value()));
            consumer.commitSync();
        }
    }
}

4.6 性能与可靠性保障