From 243d5734f193598f4279f67bf9d082b9adb23883 Mon Sep 17 00:00:00 2001 From: Tian jianyong <11429339@qq.com> Date: Wed, 24 Sep 2025 12:17:54 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E7=94=B1=E8=88=AA=E7=8F=AD?= =?UTF-8?q?=E8=BF=9B=E5=87=BA=E6=B8=AF=E9=80=9A=E7=9F=A5=E8=A7=A6=E5=8F=91?= =?UTF-8?q?=E8=B7=AF=E7=94=B1=E6=9F=A5=E8=AF=A2=E5=92=8C=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E6=8E=A8=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CLAUDE.md | 91 +++--- doc/requirement/todo_lists.md | 10 +- .../universal_autonomous_vehicle_api.md | 1 + ...sal_autonomous_vehicle_api_min_required.md | 2 +- .../src/main/resources/application-prod.yml | 2 - qaup-admin/src/main/resources/application.yml | 6 - .../service/DataCollectorService.java | 276 ++++++------------ .../service/DataProcessingService.java | 66 ++++- test_trigger_route_collection.md | 70 +++++ tools/{mock_server.py => mock_airport.py} | 48 +-- 10 files changed, 269 insertions(+), 303 deletions(-) create mode 100644 test_trigger_route_collection.md rename tools/{mock_server.py => mock_airport.py} (98%) diff --git a/CLAUDE.md b/CLAUDE.md index 2c1c2094..9a9abd61 100644 --- a/CLAUDE.md +++ b/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 \ No newline at end of file +### 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 \ No newline at end of file diff --git a/doc/requirement/todo_lists.md b/doc/requirement/todo_lists.md index a3308105..58ba1830 100644 --- a/doc/requirement/todo_lists.md +++ b/doc/requirement/todo_lists.md @@ -2,6 +2,10 @@ ### [2025-09-05] +- [x] 模拟收到进出港航班信息，访问路由接口，获取路由信息并保存处理，同时发给前端显示路由 + +### [2025-09-05] + - [ ] 无人车下行接口 - [ ] 控制指令下发 - [ ] 红绿灯通知 @@ -14,4 +18,8 @@ - [ ] 任务执行状态 - [x] 给定航空器路由和无人车路由，计算出可能的冲突点 - PathConflictDetectionService.PathConflictDetectionService() \ No newline at end of file + PathConflictDetectionService.PathConflictDetectionService() + +- [x] 实现精简版无人车状态上报接口规范和实现 + +- [x] 将 mock 服务拆分成机场、无人车、红绿灯三个python 文件 \ No newline at end of file diff --git a/doc/requirement/universal_autonomous_vehicle_api.md b/doc/requirement/universal_autonomous_vehicle_api.md index 0b60e20c..7f8ded0b 100644 --- a/doc/requirement/universal_autonomous_vehicle_api.md +++ b/doc/requirement/universal_autonomous_vehicle_api.md @@ -7,6 +7,7 @@ - **SAE J3016**: 自动驾驶分级标准 - **ISO 21448**: 预期功能安全标准(SOTIF) - **IEEE 2857**: 自动驾驶系统隐私工程标准 +- **AC-137-CA-202X-XX**: 民用机场无人驾驶车辆检测规范（报批稿） ### 1.2 API设计原则 - **RESTful设计**: 遵循REST架构风格 diff --git a/doc/requirement/universal_autonomous_vehicle_api_min_required.md b/doc/requirement/universal_autonomous_vehicle_api_min_required.md index cd40a3b6..ef2b9c78 100644 --- a/doc/requirement/universal_autonomous_vehicle_api_min_required.md +++ b/doc/requirement/universal_autonomous_vehicle_api_min_required.md @@ -1,6 +1,6 @@ # 通用无人车运行状态API（上报必须字段 - 精简版） -本说明仅包含机场部署必须按时上报的最小字段集合，便于联调与验收。完整可选字段请参见 `doc/design/universal_autonomous_vehicle_api.md`。本文件已包含完整的请求方式与可运行示例，使用本文件即可进行开发与测试。 +本说明仅包含机场部署必须按时上报的最小字段集合，便于联调与验收。完整可选字段请参见《无人车通用运行状态API接口》。本文件已包含完整的请求方式与可运行示例，使用本文件即可进行开发与测试。 ## 1. 接口信息 diff --git a/qaup-admin/src/main/resources/application-prod.yml b/qaup-admin/src/main/resources/application-prod.yml index 44a2b309..1886bc55 100644 --- a/qaup-admin/src/main/resources/application-prod.yml +++ b/qaup-admin/src/main/resources/application-prod.yml @@ -176,8 +176,6 @@ xss: data: collector: interval: 250 - route: - interval: 5000 detection: interval: 1000 airport-api: diff --git a/qaup-admin/src/main/resources/application.yml b/qaup-admin/src/main/resources/application.yml index 8afa4701..acd68f26 100644 --- a/qaup-admin/src/main/resources/application.yml +++ b/qaup-admin/src/main/resources/application.yml @@ -197,12 +197,6 @@ data: collector: # 数据采集间隔，单位：毫秒（高频采集保证数据新鲜度） interval: 250 - # 路由数据采集间隔，单位：毫秒（较低频率采集路由信息） - route: - interval: 5000 - # 航班进出港通知采集间隔，单位：毫秒 - flightnotification: - interval: 1000 # 检测和推送间隔配置 detection: # 检测间隔，单位：毫秒（控制围栏检测、冲突检测、违规检测和WebSocket推送频率） diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java index 7fbd19d9..d2fdb9a3 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java @@ -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 activeMovingObjectsCache = new ConcurrentHashMap<>(); + + // 路由获取状态缓存：Key=flightNo:type:time, Value=获取时间戳 + private final Map routeRetrievalCache = new ConcurrentHashMap<>(); + + // 航班通知缓存：Key=flightNo:type, Value=最新的航班通知 + private final Map 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 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); } diff --git a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java index a416313f..d77ef2ee 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java +++ b/qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java @@ -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 flightNotificationCache = dataCollectorService.getFlightNotificationCache(); + + if (flightNotificationCache.isEmpty()) { + log.debug("航班通知缓存为空，跳过处理"); + return; + } + + log.info("开始处理航班通知数据，缓存数量: {}", flightNotificationCache.size()); + + for (Map.Entry 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 实体用于持久化 */ diff --git a/test_trigger_route_collection.md b/test_trigger_route_collection.md new file mode 100644 index 00000000..9bca0784 --- /dev/null +++ b/test_trigger_route_collection.md @@ -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 +``` \ No newline at end of file diff --git a/tools/mock_server.py b/tools/mock_airport.py similarity index 98% rename from tools/mock_server.py rename to tools/mock_airport.py index 81c19fa0..022d3db9 100644 --- a/tools/mock_server.py +++ b/tools/mock_airport.py @@ -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" - } } }