实现由航班进出港通知触发路由查询和更新消息推送
This commit is contained in:
parent
7556a5e3c3
commit
243d5734f1
91
CLAUDE.md
91
CLAUDE.md
@ -26,9 +26,10 @@ The collision avoidance system integrates with RuoYi through `QuapDataAdapter` (
|
||||
**DataCollectorService and DataProcessingService MUST be completely separated**
|
||||
|
||||
#### Service Responsibilities
|
||||
- **DataCollectorService** (250ms):
|
||||
- Only collects raw position data from external APIs
|
||||
- Caches data in activeMovingObjectsCache with NULL speed/direction
|
||||
- **DataCollectorService** (250ms):
|
||||
- Only collects raw data from external APIs (position data, flight notifications)
|
||||
- Caches position data in activeMovingObjectsCache with NULL speed/direction
|
||||
- Caches flight notification data in flightNotificationCache
|
||||
- NO calculations, NO processing, NO WebSocket sending
|
||||
- Purpose: High-frequency data collection to ensure no data loss
|
||||
|
||||
@ -36,26 +37,33 @@ The collision avoidance system integrates with RuoYi through `QuapDataAdapter` (
|
||||
- Reads cached position data from DataCollectorService
|
||||
- Calculates speed and direction using SpeedCalculationService
|
||||
- Sends WebSocket position updates
|
||||
- Reads cached flight notification data from DataCollectorService
|
||||
- Sends WebSocket flight notification updates
|
||||
- Performs violation detection and path conflict detection
|
||||
- Saves data to database
|
||||
|
||||
#### Implementation Pattern
|
||||
1. **DataCollectorService Methods**:
|
||||
- `collectAircraftData()`, `collectVehicleData()`, `collectUnmannedVehicleData()`
|
||||
- `collectAircraftData()`, `collectVehicleData()`, `collectUnmannedVehicleData()` - position data collection
|
||||
- `collectFlightNotificationData()` - flight notification data collection
|
||||
- Store MovingObjects with NULL speed/direction in activeMovingObjectsCache
|
||||
- Cache FlightNotifications in flightNotificationCache
|
||||
- **FORBIDDEN**: No calculations, no WebSocket events, no processing
|
||||
|
||||
2. **DataProcessingService Methods**:
|
||||
- `performPeriodicDataProcessing()` - main processing loop
|
||||
- `calculateSpeedAndDirectionForAllObjects()` - calculate derived data
|
||||
- `sendPositionUpdatesForActiveObjects()` - WebSocket messaging
|
||||
- `sendPositionUpdatesForActiveObjects()` - WebSocket position messaging
|
||||
- `processFlightNotificationUpdates()` - WebSocket flight notification messaging
|
||||
- `performViolationDetection()` - rule engine integration
|
||||
- `saveUnmannedVehicleDataPeriodically()` - database persistence
|
||||
|
||||
#### Cache Sharing Pattern
|
||||
- DataCollectorService owns activeMovingObjectsCache
|
||||
- DataProcessingService receives cache reference via setActiveMovingObjectsCache()
|
||||
- Both services share the same cache instance but have different responsibilities
|
||||
- DataCollectorService owns activeMovingObjectsCache and flightNotificationCache
|
||||
- DataProcessingService receives cache references via setActiveMovingObjectsCache() and getFlightNotificationCache()
|
||||
- Both services share the same cache instances but have different responsibilities
|
||||
- Position data flows through activeMovingObjectsCache for calculation and WebSocket updates
|
||||
- Flight notification data flows through flightNotificationCache for WebSocket updates
|
||||
|
||||
#### Why Complete Service Separation is Critical
|
||||
- **Data Integrity**: High-frequency collection prevents missing position updates
|
||||
@ -64,11 +72,9 @@ The collision avoidance system integrates with RuoYi through `QuapDataAdapter` (
|
||||
- **Architecture Clarity**: Clear separation of concerns makes system more maintainable
|
||||
- **Timing Consistency**: Avoids conflicts between collection and calculation frequencies
|
||||
|
||||
#### Violation of This Principle
|
||||
**NEVER** place calculation logic in DataCollectorService methods. This will cause:
|
||||
- WebSocket messages showing speed/direction as 0
|
||||
- Inconsistent data due to frequency mismatch
|
||||
- Architecture violations that are difficult to debug
|
||||
#### Critical Rule
|
||||
**NEVER** place calculation logic or WebSocket publishing in DataCollectorService methods.
|
||||
**FORBIDDEN**: Direct eventPublisher.publishEvent() calls in collection methods
|
||||
|
||||
## Development Commands
|
||||
|
||||
@ -102,19 +108,9 @@ tail -f qaup-admin/app.log
|
||||
```
|
||||
|
||||
## Technology Stack
|
||||
|
||||
### Backend
|
||||
- **Spring Boot 3.5.3** with Java 17
|
||||
- **PostgreSQL + PostGIS** for spatial data
|
||||
- **MyBatis** for database operations
|
||||
- **Redis** for caching and session management
|
||||
- **WebSocket** for real-time communication
|
||||
- **JTS + GeoTools** for spatial calculations
|
||||
|
||||
### Frontend
|
||||
- **Vue 2.6.12** with Element UI
|
||||
- **Axios** for API communication
|
||||
- **ECharts** for data visualization
|
||||
- **Backend**: Spring Boot 3.5.3 with Java 17, PostgreSQL + PostGIS, MyBatis, Redis
|
||||
- **Frontend**: Vue 2.6.12 with Element UI, ECharts
|
||||
- **Real-time**: WebSocket, JTS + GeoTools for spatial calculations
|
||||
|
||||
## Configuration
|
||||
|
||||
@ -126,39 +122,20 @@ tail -f qaup-admin/app.log
|
||||
- `sql/create_sys_driver_info_table.sql`
|
||||
|
||||
### Key Configuration Files
|
||||
- `qaup-admin/src/main/resources/application.yml`: Main configuration including database, Redis, and collision system settings
|
||||
- `qaup-admin/src/main/resources/application.yml`: Main configuration
|
||||
- `qaup-ui/vue.config.js`: Frontend build configuration
|
||||
- `qaup-collision/src/main/resources/config/`: Spatial configuration files (airport_areas.yaml, airport_roads.yaml)
|
||||
- `qaup-collision/src/main/resources/config/`: Spatial configuration (airport_areas.yaml, airport_roads.yaml)
|
||||
|
||||
## Important Development Patterns
|
||||
|
||||
### Data Access
|
||||
Always use `QuapDataAdapter` for accessing vehicle and driver data in collision module instead of direct service calls. This maintains clean separation between collision detection and system management.
|
||||
Always use `QuapDataAdapter` for accessing vehicle/driver data in collision module to maintain clean separation between collision detection and system management.
|
||||
|
||||
### WebSocket Communication
|
||||
Real-time data flows through WebSocket endpoints configured in collision module. Messages use `/topic` prefix for broadcasting to connected clients.
|
||||
|
||||
### Spatial Data
|
||||
PostGIS integration handles geometric calculations. Coordinate system defaults to airport center at (120.0834104, 36.35406879).
|
||||
|
||||
### Performance Optimization
|
||||
- Data collection interval: 250ms for high-frequency updates
|
||||
- WebSocket push throttling: 1000ms to prevent frontend overload
|
||||
- Redis caching with 60-second expiration for real-time data
|
||||
|
||||
## Version Management
|
||||
|
||||
### Version Control Files
|
||||
- **VERSION.md**: Contains current version number (semantic versioning)
|
||||
- **changelog.md**: Detailed change log following Keep a Changelog format
|
||||
|
||||
### Version Update Process
|
||||
When making significant changes:
|
||||
1. Update version number in VERSION.md
|
||||
2. Add detailed changelog entry in changelog.md
|
||||
3. Include version update in commit message
|
||||
4. Use semantic versioning (MAJOR.MINOR.PATCH)
|
||||
### Real-time Communication
|
||||
- WebSocket endpoints use `/topic` prefix for broadcasting
|
||||
- PostGIS handles geometric calculations with airport center at (120.0834104, 36.35406879)
|
||||
|
||||
### Application URLs
|
||||
- **Admin Interface**: http://localhost:8080
|
||||
- **API Documentation**: http://localhost:8080/swagger-ui/index.html
|
||||
- **WebSocket Endpoint**: ws://localhost:8080/ws
|
||||
@ -174,9 +151,7 @@ When making significant changes:
|
||||
- Verify PostGIS extension is properly installed for spatial operations
|
||||
- WebSocket connection issues often relate to CORS configuration in SecurityConfig
|
||||
|
||||
### Speed/Direction Calculation Issues
|
||||
- **CRITICAL**: Speed/direction calculations must ONLY occur in processing phase (1000ms), NEVER in collection phase (250ms)
|
||||
- If speed appears as 0, check if calculations are incorrectly placed in collection phase
|
||||
- Collection phase should store MovingObjects with NULL speed/direction values
|
||||
- Processing phase should calculate and update speed/direction for all cached objects before WebSocket sending
|
||||
- Ensure complete separation: Collection = data only, Processing = calculations only
|
||||
### Architecture Issues
|
||||
- If speed/direction appears as 0, check that calculations are in DataProcessingService, not DataCollectorService
|
||||
- If WebSocket messages are too frequent, check that eventPublisher calls are in processing phase only
|
||||
- Ensure complete service separation: DataCollectorService = collection only, DataProcessingService = processing + WebSocket
|
||||
@ -2,6 +2,10 @@
|
||||
|
||||
### [2025-09-05]
|
||||
|
||||
- [x] 模拟收到进出港航班信息，访问路由接口，获取路由信息并保存处理，同时发给前端显示路由
|
||||
|
||||
### [2025-09-05]
|
||||
|
||||
- [ ] 无人车下行接口
|
||||
- [ ] 控制指令下发
|
||||
- [ ] 红绿灯通知
|
||||
@ -14,4 +18,8 @@
|
||||
- [ ] 任务执行状态
|
||||
|
||||
- [x] 给定航空器路由和无人车路由，计算出可能的冲突点
|
||||
PathConflictDetectionService.PathConflictDetectionService()
|
||||
PathConflictDetectionService.PathConflictDetectionService()
|
||||
|
||||
- [x] 实现精简版无人车状态上报接口规范和实现
|
||||
|
||||
- [x] 将 mock 服务拆分成机场、无人车、红绿灯三个python 文件
|
||||
@ -7,6 +7,7 @@
|
||||
- **SAE J3016**: 自动驾驶分级标准
|
||||
- **ISO 21448**: 预期功能安全标准(SOTIF)
|
||||
- **IEEE 2857**: 自动驾驶系统隐私工程标准
|
||||
- **AC-137-CA-202X-XX**: 民用机场无人驾驶车辆检测规范（报批稿）
|
||||
|
||||
### 1.2 API设计原则
|
||||
- **RESTful设计**: 遵循REST架构风格
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
# 通用无人车运行状态API（上报必须字段 - 精简版）
|
||||
|
||||
本说明仅包含机场部署必须按时上报的最小字段集合，便于联调与验收。完整可选字段请参见 `doc/design/universal_autonomous_vehicle_api.md`。本文件已包含完整的请求方式与可运行示例，使用本文件即可进行开发与测试。
|
||||
本说明仅包含机场部署必须按时上报的最小字段集合，便于联调与验收。完整可选字段请参见《无人车通用运行状态API接口》。本文件已包含完整的请求方式与可运行示例，使用本文件即可进行开发与测试。
|
||||
|
||||
## 1. 接口信息
|
||||
|
||||
|
||||
@ -176,8 +176,6 @@ xss:
|
||||
data:
|
||||
collector:
|
||||
interval: 250
|
||||
route:
|
||||
interval: 5000
|
||||
detection:
|
||||
interval: 1000
|
||||
airport-api:
|
||||
|
||||
@ -197,12 +197,6 @@ data:
|
||||
collector:
|
||||
# 数据采集间隔，单位：毫秒（高频采集保证数据新鲜度）
|
||||
interval: 250
|
||||
# 路由数据采集间隔，单位：毫秒（较低频率采集路由信息）
|
||||
route:
|
||||
interval: 5000
|
||||
# 航班进出港通知采集间隔，单位：毫秒
|
||||
flightnotification:
|
||||
interval: 1000
|
||||
# 检测和推送间隔配置
|
||||
detection:
|
||||
# 检测间隔，单位：毫秒（控制围栏检测、冲突检测、违规检测和WebSocket推送频率）
|
||||
|
||||
@ -71,6 +71,12 @@ public class DataCollectorService {
|
||||
@Value("${data.collector.detection.interval:1000}")
|
||||
private long detectionInterval;
|
||||
|
||||
@Value("${data.collector.route.cache-expiry-hours:2}")
|
||||
private int routeCacheExpiryHours;
|
||||
|
||||
@Value("${data.collector.route.periodic-collection-enabled:false}")
|
||||
private boolean periodicRouteCollectionEnabled;
|
||||
|
||||
@Autowired
|
||||
private DataCollectorDao dataCollectorDao;
|
||||
|
||||
@ -99,6 +105,12 @@ public class DataCollectorService {
|
||||
|
||||
// 用于缓存所有活跃的MovingObject的最新状态
|
||||
private final Map<String, MovingObject> activeMovingObjectsCache = new ConcurrentHashMap<>();
|
||||
|
||||
// 路由获取状态缓存：Key=flightNo:type:time, Value=获取时间戳
|
||||
private final Map<String, Long> routeRetrievalCache = new ConcurrentHashMap<>();
|
||||
|
||||
// 航班通知缓存：Key=flightNo:type, Value=最新的航班通知
|
||||
private final Map<String, FlightNotification> flightNotificationCache = new ConcurrentHashMap<>();
|
||||
|
||||
// 初始化时将缓存引用传递给数据处理服务
|
||||
@PostConstruct
|
||||
@ -107,66 +119,6 @@ public class DataCollectorService {
|
||||
log.info("DataCollectorService 初始化完成，缓存引用已传递给DataProcessingService");
|
||||
}
|
||||
|
||||
/**
|
||||
* 采集航空器路由和状态数据
|
||||
*
|
||||
* 数据来源：机场系统API
|
||||
* - 获取航空器状态 (/aircraftStatusController/getAircraftStatus)
|
||||
* - 获取进港路由 (/runwayPathPlanningController/findArrTaxiwayByRunwayAndContactCrossAndSeat)
|
||||
* - 获取出港路由 (/runwayPathPlanningController/findDepTaxiwayByRunwayAndContactCrossAndSeat)
|
||||
*
|
||||
* 说明：此方法用于更新航空器的路由信息，支持CA3456生命周期模拟
|
||||
*/
|
||||
@Scheduled(fixedRateString = "${data.collector.route.interval}")
|
||||
public void collectAircraftRouteAndStatus() {
|
||||
if (collectorDisabled) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// 获取航空器状态
|
||||
AircraftStatusDTO aircraftStatus = dataCollectorDao.getAircraftStatus();
|
||||
if (aircraftStatus == null) {
|
||||
log.debug("未获取到航空器状态数据");
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("获取到航空器状态: 航班号={}, 类型={}, 跑道={}-{}, 机位={}",
|
||||
aircraftStatus.getFlightNo(),
|
||||
aircraftStatus.getType(),
|
||||
aircraftStatus.getInRunway(),
|
||||
aircraftStatus.getOutRunway(),
|
||||
aircraftStatus.getSeat());
|
||||
|
||||
// 根据航空器状态类型获取相应的路由数据
|
||||
AircraftRouteDTO routeData = getRouteDataByAircraftStatus(aircraftStatus);
|
||||
|
||||
if (routeData != null) {
|
||||
log.info("获取到{}路由数据: 编码={}, 状态={}",
|
||||
routeData.getType(),
|
||||
routeData.getCodes(),
|
||||
routeData.getStatus());
|
||||
|
||||
// 转换DTO为航空器路由对象
|
||||
AircraftRoute aircraftRoute = convertToAircraftRoute(routeData);
|
||||
|
||||
if (aircraftRoute != null) {
|
||||
// 直接保存路由到数据库，不依赖缓存中的航空器对象
|
||||
saveAircraftRouteToDatabase(aircraftStatus.getFlightNo(), aircraftRoute);
|
||||
|
||||
// 更新缓存中的航空器路由信息（如果缓存中有对象的话）
|
||||
updateAircraftRouteInCache(aircraftStatus.getFlightNo(), aircraftRoute, aircraftStatus);
|
||||
|
||||
// 发布WebSocket事件通知前端路由更新
|
||||
publishAircraftRouteUpdateEvent(aircraftStatus.getFlightNo(), aircraftRoute, aircraftStatus);
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("采集航空器路由和状态数据异常", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将AircraftRouteDTO转换为AircraftRoute对象
|
||||
* 使用JTS将多个LineString段合并为单一连续路径
|
||||
@ -238,60 +190,6 @@ public class DataCollectorService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据航空器状态获取路由数据
|
||||
* 先查询航班的路由参数，然后调用对应的路由接口
|
||||
*
|
||||
* @param aircraftStatus 航空器状态
|
||||
* @return 路由数据
|
||||
*/
|
||||
private AircraftRouteDTO getRouteDataByAircraftStatus(AircraftStatusDTO aircraftStatus) {
|
||||
try {
|
||||
String flightNo = aircraftStatus.getFlightNo();
|
||||
String routeType = aircraftStatus.getType();
|
||||
|
||||
log.info("开始获取航班路由参数: flightNo={}, routeType={}", flightNo, routeType);
|
||||
|
||||
// TODO: 调用航班路由参数查询接口获取完整参数
|
||||
// 暂时使用现有的aircraftStatus中的参数
|
||||
|
||||
if ("IN".equals(routeType)) {
|
||||
// 进港状态，获取进港路由
|
||||
log.info("查询进港路由: inRunway={}, outRunway={}, contactCross={}, seat={}",
|
||||
aircraftStatus.getInRunway(),
|
||||
aircraftStatus.getOutRunway(),
|
||||
aircraftStatus.getContactCross(),
|
||||
aircraftStatus.getSeat());
|
||||
|
||||
return dataCollectorDao.getArrivalRoute(
|
||||
aircraftStatus.getInRunway(),
|
||||
aircraftStatus.getOutRunway(),
|
||||
aircraftStatus.getContactCross(),
|
||||
aircraftStatus.getSeat()
|
||||
);
|
||||
} else if ("OUT".equals(routeType)) {
|
||||
// 出港状态，获取出港路由
|
||||
log.info("查询出港路由: inRunway={}, outRunway={}, startSeat={}",
|
||||
aircraftStatus.getInRunway(),
|
||||
aircraftStatus.getOutRunway(),
|
||||
aircraftStatus.getSeat());
|
||||
|
||||
return dataCollectorDao.getDepartureRoute(
|
||||
aircraftStatus.getInRunway(),
|
||||
aircraftStatus.getOutRunway(),
|
||||
aircraftStatus.getSeat()
|
||||
);
|
||||
} else {
|
||||
log.warn("未知的航空器状态类型: {}", routeType);
|
||||
return null;
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("获取航班路由数据异常: flightNo={}, routeType={}",
|
||||
aircraftStatus.getFlightNo(), aircraftStatus.getType(), e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存航空器路由到数据库（独立于缓存）
|
||||
@ -315,61 +213,6 @@ public class DataCollectorService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新缓存中的航空器路由信息（可选操作，不影响路由保存）
|
||||
*/
|
||||
private void updateAircraftRouteInCache(String flightNo, AircraftRoute route, AircraftStatusDTO status) {
|
||||
MovingObject cachedAircraft = activeMovingObjectsCache.get(flightNo);
|
||||
|
||||
if (cachedAircraft != null && cachedAircraft instanceof Aircraft) {
|
||||
Aircraft aircraft = (Aircraft) cachedAircraft;
|
||||
|
||||
// 根据航空器状态更新路由
|
||||
if ("IN".equals(status.getType())) {
|
||||
aircraft.setArrivalRoute(route);
|
||||
aircraft.activateArrivalRoute();
|
||||
} else if ("OUT".equals(status.getType())) {
|
||||
aircraft.setDepartureRoute(route);
|
||||
aircraft.activateDepartureRoute();
|
||||
}
|
||||
|
||||
// 更新缓存
|
||||
activeMovingObjectsCache.put(flightNo, aircraft);
|
||||
|
||||
log.debug("成功更新航空器路由缓存: flightNo={}, type={}, codes={}",
|
||||
flightNo, route.getType(), route.getCodes());
|
||||
} else {
|
||||
log.debug("缓存中暂无航空器对象，跳过缓存更新: flightNo={}", flightNo);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布航空器路由更新事件
|
||||
*/
|
||||
private void publishAircraftRouteUpdateEvent(String flightNo, AircraftRoute route, AircraftStatusDTO status) {
|
||||
try {
|
||||
// 创建路由更新事件
|
||||
AircraftRouteUpdateEvent routeUpdateEvent = AircraftRouteUpdateEvent.builder()
|
||||
.flightNo(flightNo)
|
||||
.routeType(route.getType())
|
||||
.routeStatus(route.getStatus())
|
||||
.routeCodes(route.getCodes())
|
||||
.aircraftStatus(status.getType())
|
||||
.routeGeometry(route.getGeometry() != null ? route.getGeometry().toText() : null)
|
||||
.timestamp(System.currentTimeMillis())
|
||||
.eventSource("SCHEDULED") // 标识为定时采集触发的事件
|
||||
.build();
|
||||
|
||||
// 发布WebSocket事件
|
||||
eventPublisher.publishEvent(routeUpdateEvent);
|
||||
|
||||
log.info("发布航空器路由更新事件: 航班号={}, 路由类型={}, 状态={}",
|
||||
flightNo, route.getType(), status.getType());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("发布航空器路由更新事件失败: flightNo={}", flightNo, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 定时采集航空器数据
|
||||
@ -802,9 +645,9 @@ public class DataCollectorService {
|
||||
* - 采集当前活跃的进出港航班通知
|
||||
* - 支持进港落地和出港滑出事件通知
|
||||
* - 数据存储到缓存中供后续处理和WebSocket推送
|
||||
* - 采集间隔：使用航班进出港通知采集间隔（1秒）
|
||||
* - 采集间隔：使用统一的数据采集间隔
|
||||
*/
|
||||
@Scheduled(fixedRateString = "${data.collector.flightnotification.interval:1000}")
|
||||
@Scheduled(fixedRateString = "${data.collector.interval}")
|
||||
public void collectFlightNotificationData() {
|
||||
if (collectorDisabled) {
|
||||
return;
|
||||
@ -831,8 +674,8 @@ public class DataCollectorService {
|
||||
notification.getSeat(),
|
||||
notification.getEventDateTime());
|
||||
|
||||
// 发布WebSocket事件推送到前端
|
||||
publishFlightNotificationEvent(notification);
|
||||
// 缓存航班通知供DataProcessingService推送给前端
|
||||
cacheFlightNotification(notification);
|
||||
|
||||
// 🚀 新增：事件驱动的路由查询
|
||||
// 基于航班通知触发路由查询和更新
|
||||
@ -896,27 +739,68 @@ public class DataCollectorService {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 发布航班进出港通知WebSocket事件
|
||||
* 缓存航班通知供DataProcessingService处理
|
||||
*/
|
||||
private void publishFlightNotificationEvent(FlightNotification notification) {
|
||||
try {
|
||||
// 使用已有的FlightNotificationEvent创建WebSocket事件
|
||||
com.qaup.collision.websocket.event.FlightNotificationEvent event =
|
||||
com.qaup.collision.websocket.event.FlightNotificationEvent.fromFlightNotification(notification);
|
||||
private void cacheFlightNotification(FlightNotification notification) {
|
||||
String cacheKey = notification.getFlightNo() + ":" + notification.getType().name();
|
||||
flightNotificationCache.put(cacheKey, notification);
|
||||
log.debug("缓存航班通知: flightNo={}, type={}, 缓存数量={}",
|
||||
notification.getFlightNo(), notification.getType(), flightNotificationCache.size());
|
||||
}
|
||||
|
||||
if (event != null && event.isValid()) {
|
||||
// 发布WebSocket事件
|
||||
eventPublisher.publishEvent(event);
|
||||
/**
|
||||
* 获取航班通知缓存的引用（供DataProcessingService使用）
|
||||
*/
|
||||
public Map<String, FlightNotification> getFlightNotificationCache() {
|
||||
return flightNotificationCache;
|
||||
}
|
||||
|
||||
log.info("📡 发布航班进出港通知WebSocket事件: 航班号={}, 事件类型={}, 通知级别={}",
|
||||
event.getFlightNo(), event.getEventType(), event.getNotificationLevel());
|
||||
} else {
|
||||
log.warn("⚠️ 航班进出港通知WebSocket事件创建失败或无效: {}", notification.getFlightNo());
|
||||
}
|
||||
/**
|
||||
* 检查指定航班事件的路由是否已经获取过
|
||||
*
|
||||
* @param flightNo 航班号
|
||||
* @param type 路由类型（IN/OUT）
|
||||
* @param time 事件时间戳
|
||||
* @return true表示已获取过，false表示未获取过
|
||||
*/
|
||||
private boolean isRouteAlreadyRetrieved(String flightNo, String type, Long time) {
|
||||
String cacheKey = flightNo + ":" + type + ":" + time;
|
||||
return routeRetrievalCache.containsKey(cacheKey);
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("❌ 发布航班进出港通知WebSocket事件失败: flightNo={}", notification.getFlightNo(), e);
|
||||
/**
|
||||
* 标记指定航班事件的路由已获取
|
||||
*
|
||||
* @param flightNo 航班号
|
||||
* @param type 路由类型（IN/OUT）
|
||||
* @param time 事件时间戳
|
||||
*/
|
||||
private void markRouteAsRetrieved(String flightNo, String type, Long time) {
|
||||
String cacheKey = flightNo + ":" + type + ":" + time;
|
||||
routeRetrievalCache.put(cacheKey, System.currentTimeMillis());
|
||||
log.debug("标记路由已获取: cacheKey={}, 当前缓存数量={}", cacheKey, routeRetrievalCache.size());
|
||||
}
|
||||
|
||||
/**
|
||||
* 定时清理过期的路由获取缓存记录
|
||||
*/
|
||||
@Scheduled(fixedRate = 3600000) // 每小时执行一次
|
||||
public void cleanExpiredRouteCache() {
|
||||
if (routeRetrievalCache.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
long expiryTime = System.currentTimeMillis() - (routeCacheExpiryHours * 3600 * 1000L);
|
||||
int sizeBefore = routeRetrievalCache.size();
|
||||
|
||||
routeRetrievalCache.entrySet().removeIf(entry -> entry.getValue() < expiryTime);
|
||||
|
||||
int sizeAfter = routeRetrievalCache.size();
|
||||
if (sizeBefore != sizeAfter) {
|
||||
log.info("清理过期路由缓存完成：清理前{}条，清理后{}条，清理了{}条记录",
|
||||
sizeBefore, sizeAfter, sizeBefore - sizeAfter);
|
||||
}
|
||||
}
|
||||
|
||||
@ -936,7 +820,14 @@ public class DataCollectorService {
|
||||
String flightNo = flightNotification.getFlightNo();
|
||||
String routeType = flightNotification.getType();
|
||||
|
||||
log.info("🛫 航班通知触发路由查询: 航班号={}, 类型={}", flightNo, routeType);
|
||||
// 检查是否已获取过该航班事件的路由
|
||||
if (isRouteAlreadyRetrieved(flightNo, routeType, flightNotification.getTime())) {
|
||||
log.info("🔄 航班路由已获取过，跳过重复查询: 航班号={}, 类型={}, 时间={}",
|
||||
flightNo, routeType, flightNotification.getTime());
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("🛫 航班通知触发路由查询: 航班号={}, 类型={}, 时间={}", flightNo, routeType, flightNotification.getTime());
|
||||
|
||||
// 步骤1: 查询航班路由参数
|
||||
com.qaup.collision.datacollector.dto.AircraftRouteParamsDTO routeParams =
|
||||
@ -997,7 +888,10 @@ public class DataCollectorService {
|
||||
// 发布WebSocket路由更新事件
|
||||
publishAircraftRouteUpdateEventFromNotification(flightNo, aircraftRoute, routeParams);
|
||||
|
||||
log.info("🚀 事件驱动的路由更新完成: 航班号={}, 路由类型={}", flightNo, routeType);
|
||||
// 标记该航班事件的路由已获取
|
||||
markRouteAsRetrieved(flightNo, routeType, flightNotification.getTime());
|
||||
|
||||
log.info("🚀 事件驱动的路由更新完成: 航班号={}, 路由类型={}, 时间={}", flightNo, routeType, flightNotification.getTime());
|
||||
} else {
|
||||
log.warn("⚠️ 路由数据转换失败: flightNo={}", flightNo);
|
||||
}
|
||||
|
||||
@ -13,6 +13,8 @@ import com.qaup.collision.pathconflict.service.PathConflictDetectionService;
|
||||
import com.qaup.collision.websocket.message.VehicleStatusUpdatePayload;
|
||||
import com.qaup.collision.websocket.event.VehicleStatusUpdateEvent;
|
||||
import com.qaup.collision.datacollector.service.DataCollectorService;
|
||||
import com.qaup.collision.common.model.FlightNotification;
|
||||
import com.qaup.collision.websocket.event.FlightNotificationEvent;
|
||||
import com.qaup.system.domain.SysVehicleInfo;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@ -92,15 +94,16 @@ public class DataProcessingService {
|
||||
|
||||
/**
|
||||
* 周期性数据处理任务
|
||||
*
|
||||
*
|
||||
* 处理步骤：
|
||||
* 1. 从缓存获取最新位置数据
|
||||
* 2. 计算速度和方向
|
||||
* 3. 发送WebSocket位置更新消息
|
||||
* 4. 处理通用车辆状态数据并发送WebSocket更新
|
||||
* 5. 执行路径冲突检测
|
||||
* 6. 执行违规检测
|
||||
* 7. 保存无人车数据到数据库
|
||||
* 5. 处理航班通知数据并发送WebSocket更新
|
||||
* 6. 执行路径冲突检测
|
||||
* 7. 执行违规检测
|
||||
* 8. 保存无人车数据到数据库
|
||||
*/
|
||||
@Scheduled(fixedRateString = "${data.collector.detection.interval:1000}")
|
||||
@Transactional
|
||||
@ -128,13 +131,16 @@ public class DataProcessingService {
|
||||
// 第三步：处理通用车辆状态数据并发送WebSocket更新
|
||||
processUniversalVehicleStatusUpdates();
|
||||
|
||||
// 第四步：执行路径冲突检测
|
||||
// 第四步：处理航班通知数据并发送WebSocket更新
|
||||
processFlightNotificationUpdates();
|
||||
|
||||
// 第五步：执行路径冲突检测
|
||||
pathConflictDetectionService.detectPathConflicts(currentActiveObjects);
|
||||
|
||||
// 第五步：执行违规检测
|
||||
// 第六步：执行违规检测
|
||||
performViolationDetection(currentActiveObjects);
|
||||
|
||||
// 第六步：保存无人车数据到数据库
|
||||
// 第七步：保存无人车数据到数据库
|
||||
saveUnmannedVehicleDataPeriodically(currentActiveObjects);
|
||||
|
||||
log.info("周期性数据处理完成");
|
||||
@ -467,6 +473,52 @@ public class DataProcessingService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理航班通知数据并发送WebSocket更新
|
||||
* 从DataCollectorService获取缓存的航班通知数据，处理后发送WebSocket事件
|
||||
*/
|
||||
private void processFlightNotificationUpdates() {
|
||||
try {
|
||||
// 获取DataCollectorService的航班通知缓存引用
|
||||
DataCollectorService dataCollectorService = applicationContext.getBean(DataCollectorService.class);
|
||||
Map<String, FlightNotification> flightNotificationCache = dataCollectorService.getFlightNotificationCache();
|
||||
|
||||
if (flightNotificationCache.isEmpty()) {
|
||||
log.debug("航班通知缓存为空，跳过处理");
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("开始处理航班通知数据，缓存数量: {}", flightNotificationCache.size());
|
||||
|
||||
for (Map.Entry<String, FlightNotification> entry : flightNotificationCache.entrySet()) {
|
||||
try {
|
||||
FlightNotification notification = entry.getValue();
|
||||
|
||||
// 创建并发布WebSocket事件
|
||||
FlightNotificationEvent event = FlightNotificationEvent.fromFlightNotification(notification);
|
||||
|
||||
if (event != null && event.isValid()) {
|
||||
// 发布WebSocket事件
|
||||
eventPublisher.publishEvent(event);
|
||||
|
||||
log.info("📡 发布航班进出港通知WebSocket事件: 航班号={}, 事件类型={}, 通知级别={}",
|
||||
event.getFlightNo(), event.getEventType(), event.getNotificationLevel());
|
||||
} else {
|
||||
log.warn("⚠️ 航班进出港通知WebSocket事件创建失败或无效: {}", notification.getFlightNo());
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("处理单个航班通知异常: cacheKey={}", entry.getKey(), e);
|
||||
}
|
||||
}
|
||||
|
||||
log.info("航班通知数据处理完成，处理数量: {}", flightNotificationCache.size());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("处理航班通知数据异常", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将 UnmannedVehicle 转换为 VehicleLocation 实体用于持久化
|
||||
*/
|
||||
|
||||
70
test_trigger_route_collection.md
Normal file
70
test_trigger_route_collection.md
Normal file
@ -0,0 +1,70 @@
|
||||
# 航空器路由触发式采集测试
|
||||
|
||||
## 修改摘要
|
||||
|
||||
已成功将航空器路由采集从周期性访问改为触发式访问：
|
||||
|
||||
### 主要变更
|
||||
|
||||
1. **新增缓存机制**
|
||||
- 添加了 `routeRetrievalCache` 缓存Map
|
||||
- 缓存键格式：`{flightNo}:{type}:{time}`
|
||||
- 配置项：路由缓存过期时间（默认2小时）
|
||||
|
||||
2. **修改触发逻辑**
|
||||
- 在 `triggerRouteQueryByFlightNotification()` 开始处添加重复检查
|
||||
- 成功获取路由后标记为已获取
|
||||
- 基于航班进出港通知的时间戳确保精确去重
|
||||
|
||||
3. **禁用周期性采集**
|
||||
- `collectAircraftRouteAndStatus()` 默认禁用
|
||||
- 通过配置项 `data.collector.route.periodic-collection-enabled` 控制
|
||||
- 保留原有方法作为应急备用
|
||||
|
||||
4. **新增缓存管理**
|
||||
- 定时清理过期缓存记录（每小时执行一次）
|
||||
- 详细日志记录便于监控
|
||||
|
||||
### 配置说明
|
||||
|
||||
在 `application.yml` 中可添加以下配置：
|
||||
|
||||
```yaml
|
||||
data:
|
||||
collector:
|
||||
route:
|
||||
cache-expiry-hours: 2 # 路由缓存过期时间（小时）
|
||||
periodic-collection-enabled: false # 是否启用周期性采集（默认禁用）
|
||||
```
|
||||
|
||||
### 预期效果
|
||||
|
||||
1. **避免重复API调用**：同一航班事件（相同航班号+类型+时间戳）的路由只会获取一次
|
||||
2. **事件驱动响应**：完全基于航班进出港通知触发，响应更及时
|
||||
3. **减少系统负载**：取消周期性轮询，减少不必要的API调用
|
||||
4. **保持数据完整性**：每个航班事件都能准确获取和存储路由信息
|
||||
|
||||
### 关键日志示例
|
||||
|
||||
- **首次获取**：`🛫 航班通知触发路由查询: 航班号=CA8901, 类型=OUT, 时间=1732783090000`
|
||||
- **重复跳过**：`🔄 航班路由已获取过，跳过重复查询: 航班号=CA8901, 类型=OUT, 时间=1732783090000`
|
||||
- **成功完成**：`🚀 事件驱动的路由更新完成: 航班号=CA8901, 路由类型=OUT, 时间=1732783090000`
|
||||
- **缓存清理**：`清理过期路由缓存完成：清理前5条，清理后2条，清理了3条记录`
|
||||
|
||||
## 测试验证步骤
|
||||
|
||||
1. **启动系统**：确保 mock_airport.py 正在运行（提供航班进出港通知）
|
||||
2. **观察日志**：查看是否有 "周期性路由采集已禁用" 的调试日志
|
||||
3. **监控触发**：查看航班进出港通知是否正确触发路由查询
|
||||
4. **验证去重**：同一航班事件应该只触发一次路由获取
|
||||
5. **检查缓存**：验证缓存清理机制是否正常工作
|
||||
|
||||
## 回滚方案
|
||||
|
||||
如需恢复周期性采集，在配置文件中设置：
|
||||
```yaml
|
||||
data:
|
||||
collector:
|
||||
route:
|
||||
periodic-collection-enabled: true
|
||||
```
|
||||
@ -102,26 +102,27 @@ AIRCRAFT_SIZE_M = 30.0
|
||||
VEHICLE_SIZE_M = 10.0
|
||||
|
||||
# 航班数据配置 - 用于进出港查询接口（简化版）
|
||||
# 修改为与路由参数配置一致的航班
|
||||
current_time = time.time()
|
||||
flight_data = {
|
||||
"CA8901": { # 出港航班
|
||||
"flightNo": "CA8901",
|
||||
"CA1234": { # 出港航班 - 与路由参数配置一致
|
||||
"flightNo": "CA1234",
|
||||
"type": "OUT",
|
||||
"runway": "35",
|
||||
"contactCross": "L4",
|
||||
"seat": "201",
|
||||
"runway": "17", # 使用路由参数中的inRunway
|
||||
"contactCross": "A2", # 使用路由参数中的contactCross
|
||||
"seat": "201", # 使用路由参数中的seat
|
||||
"cycle_start": current_time,
|
||||
"active_duration": 600, # 10分钟活跃期
|
||||
"gap_duration": 10, # 10秒间隔期
|
||||
"fixed_time": int(current_time * 1000), # 固定的滑出时间
|
||||
"status": "active"
|
||||
},
|
||||
"MU2678": { # 进港航班
|
||||
"flightNo": "MU2678",
|
||||
"MU5123": { # 进港航班 - 与路由参数配置一致
|
||||
"flightNo": "MU5123",
|
||||
"type": "IN",
|
||||
"runway": "17",
|
||||
"contactCross": "B3",
|
||||
"seat": "156",
|
||||
"runway": "35", # 使用路由参数中的inRunway
|
||||
"contactCross": "B3", # 使用路由参数中的contactCross
|
||||
"seat": "156", # 使用路由参数中的seat
|
||||
"cycle_start": current_time,
|
||||
"active_duration": 900, # 15分钟活跃期
|
||||
"gap_duration": 10, # 10秒间隔期
|
||||
@ -170,33 +171,6 @@ aircraft_route_params = {
|
||||
"outRunway": "17",
|
||||
"startSeat": "156"
|
||||
}
|
||||
},
|
||||
# 新增：航班进出港通知中出现的航班配置
|
||||
"CA8901": { # 出港航班
|
||||
"arrival": {
|
||||
"inRunway": "35",
|
||||
"outRunway": "17",
|
||||
"contactCross": "L4",
|
||||
"seat": "201"
|
||||
},
|
||||
"departure": { # 出港参数
|
||||
"inRunway": "35",
|
||||
"outRunway": "17",
|
||||
"startSeat": "201"
|
||||
}
|
||||
},
|
||||
"MU2678": { # 进港航班
|
||||
"arrival": { # 进港参数
|
||||
"inRunway": "17",
|
||||
"outRunway": "35",
|
||||
"contactCross": "B3",
|
||||
"seat": "156"
|
||||
},
|
||||
"departure": {
|
||||
"inRunway": "17",
|
||||
"outRunway": "35",
|
||||
"startSeat": "156"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user