2、告警数据源接入：日志采集（Filebeat/Logstash）、消息队列（Kafka）接入、API回调接入

告警系统要跑起来，第一步就是得把数据喂进来。说白了，数据源接入就是整个风控平台的「嘴巴」——吃不到数据，后面再牛的规则引擎、再炫的告警大屏都是白搭。

我这些年做风控架构，见过太多团队在数据接入这一步就栽了跟头。有的日志格式乱七八糟，有的Kafka Topic设计得一团糟，还有的API回调直接把生产打挂了。嗯，今天咱们就把这三种主流接入方式掰开揉碎了讲清楚。

2.1 日志采集：Filebeat vs Logstash

日志采集是风控数据最传统的来源。你的应用服务器、数据库、Nginx网关，每时每刻都在吐日志。这些日志里藏着大量的风控线索——比如登录失败次数、接口调用频率、异常堆栈等等。

我个人习惯把日志采集分成两个场景：轻量采集和复杂处理。对应到工具上，就是Filebeat和Logstash。

2.1.1 Filebeat：轻量级采集器

Filebeat是Elastic家的轻量级日志采集器。它有多轻？我记得有一次在客户现场，一台4核8G的机器上跑了十几个Java服务，再加一个Filebeat，CPU占用率几乎没变化。

Filebeat的核心逻辑很简单：读文件 → 发数据。它不做什么复杂的转换，就是忠实地把日志一行一行读出来，然后发到Kafka或者Elasticsearch。

适用场景： 纯采集，不需要对日志做复杂处理。比如你的应用日志已经是JSON格式，或者你只需要原始文本。

配置起来也很直接。下面是我常用的一个Filebeat配置，专门用来采集Nginx访问日志：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: nginx_access
    env: production
  fields_under_root: true

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "raw-nginx-log"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip

这里有个小细节——fields_under_root: true。如果不加这个，你自定义的log_type和env会被塞到一个叫fields的子对象里，下游消费的时候还得再解一层。我刚开始用Filebeat时就踩过这个坑，排查了半天才发现是字段层级的问题。

2.1.2 Logstash：强大的数据处理管道

Logstash就重多了。它不光能采集，还能做各种数据转换、过滤、富化。说白了，Logstash就是一个数据加工厂。

举个例子，你的应用日志可能是纯文本的：

2024-01-15 10:30:22 ERROR [http-nio-8080-exec-3] com.example.OrderService - 订单处理失败，订单号: ORD20240115001，原因: 余额不足

这种文本日志，直接扔给告警系统是没法用的。你需要把它解析成结构化的JSON。Logstash的grok插件就是干这个的：

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:thread}\] %{JAVACLASS:class} - %{GREEDYDATA:message_body}" }
  }
  
  # 从消息体中提取订单号和原因
  if [message_body] =~ /订单号:/ {
    grok {
      match => { "message_body" => "订单号: %{DATA:order_id}，原因: %{DATA:fail_reason}" }
    }
  }
  
  mutate {
    add_field => { "source_type" => "application_log" }
    remove_field => ["message", "message_body"]
  }
}
我的建议： 如果只是采集，用Filebeat就够了。如果需要对日志做解析、过滤、字段映射，再上Logstash。别一上来就堆Logstash，资源开销不是闹着玩的。

2.2 消息队列接入：Kafka

Kafka在风控系统里扮演的角色，就像一个「数据缓冲池」。所有数据源都往Kafka里写，所有消费者都从Kafka里读。这样做的好处很明显——解耦和削峰。

你想想看，风控告警系统最怕什么？最怕流量突增。双十一大促的时候，订单量可能是平时的几十倍。如果没有Kafka做缓冲，告警系统直接对接业务系统，分分钟被打挂。

2.2.1 Topic设计原则

Topic怎么设计？我见过不少团队把所有的日志都塞到一个Topic里，结果下游消费的时候各种过滤、各种判断，性能差得一塌糊涂。

我个人习惯按数据源类型和重要等级来划分Topic：

Topic名称 数据来源 数据内容 分区数 保留时间
raw-nginx-log Filebeat采集 Nginx访问日志原始数据 6 7天
raw-app-log Filebeat采集 应用服务日志原始数据 12 7天
parsed-risk-event Logstash处理 解析后的风控事件 8 3天
api-callback-event API回调 外部系统回调数据 4 1天
注意： 分区数不是越多越好。分区数过多会导致Kafka的元数据管理开销增大，而且消费者组内的Rebalance时间也会变长。我曾经把一个Topic设了64个分区，结果每次Rebalance都要等好几分钟，生产环境差点出事故。

2.2.2 生产者配置要点

往Kafka写数据的时候，有几个参数需要特别关注：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 关键参数
props.put("acks", "1");              // 只要Leader确认就返回，兼顾性能和可靠性
props.put("retries", "3");           // 重试次数，防止网络抖动丢数据
props.put("batch.size", "16384");    // 16KB的批次大小，提高吞吐量
props.put("linger.ms", "5");         // 最多等待5ms，凑够批次再发
props.put("compression.type", "snappy"); // 压缩，减少网络传输量

这里acks=1是我比较推荐的做法。acks=0虽然快但可能丢数据，acks=all虽然最安全但性能下降明显。风控场景下，数据完整性很重要，但也没必要为了万分之一的可能性牺牲性能。

2.3 API回调接入

API回调，说白了就是外部系统主动把数据推给你。比如第三方支付平台、风控服务商、或者你们公司内部的其他系统，它们通过HTTP请求把事件数据发到你的告警系统。

这种接入方式最灵活，但也最危险。为什么？因为调用方不可控。你不知道对方会发什么数据、发多大量、什么时候发。

2.3.1 接口设计规范

我设计API回调接口时，一般遵循这几个原则：

  • 统一入口：所有回调都走同一个网关地址，方便做鉴权和限流
  • 幂等设计：同一个回调事件重复发送，不会产生重复告警
  • 异步处理：收到回调后立即返回200，然后丢到Kafka里慢慢处理

下面是一个简单的回调接口示例：

@RestController
@RequestMapping("/api/v1/callback")
public class CallbackController {
    
    @PostMapping("/risk-event")
    public ResponseEntity<String> receiveRiskEvent(@RequestBody RiskEvent event) {
        // 1. 参数校验
        if (!validateEvent(event)) {
            return ResponseEntity.badRequest().body("invalid event");
        }
        
        // 2. 幂等校验：根据eventId去重
        if (idempotentService.isProcessed(event.getEventId())) {
            return ResponseEntity.ok("duplicate");
        }
        
        // 3. 异步写入Kafka
        kafkaProducer.send("api-callback-event", event.getEventId(), JSON.toJSONString(event));
        
        // 4. 立即返回成功
        return ResponseEntity.ok("success");
    }
}
避坑指南： 我曾经遇到过一个第三方系统，回调超时时间设了30秒。如果我们的接口处理慢了，它就会重试，结果导致雪崩。所以回调接口一定要快速返回，复杂的处理逻辑全部异步化。

2.3.2 安全与限流

API回调接入最怕两件事：数据伪造和流量冲击。

数据伪造怎么防？加签名。调用方用私钥对请求体签名，你这边用公钥验签。我一般用HMAC-SHA256，简单高效：

// 验签逻辑
String signature = request.getHeader("X-Signature");
String timestamp = request.getHeader("X-Timestamp");
String body = getRequestBody(request);

String expectedSign = hmacSha256(body + timestamp, secretKey);
if (!expectedSign.equals(signature)) {
    throw new SecurityException("签名验证失败");
}

流量冲击怎么防？限流。我习惯用Guava的RateLimiter做本地限流，再用Redis做分布式限流，双层保险：

// 本地限流：每秒最多100个请求
private final RateLimiter rateLimiter = RateLimiter.create(100.0);

// 分布式限流：每分钟最多1000个请求，按调用方AppId隔离
public boolean tryAcquire(String appId) {
    if (!rateLimiter.tryAcquire()) {
        return false; // 本地限流触发
    }
    
    String key = "callback:rate:" + appId;
    Long count = redisTemplate.opsForValue().increment(key);
    if (count == 1) {
        redisTemplate.expire(key, 1, TimeUnit.MINUTES);
    }
    return count <= 1000;
}

2.4 三种接入方式对比

说了这么多，最后做个总结。三种方式各有各的适用场景：

接入方式 适用场景 优点 缺点
Filebeat/Logstash 服务器日志、应用日志 部署简单、资源占用低 只能采集文件，无法处理实时API
Kafka 高吞吐、多消费者场景 解耦、削峰、持久化 运维成本高、需要额外组件
API回调 外部系统数据推送 灵活、实时性高 安全风险大、调用方不可控

实际项目中，这三种方式往往是组合使用的。比如：Filebeat采集日志到Kafka，Logstash从Kafka消费并解析，解析后的数据再写回Kafka，最后告警引擎从Kafka消费。API回调的数据也是先写到Kafka，再统一处理。

嗯，数据源接入这块就聊到这儿。记住一句话：数据进得来，告警才能出得去。接入设计得不好，后面全是坑。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →