4. 网关数据采集层：日志采集（Nginx/Kafka）、请求响应截获、埋点规范、数据脱敏策略

数据采集层是实时风控网关的“神经末梢”，负责将业务流量、用户行为、系统日志等原始信号转化为可计算的特征数据。本节围绕四个核心环节展开：日志采集管道、请求/响应截获机制、埋点规范体系以及数据脱敏策略，确保采集过程具备高吞吐、低延迟、强一致性与合规性。

4.1 日志采集管道：Nginx → Kafka 实时流

风控网关的日志采集采用“边缘采集 + 消息队列缓冲”架构。Nginx 作为流量入口，通过 Lua 模块或自定义日志格式将请求元数据实时推送至 Kafka 集群，实现日志的异步、可靠、高吞吐传输。

4.1.1 Nginx 日志格式与 Lua 采集

在 Nginx 配置中定义 JSON 格式的日志模板，并通过 access_by_lua_block 在请求处理阶段注入自定义字段（如设备指纹、会话 ID、风控标签）。

# nginx.conf — 日志格式定义
log_format risk_json escape=json
  '{'
    '"timestamp":"$time_iso8601",'
    '"remote_addr":"$remote_addr",'
    '"request_method":"$request_method",'
    '"request_uri":"$request_uri",'
    '"status":$status,'
    '"body_bytes_sent":$body_bytes_sent,'
    '"request_time":$request_time,'
    '"http_referer":"$http_referer",'
    '"http_user_agent":"$http_user_agent",'
    '"http_x_forwarded_for":"$http_x_forwarded_for",'
    '"risk_session_id":"$cookie_risk_session_id",'
    '"device_fp":"$http_x_device_fp"'
  '}';

# Lua 采集脚本片段（access_by_lua_block）
access_by_lua_block {
  local cjson = require "cjson"
  local risk_log = {
    timestamp = ngx.time(),
    remote_addr = ngx.var.remote_addr,
    request_uri = ngx.var.request_uri,
    risk_session_id = ngx.var.cookie_risk_session_id or "unknown",
    device_fp = ngx.var.http_x_device_fp or "unknown"
  }
  ngx.log(ngx.ERR, cjson.encode(risk_log))
}

4.1.2 Kafka 生产者配置与分区策略

4.2 请求/响应截获机制

在网关层通过过滤器链（Filter Chain）或中间件（Middleware）模式，在请求进入业务逻辑前、响应返回客户端前分别截获数据包。截获点包括：

• 前置截获（Pre-request）：解析 HTTP 头部、Cookie、请求体（JSON/Form），提取特征字段。
• 后置截获（Post-response）：捕获响应状态码、响应体大小、耗时、错误码。
• 异常截获（Error-handling）：记录超时、连接断开、限流拒绝等异常事件。

// Java 网关过滤器示例 — 请求截获
@Component
public class RiskCaptureFilter implements GatewayFilter {

  @Override
  public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    ServerHttpRequest request = exchange.getRequest();
    // 提取请求特征
    String sessionId = request.getHeaders().getFirst("X-Risk-Session-Id");
    String deviceFp = request.getHeaders().getFirst("X-Device-FP");
    String requestBody = resolveBody(exchange);

    // 异步发送至 Kafka
    riskLogProducer.send(new RiskLog(sessionId, deviceFp, requestBody));

    return chain.filter(exchange).then(Mono.fromRunnable(() -> {
      // 后置截获：响应状态码与耗时
      ServerHttpResponse response = exchange.getResponse();
      int statusCode = response.getStatusCode().value();
      long elapsed = System.currentTimeMillis() - startTime;
      riskLogProducer.send(new RiskResponseLog(sessionId, statusCode, elapsed));
    }));
  }
}

4.3 埋点规范体系

埋点是特征数据的“源头定义”。统一埋点规范可避免字段歧义、类型混乱、命名冲突。规范包含三个层级：

4.3.1 埋点字段命名规范

• 命名风格：全小写 + 下划线（snake_case），如 device_fp、risk_session_id。
• 类型约束：时间戳统一为毫秒级 Long 型；金额统一为分（整数）；布尔值统一为 true/false 字符串。
• 必填字段：每个埋点必须包含 event_id（全局唯一）、timestamp、source（采集节点标识）。

// 埋点 JSON 规范示例
{
  "event_id": "evt_20250408_abc123def456",
  "timestamp": 1712563200000,
  "source": "gateway-01",
  "risk_session_id": "sess_98765",
  "device_fp": "fp_a1b2c3d4",
  "user_id": "u_12345",
  "action": "login",
  "risk_decision": "allow",
  "extra": {
    "login_ip": "192.168.1.100",
    "login_type": "password"
  }
}

4.4 数据脱敏策略

风控网关采集的数据包含大量敏感信息（手机号、身份证、银行卡、密码等），必须在线程内完成脱敏后再写入日志或消息队列。脱敏策略分为三类：

4.4.1 脱敏实现示例（Java）

public class DataMasker {

  // 手机号脱敏：保留前3后4
  public static String maskPhone(String phone) {
    if (phone == null || phone.length() != 11) return phone;
    return phone.substring(0, 3) + "****" + phone.substring(7);
  }

  // 身份证脱敏：保留前6后4
  public static String maskIdCard(String idCard) {
    if (idCard == null || idCard.length() < 10) return idCard;
    return idCard.substring(0, 6) + "********" + idCard.substring(idCard.length() - 4);
  }

  // 通用替换：将敏感字段值替换为固定掩码
  public static String replaceSensitive(String value) {
    return "******";
  }
}

4.5 数据采集层整体架构

下图展示了从 Nginx 入口到 Kafka 存储的完整数据流，以及脱敏、截获、埋点规范在链路中的位置。

4.6 关键指标与监控

数据采集层的健康状态直接影响风控特征时效性。建议对以下指标设置实时告警：

• 采集延迟：从请求到达 Nginx 到 Kafka 落盘的时间差，阈值 ≤ 500ms。
• 脱敏成功率：脱敏引擎处理记录数 / 总截获记录数，目标 ≥ 99.99%。
• 埋点合规率：通过规范校验的记录数 / 总记录数，目标 ≥ 99.9%。
• Kafka 积压量：未消费消息数，阈值 ≤ 100,000 条。