3. 回放数据采集：请求/响应日志采集、消息队列（Kafka）接入、数据脱敏与存储策略

回放对比的基石是高质量、低延迟的流量采集。本章聚焦于生产环境下的数据采集链路：如何无侵入地捕获请求/响应日志、如何通过 Kafka 实现削峰填谷、以及如何在采集阶段完成合规脱敏与分层存储。

3.1 请求/响应日志采集：从网关到日志管道

风控网关通常基于 OpenResty、Kong、Spring Cloud Gateway 或自研 Netty 网关。采集层需要做到“业务零侵入”与“全量捕获”。

3.2 消息队列（Kafka）接入：高吞吐削峰

生产环境网关 QPS 可达数万至数十万，直接写入数据库或 Elasticsearch 会导致写入瓶颈。Kafka 作为日志总线，承担“缓冲 + 解耦”的核心角色。

3.2.1 架构位置

3.2.2 生产者配置（网关侧）

# 网关 logback-spring.xml 配置 Kafka Appender
<appender name="KAFKA" class="com.github.danielwegener.logback.kafka.KafkaAppender">
    <encoder>
        <pattern>%msg%n</pattern>
    </encoder>
    <topic>risk-traffic-raw</topic>
    <keyingStrategy>org.apache.kafka.clients.producer.internals.DefaultPartitioner</keyingStrategy>
    <deliveryStrategy>async</deliveryStrategy>
    <producerConfig>
        <key>bootstrap.servers</key>
        <value>kafka01:9092,kafka02:9092,kafka03:9092</value>
    </producerConfig>
    <producerConfig>
        <key>acks</key>
        <value>1</value>  <!-- 网关日志允许丢失少量，选 acks=1 提升吞吐 -->
    </producerConfig>
    <producerConfig>
        <key>compression.type</key>
        <value>snappy</value>
    </producerConfig>
    <producerConfig>
        <key>max.request.size</key>
        <value>1048576</value>  <!-- 1MB -->
    </producerConfig>
</appender>

3.2.3 消费者配置（回放服务侧）

spring:
  kafka:
    consumer:
      bootstrap-servers: kafka01:9092,kafka02:9092,kafka03:9092
      group-id: risk-replay-consumer
      auto-offset-reset: earliest
      enable-auto-commit: false
      max-poll-records: 500
      fetch-max-bytes: 52428800  # 50MB
      properties:
        isolation.level: read_committed
    listener:
      type: batch
      concurrency: 6  # 与分区数一致

3.3 数据脱敏与存储策略

回放数据包含用户手机号、身份证、银行卡等敏感字段。在采集阶段必须完成脱敏，避免明文落入存储层。

3.3.1 脱敏规则

3.3.2 脱敏执行时机

3.3.3 存储策略：分层与生命周期

3.3.4 存储写入代码示例（ES 批量写入）

// 使用 Spring Data Elasticsearch 批量写入
@Autowired
private ElasticsearchRestTemplate esTemplate;

public void batchSaveReplayLogs(List<ReplayLog> logs) {
    List<IndexQuery> queries = logs.stream()
        .map(log -> new IndexQueryBuilder()
            .withId(log.getTraceId())
            .withObject(log)
            .build())
        .collect(Collectors.toList());

    esTemplate.bulkIndex(queries, IndexCoordinates.of("risk-replay-logs"));
}

3.4 本章小结

• 采集层：网关 Filter 全量捕获请求/响应，异步写入 Kafka，避免阻塞业务线程。
• 消息队列：Kafka 作为日志总线，生产者 acks=1 + snappy 压缩，消费者 batch 拉取 + 手动 commit。
• 脱敏：在网关内完成，手机号/身份证/银行卡按规则掩码，密码直接丢弃。
• 存储：热 ES（7天）、温 ClickHouse（30天）、冷 HDFS/S3（180天），分层降低成本。