Seongwonp/vehicle-telemetry-platform
GitHub: Seongwonp/vehicle-telemetry-platform
基于 MQTT/Kafka 的网联车辆遥测数据实时采集与监控平台，集成规则与机器学习异常检测及可视化仪表板。
Stars: 1 | Forks: 2
# 车辆遥测数据采集与监控平台
## 项目概述
这是一个 IoT 后端平台，通过 MQTT 接收来自 OBD-II dongle 或模拟器的车辆传感器数据，利用 Kafka 进行分布式处理后，执行异常检测和监控。
- **核心关键词**：实时流处理、海量数据处理、IoT 后端、网联车
- **开发周期**：2026.01 ~ 进行中
- **开发者**：박성원 (Park Sungwon)
- **移动端 App 仓库**：https://github.com/Seongwonp/vehicle-telemetry-app (Flutter，使用此后端的 API)
## 系统架构
```
flowchart TD
subgraph Vehicle["차량 / 시뮬레이터"]
OBD["OBD-II 동글\n(ELM327)"]
SIM["Python 시뮬레이터\n(vehicle_simulator.py)"]
end
subgraph Broker["MQTT 브로커"]
MQ["Eclipse Mosquitto\n포트 1883 / 8883(TLS)"]
end
subgraph SpringBoot["Spring Boot (Java 17)"]
MQTT_H["MqttMessageHandler\n@ServiceActivator"]
PROD["TelemetryProducer\n파티션 키: vehicle_id"]
CONS_S["TelemetryConsumer\ntelemetry-storage-group"]
CONS_A["TelemetryConsumer\nanomalydetector-group"]
API["REST API\nJWT + Rate Limiting"]
end
subgraph Kafka["Apache Kafka"]
T1["vehicle-telemetry\n파티션 3개"]
T2["vehicle-anomaly-alerts\n파티션 3개"]
end
subgraph Storage["데이터 저장"]
INFLUX["InfluxDB\n시계열 센서 데이터"]
PG["PostgreSQL\n차량 메타 + 이상 이력"]
REDIS["Redis\nRate Limit / BruteForce"]
end
subgraph AnomalyDetector["Python 이상 감지"]
RULES["rules.py\n룰 기반 즉시 판단"]
ML["ml_detector.py\nIsolation Forest"]
NOTIFY["notifier.py\nWebhook 알림"]
end
subgraph Monitoring["모니터링"]
PROM["Prometheus\nActuator 메트릭 수집"]
GRAFANA["Grafana\n대시보드 시각화"]
end
OBD -->|"MQTT publish"| MQ
SIM -->|"MQTT publish"| MQ
MQ -->|"Spring Integration"| MQTT_H
MQTT_H --> PROD
PROD -->|"vehicle_id 키"| T1
T1 -->|"storage-group"| CONS_S
T1 -->|"anomaly-group"| CONS_A
CONS_S --> INFLUX
CONS_S --> PG
CONS_A -->|"Kafka Consumer"| RULES
CONS_A -->|"Kafka Consumer"| ML
RULES -->|"이상 감지 시"| T2
ML -->|"이상 감지 시"| T2
T2 --> NOTIFY
T2 -->|"anomaly-storage-group"| PG
API --- INFLUX
API --- PG
API --- REDIS
PROM -->|"scrape /actuator/prometheus"| SpringBoot
PROM --> GRAFANA
INFLUX --> GRAFANA
```
## 技术栈
| 领域 | 技术 |
|------|------|
| 数据接收 | MQTT (Eclipse Mosquitto) |
| 消息队列 | Apache Kafka |
| 后端 API | Java 17 + Spring Boot 3 |
| 异常检测 | Python 3.11 (基于规则 + scikit-learn) |
| 时序数据库 | InfluxDB |
| 关系型数据库 | PostgreSQL |
| 缓存 | Redis |
| 监控 | Grafana + Prometheus |
| 车辆模拟器 | Python / C |
| 安全 | JWT, TLS/SSL, CORS, Rate Limiting, 异常访问检测 |
| 基础设施 | Docker Compose, AWS EC2 (可选) |
## 采集数据规格 (基于 OBD-II)
```
{
"vehicle_id": "KR-GA-1234",
"timestamp": "2026-05-09T10:00:00Z",
"speed": 87.3,
"rpm": 2400,
"engine_temp": 92.1,
"throttle_position": 34.5,
"fuel_level": 67.0,
"battery_voltage": 13.8,
"gps": {
"lat": 37.123456,
"lng": 127.654321
},
"dtc_codes": []
}
```
## 目录结构
```
vehicle-telemetry-platform/
├── simulator/ # 차량 데이터 시뮬레이터 (Python/C)
├── broker/ # Mosquitto MQTT 브로커 설정
├── kafka/ # Kafka 설정 및 토픽 초기화 스크립트
├── backend/ # Spring Boot API 서버
│ ├── src/
│ └── build.gradle
├── anomaly-detector/ # Python 이상 감지 모듈
├── monitoring/ # Grafana + Prometheus 설정
├── docs/ # 개발 일지 및 설계 문서
├── docker-compose.yml
├── .env.example
└── README.md
```
## 实施阶段 (Phase)
| Phase | 内容 | 状态 |
|-------|------|------|
| 1 | 数据采集 pipeline (模拟器 → MQTT → Kafka → InfluxDB) | 完成 |
| 2 | REST API 服务器 (Spring Boot + JWT + Rate Limiting) | 完成 |
| 3 | 异常检测 (基于规则 + Isolation Forest ML) | 完成 |
| 4 | 安全强化 (准备 X.509, 阻断 BruteForce, 审计日志) | 完成 |
| 5 | 监控与部署 (Grafana + Prometheus + Docker Compose) | 完成 |
| 6 | Bug 修复 (Actuator 身份验证绕过/信息泄露，强化异常处理) | 完成 |
| 7 | Refresh Token + 注销失效 (基于 Redis) | 完成 |
| 8 | 数据 pipeline 稳定性 (InfluxDB 批量写入, Kafka DLQ) | 完成 |
| 9 | AI 诊断 (Gemini API) | 完成 |
| 10 | 实际启用 MQTT mTLS (后端/模拟器客户端代码，默认仍为明文) | 完成 |
| 11 | 集成 Flutter 移动端 App — 添加 CORS 支持以允许在浏览器 (Flutter Web) 中调用 API | 完成 |
## 未来计划
| 项目 | 内容 |
|------|------|
| AWS EC2 部署 | 基于 Docker Compose 部署到实际服务器 (或 Render 免费套餐) |
| 多用户支持 | 当前 admin 单一账号 → 替换为基于数据库的用户管理。实施时还需同时进行车主验证 (阻断 IDOR) |
| WebSocket 实时仪表板 | 将 REST 轮询 → 转换为 WebSocket 推送以最小化延迟 |
| DLQ 重处理 consumer | 当前 DLQ 仅用于防丢失/隔离 — 尚未实现重处理自动化 |
| 阻止 MQTT 1883 端口运营 | mTLS 代码已在 Phase 10 完成。在实际部署运营时，颁发证书 + 启用标志 + 阻断 1883 是部署阶段的任务 |
| 完善 docker-compose 环境变量传递 | `ADMIN_USERNAME`/`ADMIN_PASSWORD`/`RATE_LIMIT_RPM`/`CORS_ALLOWED_ORIGINS` 在 `docker-compose.yml` 的 backend 服务 `environment` 中缺失，因此即使更改 `.env` 值，容器中仍会应用 Spring 的默认值 |
## 故障场景与行为
总结了在实际运营中可能发生的各种故障场景下系统的具体行为。
### 场景 1 — Kafka broker 宕机
| 步骤 | 行为 |
|------|------|
| 发生故障 | Spring Boot `TelemetryProducer` 的 `kafkaTemplate.send()` 失败 |
| 直接影响 | 车辆数据无法传送到 InfluxDB/异常检测 |
| MQTT 接收 | Mosquitto 继续独立运行。数据可到达 Spring Boot |
| 恢复时 | 重启 Kafka 后 Spring Boot 会自动重连，之后从接收到的数据开始正常处理 |
| 未实现限制 | Kafka 宕机期间接收的消息将会丢失。可以通过引入 DLQ (Dead Letter Queue) 来解决 |
### 场景 2 — Python 异常检测服务宕机
| 步骤 | 行为 |
|------|------|
| 发生故障 | `anomaly-detector` 容器关闭 |
| 直接影响 | 异常检测中断，Webhook 通知中断 |
| 数据 pipeline | `telemetry-storage-group` 是独立的 Consumer Group，因此 InfluxDB 存储不受影响继续进行 |
| Kafka 消息 | `anomaly-detector-group` 的 offset 保持暂停状态 — 重启时从积压的消息开始重新处理 |
| 恢复时 | 执行 `docker-compose restart anomaly-detector` 后自动开始处理积压消息 |
### 场景 3 — MQTT broker (Mosquitto) 重启
| 步骤 | 行为 |
|------|------|
| 发生故障 | Mosquitto 容器重启 |
| Spring Boot | 通过 `MqttPahoMessageDrivenChannelAdapter` 的 `automaticReconnect=true` 设置尝试自动重连 |
| 模拟器 | 通过 `paho-mqtt` 的重连逻辑自动重新订阅 |
| 重连耗时 | 基于 `connectionTimeout=10s`, `keepAliveInterval=60s`，在几秒内恢复 |
| 重连期间数据 | 断开连接期间模拟器发布的消息将丢失 (基于 QoS 1，因为是 broker 重启所以无法恢复会话) |
### 场景 4 — InfluxDB 写入失败
| 步骤 | 行为 |
|------|------|
| 发生故障 | InfluxDB 无法响应或写入超时 |
| 行为 | `TelemetryRepository.save()` 抛出异常 → 向上抛出 `RuntimeException` |
| Kafka offset | 发生异常时不提交该消息的 offset → Kafka 会尝试重新处理 |
| 日志 | 记录 ERROR 级别日志 `[InfluxDB] 쓰기 실패 — vehicle={} timestamp={}` |
| 未实现限制 | 反复失败的消息可能会被无限重新处理。需要引入 DLQ 和重试次数限制 |
### 场景 5 — Redis 宕机 (Rate Limiting / BruteForce)
| 步骤 | 行为 |
|------|------|
| 发生故障 | 无法连接 Redis |
| Rate Limiting | `redisTemplate.opsForValue().increment()` 抛出异常 → 请求在 `preHandle()` 中爆开 |
| 影响范围 | Rate Limiting 和 BruteForce 检测不是被禁用，而是整个 API 返回 500 响应。因为 Refresh Token (Phase 7) 也存储在 Redis 中，所以重新登录 (登录本身不受影响，仅无法重新发放) 也会一并受到影响 |
| 运维改进方向 | 考虑添加 try-catch 以在 Redis 故障时绕过 Rate Limiting (可用性与安全性的权衡) |
## 异常检测规则 (基于 Phase 3)
| 项目 | 异常条件 |
|------|----------|
| 发动机温度 | 超过 105°C |
| RPM | 超过 6000 |
| 电池电压 | 低于 11.5V 或超过 15V |
| 速度 | 超过 200km/h |
| DTC 代码 | 数组不为空时 |
## 运行测试
```
# Java (JUnit 5)
cd backend
./gradlew test
# Python — 异常检测规则测试
cd anomaly-detector
pip install -r requirements.txt pytest
pytest
# Python — 模拟器测试
cd simulator
pip install -r requirements.txt pytest
pytest
```
## 运行方法
```
# 1. 环境变量设置
cp .env.example .env
# 编辑 .env 文件（GEMINI_API_KEY 为可选 — 若无此项则仅 AI 诊断功能无法运行，其余正常运行）
# 2. 运行完整 stack
docker-compose up -d
# 3. 运行模拟器
cd simulator
python vehicle_simulator.py
```
## OBD-II 实际连接 (ELM327 dongle)
```
pip install obd
# 将 dongle 连接到车辆 OBD-II 端口后：
import obd
connection = obd.OBD()
response = connection.query(obd.commands.SPEED)
print(response.value) # 예: 87 kph
```
## 开发原则
- 安全第一：所有通信使用 TLS，禁止无身份验证的 endpoint
- 环境变量通过 `.env` 分离，禁止硬编码
- 分离接口以便于轻松切换模拟器 ↔ 实际 OBD-II
- 测试：JUnit 5 (Java), pytest (Python)
## 文档
| 文档 | 内容 |
|------|------|
| [架构决策记录 (ADR)](docs/architecture-decisions.md) | 技术选择的原因 — 不是“用了什么”，而是“为什么选择这个” |
| [数据库 schema](docs/db-schema.md) | PostgreSQL (元数据/异常历史) + InfluxDB (时序) schema |
| [部署指南](docs/deployment-guide.md) | AWS EC2 部署流程 |
| [安全自检报告](docs/security-report.md) | 基于 OWASP Top 10, UN R155 / ISO SAE 21434 的检查结果 |
| [开发日志](docs/devlog.md) | 按日期记录的工作内容、决策事项及遇到的瓶颈 |
## 参考资源
- [MQTT 协议](https://mqtt.org)
- [Apache Kafka 官方文档](https://kafka.apache.org/documentation)
- [python-OBD](https://python-obd.readthedocs.io)
- [InfluxDB 入门](https://docs.influxdata.com)
- [UN R155 / ISO SAE 21434 汽车网络安全法规]
标签：Spring Boot, 域名枚举, 异常检测, 搜索引擎查询, 数据流处理, 时序数据, 测试用例, 版权保护, 物联网, 自定义请求头, 车联网, 软件成分分析, 逆向工具, 遥测监控