2、数据源接入：传感器数据、API数据流、数据库日志、文件系统监听

好，咱们接着聊数据录制。上一章我们把录制框架搭起来了，但有个核心问题没解决——数据从哪来？

说白了，数据源接入是整个流程的「水龙头」。水龙头拧不开，后面管道再粗也没用。我这些年接过的数据源，少说也有几十种，但归纳起来，无非就是四大类：传感器、API、数据库日志、文件系统。今天咱们一个一个过。

2.1 传感器数据接入：最「硬核」的一类

传感器数据，说白了就是物理世界和数字世界的桥梁。我最早接触这个是在自动驾驶项目上，那时候车上装了6个摄像头、3个激光雷达、1个高精度IMU。嗯，数据量有多大？一分钟就能产生几个GB。

摄像头数据：最常见的是USB摄像头和MIPI接口的工业相机。我个人习惯用OpenCV的VideoCapture来读，但要注意——摄像头帧率不稳定是常态。我曾经遇到过一台相机，白天好好的，一到下午4点就开始丢帧，后来发现是散热问题。

# 摄像头数据接入示例（Python）
import cv2
import time

cap = cv2.VideoCapture(0)  # 0表示默认摄像头
cap.set(cv2.CAP_PROP_FPS, 30)

while True:
    ret, frame = cap.read()
    if not ret:
        print("丢帧了，重试...")
        continue
    
    # 记录精确时间戳
    timestamp = time.time_ns()
    
    # 这里把数据交给录制管道
    # recorder.write(frame, timestamp)
    
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()

激光雷达数据：这个比摄像头麻烦。激光雷达通常通过以太网或串口输出点云数据，格式五花八门。Velodyne用PCAP包，Livox用自定义协议，禾赛又有自己的格式。

我建议的做法是：统一转成PCL的PointCloud2格式，或者干脆用ROS的sensor_msgs。为什么？因为后面做回放和可视化时，标准格式能省掉你80%的适配工作。

IMU数据：加速度计+陀螺仪，频率通常很高（100Hz-1000Hz）。IMU数据的特点是「短时准确，长时漂移」。接入时要注意两点：一是要处理零偏校正，二是要跟其他传感器做时间同步。

2.2 API数据流接入：最「灵活」的一类

API数据流，说白了就是通过网络接口拿数据。这类数据源我接触最多的是金融行情和IoT设备数据。

REST API轮询：最简单粗暴的方式。但有个坑——轮询间隔设不好，要么漏数据，要么把服务器打爆。我曾经接手过一个项目，轮询间隔设了100毫秒，结果对方服务器直接返回429（Too Many Requests）。

# 带退避的API轮询示例
import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503])
session.mount('https://', HTTPAdapter(max_retries=retries))

def fetch_data(url):
    try:
        response = session.get(url, timeout=5)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"请求失败: {e}")
        return None

WebSocket：适合实时性要求高的场景。跟REST不同，WebSocket是长连接，服务端主动推送数据。我建议用websockets库（Python）或者Socket.IO（Node.js），它们处理重连和心跳都比较成熟。

gRPC流：如果你在微服务架构里，gRPC的双向流是个好东西。它基于HTTP/2，性能比REST好很多。但要注意——gRPC的流式数据在录制时，需要把protobuf消息序列化成字节流再存盘。

2.3 数据库日志接入：最「隐蔽」的一类

数据库日志接入，说白了就是监听数据库的变化。你想想看，很多时候数据已经存在数据库里了，我们不想改业务代码，但又想把数据变化录下来——这时候CDC（Change Data Capture）就派上用场了。

MySQL Binlog：MySQL的二进制日志记录了所有数据变更。我常用的工具是Canal（阿里开源）或者Debezium。它们能模拟成MySQL的从库，实时接收binlog事件。

工具	语言	支持数据库	我的评价
Canal	Java	MySQL	成熟稳定，部署稍重
Debezium	Java	MySQL, PostgreSQL, MongoDB等	生态好，Kafka集成方便
pgoutput	C	PostgreSQL	原生插件，轻量级

PostgreSQL WAL：PostgreSQL的预写日志（WAL）同样可以用于CDC。PostgreSQL 10+内置了逻辑复制功能，通过pgoutput插件就能把数据变更流出来。我个人觉得比MySQL的binlog方案更优雅，因为不需要额外部署中间件。

MongoDB Change Streams：MongoDB 3.6+提供了Change Streams API，可以直接监听集合级别的数据变更。这个API返回的是游标（cursor），你可以像读文件一样持续读取变更事件。

2.4 文件系统监听：最「朴素」的一类

文件系统监听，说白了就是盯着某个文件夹，看有没有新文件出现、旧文件被修改。这招在传统行业特别常见——很多老系统没有API，只能通过文件交换数据。

inotify（Linux）：Linux内核提供的文件系统事件监控机制。性能很好，能监控到文件创建、删除、修改、移动等事件。Python的watchdog库底层就是用的inotify。

# 使用watchdog监听文件变化
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            print(f"新文件来了: {event.src_path}")
            # 这里触发录制逻辑
    
    def on_modified(self, event):
        if not event.is_directory:
            print(f"文件被修改: {event.src_path}")

observer = Observer()
observer.schedule(MyHandler(), path='/data/incoming', recursive=False)
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

FSEvents（macOS）：macOS上的文件系统事件API，跟inotify类似。但有个区别——FSEvents会记录事件历史，即使你的程序没在运行，重启后也能拿到之前错过的事件。这个特性在某些场景下很实用。

ReadDirectoryChangesW（Windows）：Windows上的文件监控API。嗯，说实话Windows的文件监控比Linux复杂一些，涉及到I/O完成端口（IOCP）这些概念。如果你在Windows上做开发，直接用Python的watchdog库就好，它帮你封装好了底层差异。

2.5 数据源接入的通用原则

说了这么多，其实不管哪种数据源，有几个原则是通用的：

• 幂等性：同样的数据重复接入，不能产生副作用。我见过太多因为网络重试导致数据重复的案例。
• 背压处理：数据源生产速度 > 录制消费速度时，要有缓冲或丢弃策略。别让数据源把系统打崩。
• 可观测性：每个数据源都要有健康检查、延迟监控、吞吐量统计。不然出了问题你都不知道是哪个环节挂了。
• 优雅关闭：程序退出时，要确保正在处理的数据不丢失。信号处理（SIGTERM/SIGINT）是基本功。

好了，数据源接入就聊到这儿。下一章我们聊聊数据怎么在管道里流转——嗯，就是那个让很多人头疼的「数据管道设计」。