4、数据采集与接入：实时数据源识别、API对接、消息队列（Kafka）基础
好，咱们进入第四章。前面几章我们把实时指标和主数据的关系理清了，也知道了该监控哪些关键指标。但有个现实问题——数据从哪来？怎么把它实时地、可靠地接入到我们的系统里？
这一章，我就带你走一遍数据采集与接入的完整路径。说白了，就是解决「数据怎么进管道」的问题。
4.1 实时数据源识别：你得知道该接谁
我见过不少团队，一上来就急着写代码对接，结果接了半天发现数据源根本不对——要么延迟太高，要么字段对不上。所以第一步，先搞清楚你的实时数据源有哪些。
在企业里，常见的实时数据源大概分这么几类：
| 数据源类型 | 典型代表 | 实时性 | 接入方式 |
|---|---|---|---|
| 业务数据库 | MySQL、PostgreSQL、Oracle | 秒级（CDC） | Binlog / 触发器 / 定时轮询 |
| 应用日志 | Nginx、Tomcat、业务系统日志 | 准实时（秒~分钟） | Filebeat / Fluentd / Logstash |
| 物联网设备 | 传感器、PLC、RFID | 毫秒~秒级 | MQTT / CoAP / 自定义TCP |
| 外部API | 第三方平台、SaaS服务 | 取决于接口限制 | RESTful / WebSocket / gRPC |
| 消息中间件 | Kafka、RabbitMQ、RocketMQ | 毫秒级 | Consumer API / 桥接 |
我个人习惯，在项目启动阶段先画一张「数据源地图」。把每个业务系统、每台服务器、每个外部接口都列出来，标注它的数据格式、更新频率、网络位置。这张图后面会帮你省很多排查问题的功夫。
4.2 API对接：最直接的接入方式
API对接，应该是大家最熟悉的接入方式了。但熟悉不等于做得好。我在项目中见过太多「API对接事故」——接口突然改了字段名、返回格式变了、认证过期没人管……
做API对接，我建议你关注三个核心点：
4.2.1 接口协议与认证
现在主流还是RESTful API，但gRPC在内部服务间越来越流行。认证方式上，Basic Auth已经很少用了，JWT和OAuth2.0是主流。嗯，这里要注意——JWT的过期时间别设太长，也别设太短。我一般设15分钟，配合refresh 机制。
4.2.2 数据拉取策略
实时不等于「每秒都拉」。要根据数据变化频率来定策略：
- 轮询（Polling）： 定时请求接口，适合变化不频繁的数据。间隔别太短，我见过有人设100ms轮询，结果把对方服务器打挂了。
- 长轮询（Long Polling）： 请求发出去后，服务端有数据才返回。比普通轮询效率高，但实现复杂一些。
- WebSocket： 真正的双向实时通信。适合高频交易、实时监控这类场景。
- Webhook： 反向模式——你提供一个回调地址，对方有数据变更时主动推给你。省资源，但需要你的服务能接收外部请求。
4.2.3 异常处理与重试
API对接一定会遇到网络抖动、服务不可用、限流等问题。我的经验是：
- 必须加超时控制，一般设3-5秒
- 重试要有退避策略（exponential backoff），别一失败就立刻重试
- 记录每次调用的状态码和响应时间，方便排查
核心原则： API对接不是「调通就行」，而是要能稳定运行、可监控、可恢复。我曾经因为没做重试退避，导致一个接口故障时整个数据管道雪崩——所有线程都在重试，CPU直接打满。
4.3 消息队列（Kafka）基础：数据管道的主动脉
说到实时数据接入，绕不开消息队列。而在消息队列里，Kafka是当之无愧的「事实标准」。为什么？因为它能扛住海量数据、支持高吞吐、还能保证数据不丢。
我最早接触Kafka是在一个电商大促项目里。当时业务方要求「下单后3秒内看到实时库存变化」，用传统数据库轮询根本扛不住。后来上了Kafka，问题迎刃而解。
4.3.1 Kafka的核心概念
先快速过一遍Kafka的几个关键角色，后面代码里会用到：
| 概念 | 一句话解释 | 类比 |
|---|---|---|
| Producer（生产者） | 发送消息的一方 | 快递员 |
| Consumer（消费者） | 接收消息的一方 | 收件人 |
| Topic（主题） | 消息的分类，类似数据库的表 | 快递分类（文件、衣物） |
| Partition（分区） | Topic的物理分片，支持并行读写 | 快递分到不同货架 |
| Broker（代理） | Kafka集群中的一台服务器 | 快递 |
| Offset（偏移量） | 消息在分区内的位置编号 | 快递单号 |
4.3.2 一个简单的Kafka生产-消费示例
光讲概念太虚，咱们直接看代码。下面是一个用Java写的Kafka生产者，把实时订单数据发到Kafka：
// Kafka生产者示例
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.100:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 开启幂等性，防止消息重复
props.put("enable.idempotence", "true");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// 模拟发送实时订单数据
String orderJson = "{\"orderId\":\"ORD202410001\",\"productId\":\"P10086\",\"amount\":299.00,\"timestamp\":1727683200000}";
ProducerRecord<String, String> record = new ProducerRecord<>("order_topic", orderJson);
producer.send(record, (metadata, exception) -> {
if (exception == null) {
System.out.println("消息发送成功，分区：" + metadata.partition() + "，偏移量：" + metadata.offset());
} else {
System.err.println("消息发送失败：" + exception.getMessage());
}
});
producer.close();对应的消费者代码也很简洁：
// Kafka消费者示例
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.100:9092");
props.put("group.id", "order_consumer_group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 从最早的消息开始消费
props.put("auto.offset.reset", "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("order_topic"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：offset=%d, key=%s, value=%s%n",
record.offset(), record.key(), record.value());
// 这里可以写业务逻辑，比如更新实时指标
}
}4.3.3 Kafka在实时数据接入中的定位
你想想看，如果没有Kafka，我们怎么接实时数据？每个数据源直接连到业务系统？那系统间的耦合度会高得吓人。一个数据源挂了，可能拖垮整个链路。
Kafka在这里扮演的是「缓冲层」和「解耦层」的角色：
- 缓冲： 数据源突发高峰时，Kafka能扛住，消费者慢慢消化
- 解耦： 生产者和消费者互不感知对方存在，各自独立演进
- 持久化： 消息默认保留7天，即使消费者挂了，恢复后还能从断点继续消费
下面这张图，是我在项目中常用的实时数据接入架构，你可以看看Kafka在整个链路中的位置：
4.4 接入策略选择：没有银弹
讲到这里，你可能会问：「那我到底该用API直接对接，还是上Kafka？」
我的回答是：看场景。没有一种方案能包打天下。我一般按这几个维度来判断：
- 数据量： 日均百万级以下，API直接对接够用；千万级以上，必须上消息队列
- 实时性要求： 秒级以内，优先考虑Kafka；分钟级延迟，API轮询也能接受
- 系统耦合度： 如果下游有多个系统需要同一份数据，用Kafka做广播最合适
- 运维能力： Kafka集群需要专人维护，小团队要考虑成本