442 lines
14 KiB
Markdown
442 lines
14 KiB
Markdown
# 碰撞避免系统数据采集模块设计文档
|
||
|
||
本文档详细描述了碰撞避免系统中数据采集模块的设计和实现，包括其架构、主要组件、工作流程以及与其他模块的交互方式。
|
||
|
||
## 1. 模块概述
|
||
|
||
数据采集模块（DataCollector）是碰撞避免系统的核心组件之一，负责从多种数据源实时采集不同类型移动物体（航空器、特勤车辆、无人车等）的位置和状态信息，并提供给系统的其他模块进行处理和分析。该模块采用定时任务机制，定期从外部API获取数据，并维护移动物体的实时状态和历史轨迹。
|
||
|
||
### 1.1 功能职责
|
||
|
||
- 从多个外部数据源采集移动物体的位置和状态信息
|
||
- 对采集的原始数据进行初步处理和转换
|
||
- 维护移动物体的历史轨迹记录
|
||
- 提供数据清理机制，避免历史数据过度积累
|
||
- 与系统其他模块协作，为碰撞检测和避险决策提供数据支持
|
||
|
||
### 1.2 模块结构
|
||
|
||
数据采集模块采用分层架构设计，主要包含以下组件：
|
||
|
||
```
|
||
dataCollector/
|
||
├── config/ # 配置类，如RestTemplate配置等
|
||
├── dao/ # 数据访问层，负责与外部数据源交互
|
||
├── model/ # 数据模型，定义数据结构
|
||
│ └── enums/ # 枚举类型定义
|
||
├── repository/ # 数据仓库，负责数据存储和检索
|
||
└── service/ # 服务层，实现核心业务逻辑
|
||
```
|
||
|
||
## 2. 核心组件设计
|
||
|
||
### 2.1 数据模型 (model)
|
||
|
||
数据模型定义了数据采集模块处理的各类数据结构：
|
||
|
||
#### 2.1.1 VehicleLocationInfo.java
|
||
|
||
用于表示车辆位置信息的数据模型：
|
||
|
||
```java
|
||
@Data
|
||
public class VehicleLocationInfo {
|
||
private String transId; // 消息唯一id
|
||
private long timestamp; // 时间戳
|
||
private String vehicleId; // 车辆ID
|
||
private double longitude; // 经度
|
||
private double latitude; // 纬度
|
||
private double direction; // 车头航向角
|
||
private double speed; // 车速
|
||
}
|
||
```
|
||
|
||
#### 2.1.2 VehicleCommand.java
|
||
|
||
用于发送控制指令到无人车的命令模型：
|
||
|
||
```java
|
||
// 简化表示，实际实现可能包含更多字段
|
||
@Data
|
||
public class VehicleCommand {
|
||
private String commandId; // 命令ID
|
||
private String vehicleId; // 目标车辆ID
|
||
private String commandType; // 命令类型（如：转向、加速、减速等）
|
||
private Map<String, Object> parameters; // 命令参数
|
||
private long timestamp; // 命令发送时间戳
|
||
}
|
||
```
|
||
|
||
#### 2.1.3 CommandResponse.java
|
||
|
||
命令执行响应模型，用于接收无人车对命令的执行结果：
|
||
|
||
```java
|
||
@Data
|
||
public class CommandResponse {
|
||
private String commandId; // 对应的命令ID
|
||
private boolean success; // 执行是否成功
|
||
private String message; // 执行结果消息
|
||
private long timestamp; // 响应时间戳
|
||
}
|
||
```
|
||
|
||
### 2.2 数据访问对象 (dao)
|
||
|
||
数据访问对象负责与外部数据源的通信，获取原始数据：
|
||
|
||
#### 2.2.1 DataCollectorDao.java
|
||
|
||
```java
|
||
@Slf4j
|
||
@Component
|
||
public class DataCollectorDao {
|
||
// 配置属性
|
||
@Value("${data.collector.vehicle-api.base-url}")
|
||
private String vehicleBaseUrl;
|
||
|
||
@Value("${data.collector.vehicle-api.endpoints.vehicle-location}")
|
||
private String vehicleLocationEndpoint;
|
||
|
||
private final RestTemplate restTemplate;
|
||
private final AuthService authService;
|
||
|
||
// 构造函数注入依赖
|
||
public DataCollectorDao(RestTemplate restTemplate, AuthService authService) {
|
||
this.restTemplate = restTemplate;
|
||
this.authService = authService;
|
||
}
|
||
|
||
// 采集航空器数据
|
||
public List<Aircraft> collectAircraftData(String endpoint, String baseUrl) {
|
||
// 通过HTTP请求获取航空器数据
|
||
// 处理响应并返回数据列表
|
||
}
|
||
|
||
// 采集特种车辆数据
|
||
public List<SpecialVehicle> collectVehicleData(String endpoint, String baseUrl) {
|
||
// 通过HTTP请求获取特种车辆数据
|
||
// 处理响应并返回数据列表
|
||
}
|
||
|
||
// 获取无人车位置信息
|
||
public List<UnmannedVehicle> getVehicleLocationInfo() {
|
||
// 通过HTTP请求获取无人车位置数据
|
||
// 处理响应并返回数据列表
|
||
}
|
||
}
|
||
```
|
||
|
||
DAO层的主要职责：
|
||
- 构建HTTP请求，包括URL、头信息等
|
||
- 处理授权认证（通过AuthService获取令牌）
|
||
- 发送请求并接收响应
|
||
- 将响应数据转换为系统内部数据模型
|
||
- 提供异常处理和日志记录
|
||
|
||
### 2.3 服务层 (service)
|
||
|
||
服务层实现核心业务逻辑，包括数据采集调度、处理和存储：
|
||
|
||
#### 2.3.1 DataCollectorService.java
|
||
|
||
```java
|
||
@Slf4j
|
||
@Service
|
||
public class DataCollectorService {
|
||
// 配置信息
|
||
@Value("${data.collector.airport-api.endpoints.vehicle}")
|
||
private String airportVehicleEndpoint;
|
||
|
||
@Value("${data.collector.airport-api.endpoints.aircraft}")
|
||
private String airportAircraftEndpoint;
|
||
|
||
@Value("${data.collector.airport-api.base-url}")
|
||
private String airportBaseUrl;
|
||
|
||
// 内存数据存储
|
||
@Getter
|
||
ConcurrentHashMap<String, List<Object>> dataMap = new ConcurrentHashMap<>();
|
||
|
||
|
||
@Autowired
|
||
private DataCollectorDao dataCollectorDao;
|
||
|
||
@Autowired
|
||
private MovingObjectRepository movingObjectRepository;
|
||
|
||
// 定时采集航空器数据
|
||
@Scheduled(fixedRateString = "${data.collector.interval}")
|
||
public void collectAircraftData() {
|
||
// 调用DAO获取最新数据
|
||
// 处理和更新历史状态
|
||
// 存储到仓库中
|
||
}
|
||
|
||
// 定时采集特种车辆数据
|
||
@Scheduled(fixedRateString = "${data.collector.interval}")
|
||
@Async // 异步执行
|
||
public void collectVehicleData() {
|
||
// 调用DAO获取最新数据
|
||
// 处理和更新历史状态
|
||
// 存储到仓库中
|
||
}
|
||
|
||
// 定时采集无人车数据
|
||
@Scheduled(fixedRateString = "${data.collector.interval}")
|
||
@Async // 异步执行
|
||
public void collectVehicleLocationData() {
|
||
// 调用DAO获取最新数据
|
||
// 处理和更新历史状态
|
||
// 存储到仓库中
|
||
}
|
||
}
|
||
```
|
||
|
||
服务层的主要职责：
|
||
- 调度定期数据采集任务
|
||
- 处理和转换数据，更新移动物体状态
|
||
- 维护移动物体的历史状态记录
|
||
- 提供数据缓存和快速访问机制
|
||
- 与仓库层交互，进行数据持久化
|
||
|
||
#### 2.3.2 AuthService.java
|
||
|
||
负责处理与外部API的认证授权：
|
||
|
||
```java
|
||
@Service
|
||
public class AuthService {
|
||
// 获取访问令牌
|
||
public String getToken() {
|
||
// 实现获取、缓存和刷新令牌的逻辑
|
||
// 可能包括用户名/密码认证或其他方式
|
||
}
|
||
}
|
||
```
|
||
|
||
#### 2.3.3 DataCleanupService.java
|
||
|
||
负责定期清理过期数据，防止系统资源占用过多：
|
||
|
||
```java
|
||
@Service
|
||
@Slf4j
|
||
public class DataCleanupService {
|
||
// 定期清理历史数据
|
||
@Scheduled(cron = "0 0 1 * * ?") // 每天凌晨1点执行
|
||
public void cleanupOldData() {
|
||
// 实现清理逻辑
|
||
// 例如：删除30天前的数据
|
||
}
|
||
}
|
||
```
|
||
|
||
### 2.4 配置 (config)
|
||
|
||
定义模块所需的配置类：
|
||
|
||
#### 2.4.1 RestTemplateConfig.java
|
||
|
||
配置HTTP客户端，用于与外部API通信：
|
||
|
||
```java
|
||
@Configuration
|
||
public class RestTemplateConfig {
|
||
@Bean
|
||
public RestTemplate restTemplate(ObjectMapper objectMapper) {
|
||
RestTemplate restTemplate = new RestTemplate();
|
||
// 配置消息转换器，使用自定义的ObjectMapper
|
||
// 设置连接超时、读取超时等参数
|
||
return restTemplate;
|
||
}
|
||
|
||
@Bean
|
||
public ObjectMapper objectMapper() {
|
||
ObjectMapper mapper = new ObjectMapper();
|
||
// 配置特性，如忽略未知属性等
|
||
return mapper;
|
||
}
|
||
}
|
||
```
|
||
|
||
## 3. 数据流程和工作机制
|
||
|
||
### 3.1 数据采集流程
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant 外部API
|
||
participant DataCollectorDao
|
||
participant DataCollectorService
|
||
participant MovingObjectRepository
|
||
|
||
Note over DataCollectorService: 定时触发(@Scheduled)
|
||
DataCollectorService->>DataCollectorDao: 请求最新数据
|
||
DataCollectorDao->>外部API: HTTP请求(带认证令牌)
|
||
外部API-->>DataCollectorDao: 返回原始数据
|
||
DataCollectorDao-->>DataCollectorService: 转换为内部数据模型
|
||
|
||
Note over DataCollectorService: 处理历史状态
|
||
DataCollectorService->>DataCollectorService: 创建MovementState对象
|
||
DataCollectorService->>DataCollectorService: 添加到历史队列
|
||
DataCollectorService->>DataCollectorService: 控制历史记录长度
|
||
|
||
DataCollectorService->>MovingObjectRepository: 更新移动物体状态
|
||
Note over MovingObjectRepository: 存储最新状态供其他模块使用
|
||
```
|
||
|
||
### 3.2 数据更新机制
|
||
|
||
数据采集模块采用如下机制维护移动物体的状态：
|
||
|
||
1. **定时轮询**: 以固定时间间隔（通过`${data.collector.interval}`配置）向外部API发送请求
|
||
2. **增量更新**: 每次只更新发生变化的数据，减少系统负担
|
||
3. **历史记录**: 为每个移动物体维护一个固定长度（MAX_HISTORY）的历史状态队列
|
||
4. **并发处理**: 使用ConcurrentHashMap等线程安全容器存储数据
|
||
5. **异步执行**: 通过@Async注解实现采集任务的异步处理
|
||
|
||
### 3.3 数据清理机制
|
||
|
||
为防止数据无限增长占用系统资源，模块实现了数据清理机制：
|
||
|
||
1. **内存数据控制**: 移动物体历史状态队列限制最大长度（MAX_HISTORY）
|
||
2. **定期清理**: DataCleanupService定期（默认每天凌晨）清理过期数据
|
||
3. **按时间阈值**: 默认清理30天前的历史数据
|
||
|
||
## 4. 与其他模块的交互
|
||
|
||
### 4.1 数据提供
|
||
|
||
数据采集模块作为数据提供方，与其他模块的交互如下：
|
||
|
||
```mermaid
|
||
flowchart TD
|
||
DC[数据采集模块] --> |提供实时位置数据| DP[数据处理模块]
|
||
DC --> |提供历史轨迹数据| DP
|
||
DC --> |提供移动物体更新| WS[WebSocket模块]
|
||
WS --> |推送位置更新给客户端| Client[客户端]
|
||
DP --> |碰撞风险分析| WS
|
||
DC --> |查询历史数据| API[控制器API]
|
||
```
|
||
|
||
### 4.2 接口定义
|
||
|
||
数据采集模块通过以下方式向其他模块提供数据：
|
||
|
||
1. **直接依赖注入**:
|
||
```java
|
||
@Service
|
||
public class ProcessingService {
|
||
@Autowired
|
||
private MovingObjectRepository movingObjectRepository;
|
||
|
||
// 使用仓库获取最新数据进行处理
|
||
}
|
||
```
|
||
|
||
2. **事件驱动**:
|
||
```java
|
||
// 在DataCollectorService中
|
||
private final ApplicationEventPublisher eventPublisher;
|
||
|
||
// 当检测到新数据时发布事件
|
||
eventPublisher.publishEvent(new NewDataEvent(data));
|
||
|
||
// 在其他模块中
|
||
@EventListener
|
||
public void handleNewData(NewDataEvent event) {
|
||
// 处理新数据
|
||
}
|
||
```
|
||
|
||
## 5. 配置项
|
||
|
||
数据采集模块通过application.yml或application.properties文件配置以下参数：
|
||
|
||
```yaml
|
||
data:
|
||
collector:
|
||
interval: 5000 # 数据采集间隔(毫秒)
|
||
airport-api:
|
||
base-url: "https://api.airport.example.com"
|
||
endpoints:
|
||
vehicle: "/api/vehicles"
|
||
aircraft: "/api/aircrafts"
|
||
vehicle-api:
|
||
base-url: "https://api.vehicle-vendor.example.com"
|
||
endpoints:
|
||
vehicle-location: "/api/location"
|
||
```
|
||
|
||
## 6. 扩展性设计
|
||
|
||
### 6.1 增加新数据源
|
||
|
||
系统设计支持轻松添加新的数据源：
|
||
|
||
1. 在DataCollectorDao中添加新的数据采集方法
|
||
2. 在DataCollectorService中添加对应的定时任务方法
|
||
3. 更新配置文件，添加新数据源的URL和端点
|
||
4. 根据需要添加新的数据模型类
|
||
|
||
### 6.2 支持不同协议
|
||
|
||
当前系统主要通过HTTP REST API获取数据，但架构设计支持扩展其他协议：
|
||
|
||
1. 通过创建专用的连接器类，如 MqttConnector、WebSocketConnector等
|
||
2. 在配置中指定通信协议和参数
|
||
3. 实现相应的数据处理和转换逻辑
|
||
|
||
## 7. 安全考虑
|
||
|
||
数据采集模块实现了以下安全措施：
|
||
|
||
1. **认证授权**：通过AuthService管理API访问令牌，定期刷新
|
||
2. **数据校验**：验证接收数据的完整性和有效性
|
||
3. **异常处理**：妥善处理网络错误和数据异常，防止系统崩溃
|
||
4. **数据隔离**：使用独立的数据模型，防止外部数据直接影响核心系统
|
||
|
||
## 8. 性能优化
|
||
|
||
为确保高性能运行，数据采集模块采用以下策略：
|
||
|
||
1. **并发处理**：使用线程池和异步任务处理多个数据源
|
||
2. **数据缓存**：通过内存缓存减少重复数据处理
|
||
3. **批量处理**：一次处理多条记录，减少系统调用开销
|
||
4. **增量更新**：只处理和传输变化的数据
|
||
5. **连接池**：复用HTTP连接，减少连接建立开销
|
||
|
||
## 9. 监控和错误处理
|
||
|
||
### 9.1 日志记录
|
||
|
||
系统使用SLF4J进行全面的日志记录：
|
||
|
||
```java
|
||
log.info("成功获取航空器数据，数量: {}", dataList.size());
|
||
log.error("采集航空器数据失败: {}", endpoint, e);
|
||
```
|
||
|
||
### 9.2 异常处理
|
||
|
||
采用try-catch块捕获并处理异常，确保数据采集失败不会影响整个系统运行：
|
||
|
||
```java
|
||
try {
|
||
// 数据采集逻辑
|
||
} catch (Exception e) {
|
||
log.error("数据采集异常: {}", e.getMessage(), e);
|
||
return Collections.emptyList(); // 返回空结果而非抛出异常
|
||
}
|
||
```
|
||
|
||
## 10. 未来改进
|
||
|
||
数据采集模块计划的未来改进方向：
|
||
|
||
1. **自适应采集频率**：根据数据变化频率动态调整采集间隔
|
||
2. **数据源健康检查**：定期检测数据源可用性，自动切换备用源
|
||
3. **数据质量评估**：引入数据质量评分机制，过滤低质量数据
|
||
4. **实时数据流**：从轮询机制升级到实时数据流（如WebSocket、MQTT）
|
||
5. **数据压缩**：对历史数据进行智能压缩，减少存储需求 |