From 0717201c585940f23f12d6a6e4f09d97185c54ef Mon Sep 17 00:00:00 2001 From: shan <1653261938@qq.com> Date: Wed, 24 Dec 2025 18:21:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=201.=E5=8E=BB=E6=8E=89redis=E4=B8=80?= =?UTF-8?q?=E7=9B=B4=E7=BC=93=E5=AD=98=E6=97=A0=E7=94=A8=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=B3=A8=E8=A7=A32.=E5=AE=9E=E7=8E=B0adapter=E4=B8=8E=E4=B8=BB?= =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=88=9D=E6=AD=A5=E6=8E=A5=E6=94=B6=E5=A4=84?= =?UTF-8?q?=E7=90=86=E6=8A=A5=E6=96=873=E4=B8=BB=E7=A8=8B=E5=BA=8F?= =?UTF-8?q?=E6=8E=A5=E6=94=B6websocket=E6=8A=A5=E6=96=87=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E5=88=86=E7=89=87=E6=8E=A5=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- adxp-adapter/pom.xml | 11 + .../com/qaup/adxp/adapter/dto/AllXml.java | 398 ++++++++++++++++++ .../qaup/adxp/adapter/dto/FlightMessage.java | 9 + .../adxp/adapter/service/AdxpSdkService.java | 266 +++++++++++- .../websocket/AdxpWebSocketHandler.java | 13 + doc/requirement/official_api.md | 3 +- .../common/model/FlightNotification.java | 6 +- .../config/ADXPWebSocketConfig.java | 21 +- .../datacollector/dao/DataCollectorDao.java | 75 ++-- .../service/DataCollectorService.java | 49 ++- .../AdxpFlightServiceWebSocketClient.java | 238 ++++++++++- .../websocket/AdxpWebSocketStarter.java | 20 + .../service/DataProcessingService.java | 34 ++ .../impl/LocationRuleQueryServiceImpl.java | 2 +- 14 files changed, 1071 insertions(+), 74 deletions(-) create mode 100644 adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/AllXml.java create mode 100644 qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpWebSocketStarter.java diff --git a/adxp-adapter/pom.xml b/adxp-adapter/pom.xml index 15035e7d..94c3590b 100644 --- a/adxp-adapter/pom.xml +++ b/adxp-adapter/pom.xml @@ -108,6 +108,17 @@ dom4j 1.6.1 + + org.junit.jupiter + junit-jupiter + test + + + org.springframework.boot + spring-boot-starter-test + test + + diff --git a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/AllXml.java b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/AllXml.java new file mode 100644 index 00000000..bd506c58 --- /dev/null +++ b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/AllXml.java @@ -0,0 +1,398 @@ +package com.qaup.adxp.adapter.dto; + +public class AllXml { + // 将原始的 XML 字符串存储为常量 + public static final String XML_ARR = "\n" + + "\n" + + " \n" + + " ADXP_NAOMS_O_DYN_ARR\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " \n" + + " be58e345-7bcf-4cbb-915f-28c16f0e09a4\n" + + " 20171208140658867\n" + + " 20171208140658867\n" + + " \n" + + " \n" + + " \n" + + " 4155920\n" + + " CZ3158-A-20170906230500\n" + + " TAO\n" + + " ARR\n" + + " 20170906230500\n" + + " \n" + + " \n" + + ""; + + + // + public static final String XML_CRAFTSEAT = "\n" + + "\n" + + " \n" + + " ADXP_NAOMS_O_DYN_CRAFTSEAT\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " \n" + + " 9a6b67e0-f061-4d0d-9a6a-f31328283d54\n" + + " 20171208140658867\n" + + " 20171208140658867\n" + + " \n" + + " \n" + + " \n" + + " 4155920\n" + + " CZ3158-D-20170906230500\n" + + " \n" + + " \n" + + " 32\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + ""; + + // 存储 XML 字符串作为常量 + public static final String XML_DFIE = "\n" + + "\n" + + " \n" + + " ADXP_NAOMS_O_DYN_DFIE\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " \n" + + " 97b79684-f755-41a1-82ab-f7e34352c6af\n" + + " 20171208140658867\n" + + " 20171208140658867\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 4155920\n" + + " CZ3158-A-20170906230500\n" + + " 4155935\n" + + " CZ3158-D-20170906230500\n" + + " 20170906\n" + + " 20170906\n" + + " 20170906230500\n" + + " SC\n" + + " 4753\n" + + " D\n" + + " A\n" + + " CRJ200\n" + + " B738L\n" + + " W/Z\n" + + " 1\n" + + " T1\n" + + " 1\n" + + " 一级特勤保障\n" + + " 1\n" + + " \n" + + " \n" + + " TAO\n" + + " ALT\n" + + " 20170906232200\n" + + " 20170906232500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 1\n" + + " \n" + + " \n" + + " HHA\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 20170906232200\n" + + " \n" + + " \n" + + " 1\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " SC\n" + + " 1234\n" + + " SC1234\n" + + " \n" + + " \n" + + " MU\n" + + " 2345\n" + + " MU2345\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " SC\n" + + " 1234\n" + + " SC1234\n" + + " D\n" + + " TAO\n" + + " HGH\n" + + " SZX\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " \n" + + ""; + + public static final String XML_DFDL ="\n" + + "\n" + + "\n" + + " ADXP_NAOMS_O_DYN_DFDL\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " ADXP\n" + + " FIDS\n" + + " \n" + + " 20171208140658867\n" + + " 20171208140658867\n" + + "\n" + + "\n" + + "5-3\n" + + " \n" + + " \n" + + " 4155920\n" + + " CZ3158-A-20170906230500\n" + + " 4155935\n" + + " CZ3158-D-20170906230500\n" + + " 20170906\n" + + " 20170906\n" + + " 20170906230500\n" + + " SC\n" + + " 4753\n" + + " D\n" + + " A\n" + + " CRJ200\n" + + " B738L\n" + + " W/Z\n" + + " 1\n" + + " T1\n" + + " 1\n" + + " 一级特勤保障\n" + + " 1\n" + + " BOR\n" + + " BOR\n" + + " \n" + + " \n" + + " TAO\n" + + " ALT\n" + + " 20170906232200\n" + + " 20170906232500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 1\n" + + " \n" + + " \n" + + " HHA\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 20170906232200\n" + + " \n" + + " \n" + + " 1\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " SC\n" + + " 1234\n" + + " SC1234\n" + + " \n" + + " \n" + + " MU\n" + + " 2345\n" + + " MU2345\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " SC\n" + + " 1234\n" + + " SC1234\n" + + " D\n" + + " TAO\n" + + " HGH\n" + + " SZX\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " TAO\n" + + " D\n" + + " CAN\n" + + " 取消\n" + + " 1001\n" + + " 20170906234200\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " D\n" + + " 20170920173100\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920175500\n" + + " 20170920175800\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " E35\n" + + " D\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " CZ\n" + + " 0701\n" + + " N\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 32\n" + + " D\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 32\n" + + " D\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 32\n" + + "12\n" + + " D\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 32\n" + + " 20170920173500\n" + + " 20170920171500\n" + + " 20170920173500\n" + + " 20170920173500\n" + + " \n" + + " \n" + + " \n" + + " \n" + + "\n" + + ""; + + + public static final String XML_RUNWAY = "\n" + + "\n" + + " \n" + + " " + + "\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " \n" + + " \n" + + " 20200331103923314\n" + + " 20200331103923314\n" + + " \n" + + " \n" + + " \n" + + " QDCDM\n" + + " SC8775-D-20200331132500\n" + + " 35L\n" + + " \n" + + " \n" + + " \n" + + ""; + + public static final String XML_AXOT = "\n" + + "\n" + + " \n" + + " ADXP_NAOMS_O_CDM_AXOT\n" + + " 1.0\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " \n" + + " \n" + + " 20200331103923314\n" + + " 20200331103923314\n" + + " \n" + + " \n" + + " \n" + + " HCDM\n" + + " SC8775-D-20200331132500\n" + + " 202003311401\n" + + " \n" + + " \n" + + ""; + + public static final String XML_TISFLIGHT = "\n" + + "\n" + + " \n" + + " ADXP_NAOMS_O_DYN_TISFLIGHT\n" + + " 1.0\n" + + " ADXP\n" + + " naomsdevc\n" + + " \n" + + " 5b2fc9a7-c40b-4d53-8267-9e9ff36cb7a8\n" + + " 20251110133327781\n" + + " 20251110133327781\n" + + " \n" + + " \n" + + " \n" + + " \n" + + " 41317B75FC8311BCE0633D411CAC4CEB\n" + + " CDG4969\n" + + " CDG\n" + + " ZSQD\n" + + " ZLXY\n" + + " 20251104123500\n" + + " 20251104145000\n" + + " \n" + + " \n" + + " 132\n" + + " 道口22\n" + + " 前3后3;\n" + + " WFG\n" + + " \n" + + " \n" + + " DEP\n" + + " APE\n" + + " TXI\n" + + " \n" + + " \n" + + " \n" + + " \n" + + ""; +} diff --git a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/FlightMessage.java b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/FlightMessage.java index 4c676342..43f914aa 100644 --- a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/FlightMessage.java +++ b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/dto/FlightMessage.java @@ -37,4 +37,13 @@ public class FlightMessage { public void setContent(String content) { this.content = content; } + + @Override + public String toString() { + return "FlightMessage{" + + "serviceCode='" + serviceCode + '\'' + + ", actionCode='" + actionCode + '\'' + + ", content='" + content + '\'' + + '}'; + } } diff --git a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/service/AdxpSdkService.java b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/service/AdxpSdkService.java index 5906c54c..517ed578 100644 --- a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/service/AdxpSdkService.java +++ b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/service/AdxpSdkService.java @@ -15,13 +15,32 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.w3c.dom.Document; +import org.w3c.dom.Element; +import org.w3c.dom.NodeList; + import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; +import javax.xml.bind.JAXBContext; +import javax.xml.bind.JAXBException; +import javax.xml.bind.Marshaller; +import javax.xml.parsers.DocumentBuilder; +import javax.xml.parsers.DocumentBuilderFactory; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.io.StringWriter; import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; + +import org.w3c.dom.*; +import org.xml.sax.SAXException; + +import javax.xml.parsers.*; +import java.io.*; + @Service public class AdxpSdkService { @@ -223,7 +242,6 @@ public class AdxpSdkService { try { MessageResult result = adxpClient.receiveMessage(); - // 使用统一的错误码处理方法 if (handleMessageResultError(result)) { return Collections.emptyList(); @@ -235,17 +253,24 @@ public class AdxpSdkService { List messages = new ArrayList<>(); MessageList messageList = result.getMessageList(); - + + // 处理XML字符串 + List messageStringList = result.getMessageStringList(); + + if (messageList != null && messageList.getMsg() != null) { - for (MsgType msg : messageList.getMsg()) { + for (int i = 0; i < messageList.getMsg().size(); i++) { + MsgType msg = messageList.getMsg().get(i); // 获取 MsgType + String messageString = messageStringList.get(i); // 获取对应的字符串 HeadType head = msg.getHead(); Object body = msg.getBody(); if (head != null && body != null) { + log.info("本条消息的接收到消息头{}，消息体{}", head.getSvcServiceCode(), messageString); FlightMessage flightMessage = new FlightMessage( head.getSvcServiceCode(), null, // actionCode 在 body 中 - body.toString() + messageString ); messages.add(flightMessage); } @@ -270,8 +295,46 @@ public class AdxpSdkService { /** * 处理接收到的消息并广播到WebSocket客户端 */ - private void processMessages(List messages) { + private void processMessages(List messages) throws ParserConfigurationException, IOException, SAXException { if (messages != null && !messages.isEmpty()) { + DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); + for (FlightMessage message : messages){ + String txt = message.getContent(); + log.info("接收到的XML字符串：{}", txt); + // 解析 XML 字符串 + DocumentBuilder builder = factory.newDocumentBuilder(); + InputStream is = new ByteArrayInputStream(txt.getBytes("UTF-8")); + Document document = builder.parse(is); + + log.info("事件的类型是{}",message.getServiceCode()); + switch (message.getServiceCode()){ + case "ADXP_NAOMS_O_DYN_DFDL": + handleDFDL(document); + break; + case "ADXP_NAOMS_O_DYN_CRAFTSEAT": + handleCRAFTSEAT(document); + break; + case "ADXP_NAOMS_O_DYN_ARR": + handleARR(document); + break; + case "ADXP_NAOMS_O_DYN_DFIE": + handleDFIE(document); + break; + case "ADXP_NAOMS_O_CDM_AXOT": + handleAXOT(document); + break; + case "ADXP_NAOMS_O_CDM_RUNWAY": + handleRUNWAY(document); + break; + case "ADXP_NAOMS_O_DYN_TISFLIGHT": + handleTISFLIGHT(document); + break; + default: + System.out.println("未定义的消息"); + break; + } + } + try { log.info("接收到 {} 条消息，准备广播", messages.size()); // 将消息广播到WebSocket客户端 @@ -371,4 +434,197 @@ public class AdxpSdkService { return health; } + + /** + *向数据中台发送 动态航班请求事件RQDF + */ + public MessageResult sendMessage(List message) { + if (!connected) { + log.error("ADXP数据中台未连接，无法发送消息"); + return null; + } + String xml = "\n" + + "\n" + + "\n" + + " ADXP_NAOMS_O_DYN_RQDF\n" + + " 1.0\n" + + " ADXP\n" + + " FIDS\n" + + " ADXP\n" + + " NAOMS\n" + + " \n" + + " 2017120814065886\n" + + " 20171208140658867\n" + + "\n" + + "\n" + + " \n" + + " 20171209\n" + + " \n" + + "\n" + + ""; + List xmls = new ArrayList<>(); + xmls.add( xml); + adxpClient.sendMessage(xmls); + return null; + } + + /** + * 处理 动态航班响应事件DFDL 消息 + * @param document + */ + private static void handleDFDL(Document document) { + // 获取 元素下的 + NodeList dynFlightNodes = document.getElementsByTagName("DynFlight"); + if (dynFlightNodes.getLength() > 0) { + Element dynFlight = (Element) dynFlightNodes.item(0); + + // 提取所需的值 + String flightId = getElementValue(dynFlight, "FlightId"); + String inOut = getElementValue(dynFlight, "InOut"); + String runwayNum = getElementValue(dynFlight, "RunwayNum"); + + // 输出提取的值 + System.out.println("FlightId: " + flightId); + System.out.println("InOut: " + inOut); + System.out.println("RunwayNum: " + runwayNum); + + // 获取 CraftseatList 并提取 Code + NodeList craftseatListNodes = dynFlight.getElementsByTagName("CraftseatList"); + if (craftseatListNodes.getLength() > 0) { + Element craftseatList = (Element) craftseatListNodes.item(0); + NodeList craftseatNodes = craftseatList.getElementsByTagName("Craftseat"); + + for (int i = 0; i < craftseatNodes.getLength(); i++) { + Element craftseat = (Element) craftseatNodes.item(i); + + // 提取 Code + String code = getElementValue(craftseat, "Code"); + + // 输出提取的值 + System.out.println("Craftseat Code: " + code); + } + } + } + } + + /** + * 处理 运行跑道响应事件RUNWAY 消息 + * @param document + */ + private static void handleRUNWAY(Document document) { + // 获取 元素 + NodeList flightList = document.getElementsByTagName("Flight"); + if (flightList.getLength() > 0) { + Element flightElement = (Element) flightList.item(0); + + // 提取 BizKey + String bizKey = getElementValue(flightElement, "BizKey"); + System.out.println("BizKey: " + bizKey); + + // 提取 RUNWAYDEP + String runwayDep = getElementValue(flightElement, "RUNWAYDEP"); + System.out.println("RUNWAYDEP: " + runwayDep); + + // 提取 RUNWAYARR + String runwayArr = getElementValue(flightElement, "RUNWAYARR"); + System.out.println("RUNWAYARR: " + runwayArr); + } + } + private static void handleCRAFTSEAT(Document document) { + // 获取 元素 + NodeList dynFlightList = document.getElementsByTagName("DynFlight"); + if (dynFlightList.getLength() > 0) { + Element dynFlightElement = (Element) dynFlightList.item(0); + + // 提取 FlightId + String flightId = getElementValue(dynFlightElement, "FlightId"); + System.out.println("FlightId: " + flightId); + + // 获取 CraftseatList 元素 + NodeList craftseatList = dynFlightElement.getElementsByTagName("Craftseat"); + if (craftseatList.getLength() > 0) { + Element craftseatElement = (Element) craftseatList.item(0); + + // 提取 Code + String code = getElementValue(craftseatElement, "Code"); + System.out.println("Code: " + code); + } + } + } + private static void handleARR(Document document) { + // 获取 元素 + NodeList dynFlightNodes = document.getElementsByTagName("DynFlight"); + if (dynFlightNodes.getLength() > 0) { + Element dynFlight = (Element) dynFlightNodes.item(0); + + // 提取所需的值 + String flightId = getElementValue(dynFlight, "FlightId"); + String bizKey = getElementValue(dynFlight, "BizKey"); + System.out.println("FlightId: " + flightId); + System.out.println("BizKey: " + bizKey); + } + } + private static void handleAXOT(Document document) { + // 获取 元素 + NodeList bodyNodes = document.getElementsByTagName("Body"); + if (bodyNodes.getLength() > 0) { + Element body = (Element) bodyNodes.item(0); + + // 获取 元素 + NodeList flightNodes = body.getElementsByTagName("Flight"); + if (flightNodes.getLength() > 0) { + Element flight = (Element) flightNodes.item(0); + + // 提取 BizKey 元素的值 + String bizKey = getElementValue(flight, "BizKey"); + System.out.println("BizKey: " + bizKey); + } + } + } + private static void handleDFIE(Document document) { + // 获取 元素 + NodeList dynFlightNodes = document.getElementsByTagName("DynFlight"); + if (dynFlightNodes.getLength() > 0) { + Element dynFlight = (Element) dynFlightNodes.item(0); + // 提取所需的值 + String flightId = getElementValue(dynFlight, "FlightId"); + String inOut = getElementValue(dynFlight, "InOut"); + String runwayNum = getElementValue(dynFlight, "RunwayNum"); + String bizKey = getElementValue(dynFlight, "BizKey"); + + System.out.println("FlightId: " + flightId); + System.out.println("InOut: " + inOut); + System.out.println("RunwayNum: " + runwayNum); + System.out.println("BizKey: " + bizKey); + } + } + + private static void handleTISFLIGHT(Document document) { + // 获取 元素 + NodeList tisFlightNodes = document.getElementsByTagName("TisFlight"); + if (tisFlightNodes.getLength() > 0) { + Element tisFlight = (Element) tisFlightNodes.item(0); + + // 提取所需的值 + String fuId = getElementValue(tisFlight, "FuId"); + String state = getElementValue(tisFlight, "State"); + String contactCross = getElementValue(tisFlight, "ContactCross"); + + // 打印提取的内容 + System.out.println("FuId: " + fuId); + System.out.println("State: " + state); + System.out.println("ContactCross: " + contactCross); + } + } + + + // 获取元素的值 + private static String getElementValue(Element parentElement, String tagName) { + NodeList nodeList = parentElement.getElementsByTagName(tagName); + if (nodeList.getLength() > 0) { + return nodeList.item(0).getTextContent(); + } + return null; + } + } diff --git a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/websocket/AdxpWebSocketHandler.java b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/websocket/AdxpWebSocketHandler.java index dc02107d..25306539 100644 --- a/adxp-adapter/src/main/java/com/qaup/adxp/adapter/websocket/AdxpWebSocketHandler.java +++ b/adxp-adapter/src/main/java/com/qaup/adxp/adapter/websocket/AdxpWebSocketHandler.java @@ -2,6 +2,7 @@ package com.qaup.adxp.adapter.websocket; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; +import com.qaup.adxp.adapter.dto.AllXml; import com.qaup.adxp.adapter.dto.FlightMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,6 +42,18 @@ public class AdxpWebSocketHandler extends TextWebSocketHandler { // 可以处理一些控制命令，比如心跳 if ("ping".equals(message.getPayload())) { session.sendMessage(new TextMessage("pong")); + List messages = new CopyOnWriteArrayList<>(); + FlightMessage mgARR = new FlightMessage("ADXP_NAOMS_O_DYN_ARR", "pong", AllXml.XML_ARR); + messages.add(mgARR); + FlightMessage mgDFIE = new FlightMessage("ADXP_NAOMS_O_DYN_DFIE", "pong", AllXml.XML_DFIE); + messages.add(mgDFIE); + FlightMessage mgRUNWAY = new FlightMessage("ADXP_NAOMS_O_CDM_RUNWAY", "pong", AllXml.XML_RUNWAY); + messages.add(mgRUNWAY); + FlightMessage mgCRAFTSEAT = new FlightMessage("ADXP_NAOMS_O_DYN_CRAFTSEAT", "pong", AllXml.XML_CRAFTSEAT); + messages.add(mgCRAFTSEAT); + FlightMessage mgTISFLIGHT = new FlightMessage("ADXP_NAOMS_O_DYN_TISFLIGHT", "pong", AllXml.XML_TISFLIGHT); + messages.add(mgTISFLIGHT); + broadcastMessages(messages); log.debug("发送心跳响应到客户端: sessionId={}", session.getId()); } } diff --git a/doc/requirement/official_api.md b/doc/requirement/official_api.md index 08a40d51..122cb857 100644 --- a/doc/requirement/official_api.md +++ b/doc/requirement/official_api.md @@ -19,7 +19,8 @@ 数据来源：接入并转发从空管接收到的融合数据 -1. 接口地址： +1. 接口地址： + 弃用接口地址： 2. 请求方式：get，需要在 Header 中携带认证信息，字段名为 Authorization，值为认证接口返回的token diff --git a/qaup-collision/src/main/java/com/qaup/collision/common/model/FlightNotification.java b/qaup-collision/src/main/java/com/qaup/collision/common/model/FlightNotification.java index efcabc73..7acb897c 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/common/model/FlightNotification.java +++ b/qaup-collision/src/main/java/com/qaup/collision/common/model/FlightNotification.java @@ -134,8 +134,10 @@ public class FlightNotification { * 检查通知是否有效 */ public boolean isValid() { +// return flightNo != null && !flightNo.trim().isEmpty() +// && type != null +// && eventTime != null; return flightNo != null && !flightNo.trim().isEmpty() - && type != null - && eventTime != null; + && type != null; } } \ No newline at end of file diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/config/ADXPWebSocketConfig.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/config/ADXPWebSocketConfig.java index 7fdc9e66..d5385ec6 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/config/ADXPWebSocketConfig.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/config/ADXPWebSocketConfig.java @@ -6,12 +6,31 @@ import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.client.WebSocketClient; import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import java.util.HashMap; +import java.util.Map; + @Configuration public class ADXPWebSocketConfig { @Bean public WebSocketClient webSocketClient() { - return new StandardWebSocketClient(); + StandardWebSocketClient client = new StandardWebSocketClient(); + + Map props = new HashMap<>(); + // ====== JSR-356 标准配置（一定生效）====== + props.put("javax.websocket.maxTextMessageBufferSize", 10 * 1024 * 1024); // 10MB + props.put("javax.websocket.maxBinaryMessageBufferSize", 10 * 1024 * 1024); + + // ====== Tomcat 实现专用（有则更稳）====== + props.put("org.apache.tomcat.websocket.textBufferSize", 10 * 1024 * 1024); + props.put("org.apache.tomcat.websocket.binaryBufferSize", 10 * 1024 * 1024); + + client.setUserProperties(props); + + client.setUserProperties(props); + + return client; + //return new StandardWebSocketClient(); } @Bean diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java index db53ee71..62ccabd2 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java @@ -354,44 +354,43 @@ public class DataCollectorDao { * @return 航班路由参数数据 */ public AircraftRouteParamsDTO getAircraftRouteParams(String flightNo, String routeType) { - try { - String token = authService.getToken(airportBaseUrl); - if (token == null) { - log.error("无法获取有效的认证token"); - return null; - } - - String url = UriComponentsBuilder.fromUriString(airportBaseUrl) - .path("/aircraftRouteParamsController/getRouteParams") - .queryParam("flightNo", flightNo) - .queryParam("routeType", routeType) - .toUriString(); - - HttpHeaders headers = new HttpHeaders(); - headers.set("Authorization", token); - headers.set("Content-Type", "application/json"); - - HttpEntity requestEntity = new HttpEntity<>(headers); - - ResponseEntity> response = restTemplate.exchange( - url, - HttpMethod.GET, - requestEntity, - new ParameterizedTypeReference>() {} - ); - - Response responseBody = response.getBody(); - if (responseBody != null && responseBody.getStatus() == 200) { - log.info("成功获取航班路由参数: flightNo={}, routeType={}", flightNo, routeType); - return responseBody.getData(); - } else { - log.warn("获取航班路由参数失败: {}", responseBody != null ? responseBody.getMsg() : "未知错误"); - return null; - } - } catch (Exception e) { - log.error("获取航班路由参数时发生异常: flightNo={}, routeType={}", flightNo, routeType, e); - return null; - } +// try { +// String token = authService.getToken(airportBaseUrl); +// if (token == null) { +// log.error("无法获取有效的认证token"); +// return null; +// } +// +// String url = UriComponentsBuilder.fromUriString(airportBaseUrl) +// .path("/aircraftRouteParamsController/getRouteParams") +// .queryParam("flightNo", flightNo) +// .queryParam("routeType", routeType) +// .toUriString(); +// +// HttpHeaders headers = new HttpHeaders(); +// headers.set("Authorization", token); +// headers.set("Content-Type", "application/json"); +// +// HttpEntity requestEntity = new HttpEntity<>(headers); +// ResponseEntity> response = restTemplate.exchange( +// url, +// HttpMethod.GET, +// requestEntity, +// new ParameterizedTypeReference>() {} +// ); +// Response responseBody = response.getBody(); +// if (responseBody != null && responseBody.getStatus() == 200) { +// log.info("成功获取航班路由参数: flightNo={}, routeType={}", flightNo, routeType); +// return responseBody.getData(); +// } else { +// log.warn("获取航班路由参数失败: {}", responseBody != null ? responseBody.getMsg() : "未知错误"); +// return null; +// } +// } catch (Exception e) { +// log.error("获取航班路由参数时发生异常: flightNo={}, routeType={}", flightNo, routeType, e); +// return null; +// } + return null; } /** 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 caec7462..5e122533 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 @@ -137,7 +137,9 @@ public class DataCollectorService { return; } - log.info("通过WebSocket接收到 {} 条航班进出港通知", notifications.size()); + //log.info("通过WebSocket接收到 {} 条航班进出港通知", notifications.size()); + log.error("通过WebSocket接收到 {} 条航班进出港通知hhhhhhhhhhhhhhh", notifications.size()); + log.error("通过WebSocket接收到 {} 条航班进出港通知hhhhhhhhhhhhhhh", notifications); // 将DTO转换为业务对象并处理 for (FlightNotificationDTO dto : notifications) { @@ -681,6 +683,7 @@ public class DataCollectorService { if (adxpFlightServiceWebSocketClient != null && adxpFlightServiceWebSocketClient.isConnected()) { // WebSocket客户端已连接，消息通过WebSocket实时接收 log.debug("使用WebSocket客户端接收航班通知"); + return; } @@ -778,9 +781,32 @@ public class DataCollectorService { */ 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()); + + // 检查缓存中是否已有该键值 + FlightNotification existingNotification = flightNotificationCache.get(cacheKey); + if (existingNotification != null) { + // 合并通知：新的有值的字段覆盖旧的，没有值的字段延用旧的 + FlightNotification mergedNotification = FlightNotification.builder() + .flightNo(notification.getFlightNo()) + .type(notification.getType()) + .runway(notification.getRunway() != null ? notification.getRunway() : existingNotification.getRunway()) + .contactCross(notification.getContactCross() != null ? notification.getContactCross() : existingNotification.getContactCross()) + .seat(notification.getSeat() != null ? notification.getSeat() : existingNotification.getSeat()) + .eventTime(notification.getEventTime() != null ? notification.getEventTime() : existingNotification.getEventTime()) + .eventDateTime(notification.getEventDateTime() != null ? notification.getEventDateTime() : existingNotification.getEventDateTime()) + .receivedTime(existingNotification.getReceivedTime()) // 保持接收时间不变 + .active(notification.getActive() != null ? notification.getActive() : existingNotification.getActive()) + .build(); + + flightNotificationCache.put(cacheKey, mergedNotification); + log.info("合并航班通知缓存: flightNo={}, type={}, 缓存数量={}", + notification.getFlightNo(), notification.getType(), flightNotificationCache.size()); + } else { + // 缓存中没有该键值，直接放入 + flightNotificationCache.put(cacheKey, notification); + log.info("新增航班通知缓存: flightNo={}, type={}, 缓存数量={}", + notification.getFlightNo(), notification.getType(), flightNotificationCache.size()); + } } /** @@ -813,13 +839,13 @@ public class DataCollectorService { } // 清理 flightNotificationCache - 使用 eventTime 字段判断 - int sizeBefore = flightNotificationCache.size(); - flightNotificationCache.entrySet().removeIf(entry -> { - FlightNotification notification = entry.getValue(); - long eventTime = (notification.getEventTime() != null) ? notification.getEventTime() : 0; - return (now - eventTime) > inactiveThreshold; - }); - removedFlightNotifications = sizeBefore - flightNotificationCache.size(); +// int sizeBefore = flightNotificationCache.size(); +// flightNotificationCache.entrySet().removeIf(entry -> { +// FlightNotification notification = entry.getValue(); +// long eventTime = (notification.getEventTime() != null) ? notification.getEventTime() : 0; +// return (now - eventTime) > inactiveThreshold; +// }); +// removedFlightNotifications = sizeBefore - flightNotificationCache.size(); if (removedMovingObjects > 0 || removedFlightNotifications > 0) { log.info("清理不活跃缓存对象: {} 个移动对象, {} 个航班通知; 剩余: {} 个移动对象, {} 个航班通知", @@ -842,3 +868,4 @@ public class DataCollectorService { log.info("缓存已清理"); } } + diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java index a15784a0..2f765ae6 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java @@ -4,9 +4,12 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import com.qaup.collision.datacollector.config.FlightSdkProperties; import com.qaup.collision.datacollector.dto.FlightNotificationDTO; +import com.qaup.collision.dataprocessing.service.DataProcessingService; +import com.qaup.common.core.redis.RedisCache; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketHandler; @@ -38,6 +41,12 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { private Thread reconnectThread; + @Autowired + private RedisCache redisCache; + + @Autowired + private DataProcessingService dataProcessingService; + public AdxpFlightServiceWebSocketClient(WebSocketClient webSocketClient, FlightSdkProperties properties) { this.webSocketClient = webSocketClient; @@ -49,7 +58,7 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { /** * 启动WebSocket客户端 */ - @PostConstruct + //@PostConstruct public void start() { if (isRunning.compareAndSet(false, true)) { log.info("启动ADXP航班通知WebSocket客户端"); @@ -75,6 +84,7 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { public void connect() { if (!properties.isConfigurationReady()) { log.warn("数据中台航班 SDK 配置不完整，WebSocket客户端将无法正常工作"); + scheduleReconnect(); return; } @@ -159,20 +169,48 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { isConnected.set(true); messageCount.set(0); // 重置消息计数 } - + + // 【新增】用于拼接 WebSocket 分片消息 + private final StringBuilder partialMessageBuffer = new StringBuilder(1024 * 1024); @Override public void handleMessage(@NonNull WebSocketSession session, @NonNull WebSocketMessage message) throws Exception { - if (message instanceof TextMessage) { - handleTextMessage(session, (TextMessage) message); + + if (!(message instanceof TextMessage textMessage)) { + return; } + + //每来一片就拼 + partialMessageBuffer.append(textMessage.getPayload()); + + //不是最后一片，直接等 + if (!textMessage.isLast()) { + return; + } + + //最后一片，才算一条完整消息 + String fullPayload = partialMessageBuffer.toString(); + partialMessageBuffer.setLength(0); + + try { + // 👇 用你原来的逻辑处理 + handleTextMessage(session, fullPayload); + } catch (Exception e) { + log.error("处理完整 WebSocket 消息失败", e); + } + + +// if (message instanceof TextMessage) { +// handleTextMessage(session, (TextMessage) message); +// } } - public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { + public void handleTextMessage(WebSocketSession session, String payload) throws Exception { try { - // 解析接收到的消息 - String payload = message.getPayload(); +// // 解析接收到的消息 +// String payload = message.getPayload(); + log.debug("📥 收到WebSocket消息 ({} bytes)", payload.length()); - + // 增加消息计数 messageCount.incrementAndGet(); @@ -181,7 +219,8 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { if (!notifications.isEmpty()) { log.info("🛬 接收到 {} 条航班通知", notifications.size()); - + log.error("🛬 接收到 {} 条航班通知", notifications.size()); + // 通知所有监听器 for (Consumer> listener : messageListeners) { try { @@ -227,7 +266,7 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { @Override public boolean supportsPartialMessages() { - return false; + return true; } /** @@ -238,11 +277,17 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { // 解析FlightMessage列表 List flightMessages = objectMapper.readValue(jsonContent, new TypeReference>() {}); + + log.error("📥 收到WebSocket消息 ({})", flightMessages); + + List notifications = new java.util.ArrayList<>(); for (FlightMessage message : flightMessages) { try { + log.error("hhhhh获取消息内容: serviceCode={}", message.getServiceCode()); + log.error("hhhhh获取消息内容: content={}", message.getContent()); FlightNotificationDTO dto = parseXmlMessage(message.getServiceCode(), message.getContent()); if (dto != null) { notifications.add(dto); @@ -252,7 +297,7 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { errorCount.incrementAndGet(); } } - + log.error("notifications={}", notifications); return notifications; } catch (Exception e) { log.error("解析消息失败", e); @@ -272,13 +317,20 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { switch (serviceCode) { case "ADXP_NAOMS_O_DYN_ARR": - return parseArrival(root); + //return parseArrival(root); + return handleARR(root); case "ADXP_NAOMS_O_CDM_AXOT": return parsePushback(root); case "ADXP_NAOMS_O_CDM_RUNWAY": - return parseRunway(root); + //return parseRunway(root); + return handleRUNWAY(root); case "ADXP_NAOMS_O_DYN_CRAFTSEAT": - return parseGate(root); + //return parseGate(root); + return handleCRAFTSEAT(root); + case "ADXP_NAOMS_O_DYN_DFIE": + return handleDFIE(root); + case "ADXP_NAOMS_O_DYN_TISFLIGHT": + return handleTISFLIGHT(root); default: log.debug("未知的服务代码: {}", serviceCode); return null; @@ -312,7 +364,7 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { return dto; } - + private FlightNotificationDTO parsePushback(org.w3c.dom.Element root) { String flightNo = getTextContent(root, "FlightNumber"); String actualPushback = getTextContent(root, "ActualPushback"); @@ -439,4 +491,160 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler { errorCount.get(), isRunning.get()); } + + + //处理ARR事件 + private FlightNotificationDTO handleARR(org.w3c.dom.Element root) { + String flightNo = getTextContent(root, "BizKey"); + //分解BizKey + String[] arr = flightNo.split("-", 3); // 最多切 3 段 + FlightNotificationDTO dto = new FlightNotificationDTO(); + dto.setFlightNo(flightNo); + dto.setType("IN"); + // 存储到Redis + if (arr.length >= 3) { + try { + String key = "flight:" + arr[0]; // 使用航班号作为键 + redisCache.setCacheMapValue(key, "flightNumber", arr[0]); + redisCache.setCacheMapValue(key, "type", arr[1]); + redisCache.setCacheMapValue(key, "time", arr[2]); + log.info("成功将航班数据存储到Redis: flightNumber={}, type={}, time={}", arr[0], arr[1], arr[2]); + } catch (Exception e) { + log.error("存储航班数据到Redis失败: {}", e.getMessage()); + } + } + //假装模拟请求接口 + dataProcessingService.hello(arr[0]); + return dto; + } + + private FlightNotificationDTO handleDFIE(org.w3c.dom.Element root) { + String flightNo = getTextContent(root, "BizKey"); + String RunwayNum = getTextContent(root, "RunwayNum"); + String InOut = getTextContent(root, "InOut"); + //分解BizKey + String[] arr = flightNo.split("-", 3); // 最多切 3 段 + + FlightNotificationDTO dto = new FlightNotificationDTO(); + dto.setFlightNo(flightNo); + if (InOut.equals("A")){ + dto.setType("IN"); + }else { + dto.setType("OUT"); + } + dto.setRunway(RunwayNum); + + // 存储到Redis + if (arr.length >= 3) { + try { + String key = "flight:" + arr[0]; // 使用航班号作为键 + redisCache.setCacheMapValue(key, "flightNumber", arr[0]); + redisCache.setCacheMapValue(key, "type", arr[1]); + redisCache.setCacheMapValue(key, "time", arr[2]); + if (InOut.equals("A")){ + redisCache.setCacheMapValue(key, "inRunway", RunwayNum); + }else { + redisCache.setCacheMapValue(key, "outRunway", RunwayNum); + } + log.info("成功将航班数据存储到Redis: flightNumber={}, type={}, time={}", arr[0], arr[1], arr[2]); + } catch (Exception e) { + log.error("存储航班数据到Redis失败: {}", e.getMessage()); + } + } + + return dto; + } + + //处理RUNWAY事件 + private FlightNotificationDTO handleRUNWAY(org.w3c.dom.Element root) { + String flightNo = getTextContent(root, "BizKey"); + String outRunway = getTextContent(root, "RUNWAYDEP"); + String inRunway = getTextContent(root, "RUNWAYARR"); + if (inRunway.equals("")){ + log.error("inRunway为空字符串"); + } + + log.error("inRunway-{}",inRunway); + log.error("outRunway-{}",outRunway); + //分解BizKey + String[] arr = flightNo.split("-", 3); // 最多切 3 段 + FlightNotificationDTO dto = new FlightNotificationDTO(); + dto.setFlightNo(flightNo); + if (arr[1].equals("A")){ + dto.setType("IN"); + }else { + dto.setType("OUT"); + } + // 存储到Redis + if (arr.length >= 3) { + try { + String key = "flight:" + arr[0]; // 使用航班号作为键 + redisCache.setCacheMapValue(key, "flightNumber", arr[0]); + redisCache.setCacheMapValue(key, "type", arr[1]); + redisCache.setCacheMapValue(key, "time", arr[2]); + if (!inRunway.equals("")){ + redisCache.setCacheMapValue(key, "inRunway", inRunway); + } + if (!outRunway.equals("")){ + redisCache.setCacheMapValue(key, "outRunway", outRunway); + } + log.info("成功将航班数据存储到Redis: flightNumber={}, type={}, time={}", arr[0], arr[1], arr[2]); + } catch (Exception e) { + log.error("存储航班数据到Redis失败: {}", e.getMessage()); + } + } + //假装模拟请求接口 + dataProcessingService.hello(arr[0]); + return dto; + } + + //处理CRAFTSEAT事件 + private FlightNotificationDTO handleCRAFTSEAT(org.w3c.dom.Element root) { + String flightNo = getTextContent(root, "BizKey"); + String code = getTextContent(root, "Code"); + log.error("seat-{}",code); + //分解BizKey + String[] arr = flightNo.split("-", 3); // 最多切 3 段 + FlightNotificationDTO dto = new FlightNotificationDTO(); + dto.setFlightNo(flightNo); + // 存储到Redis + if (arr.length >= 3) { + try { + String key = "flight:" + arr[0]; // 使用航班号作为键 + redisCache.setCacheMapValue(key, "flightNumber", arr[0]); + redisCache.setCacheMapValue(key, "type", arr[1]); + redisCache.setCacheMapValue(key, "time", arr[2]); + redisCache.setCacheMapValue(key, "seat", code); + log.info("成功将航班数据存储到Redis: flightNumber={}, type={}, time={}", arr[0], arr[1], arr[2]); + } catch (Exception e) { + log.error("存储航班数据到Redis失败: {}", e.getMessage()); + } + } + //假装模拟请求接口 + dataProcessingService.hello(arr[0]); + return dto; + } + + private FlightNotificationDTO handleTISFLIGHT(org.w3c.dom.Element root) { + String flightNumber = getTextContent(root, "FlNo"); + String contactCross = getTextContent(root, "ContactCross"); + + FlightNotificationDTO dto = new FlightNotificationDTO(); + dto.setFlightNo(flightNumber); + dto.setType("IN"); + + String key = "flight:" + flightNumber; + + try { + redisCache.setCacheMapValue(key, "flightNumber", flightNumber); + redisCache.setCacheMapValue(key, "contactCross",contactCross); + } catch (Exception e) { + log.error("存储航班数据到Redis失败: {}", e.getMessage()); + } + + return dto; + } + + + } \ No newline at end of file diff --git a/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpWebSocketStarter.java b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpWebSocketStarter.java new file mode 100644 index 00000000..d18e6f47 --- /dev/null +++ b/qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpWebSocketStarter.java @@ -0,0 +1,20 @@ +package com.qaup.collision.datacollector.websocket; + +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.stereotype.Component; + +@Component +public class AdxpWebSocketStarter { + + private final AdxpFlightServiceWebSocketClient client; + + public AdxpWebSocketStarter(AdxpFlightServiceWebSocketClient client) { + this.client = client; + } + + @EventListener(ApplicationReadyEvent.class) + public void startWhenReady() { + client.start(); + } +} 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 97be3ca0..a9f6d17b 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 @@ -15,6 +15,7 @@ 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.common.core.redis.RedisCache; import com.qaup.system.domain.SysVehicleInfo; import lombok.extern.slf4j.Slf4j; @@ -87,6 +88,9 @@ public class DataProcessingService { @Autowired private ApplicationContext applicationContext; + @Autowired + private RedisCache redisCache; + // 从DataCollectorService获取缓存的引用 private Map activeMovingObjectsCache; @@ -488,6 +492,10 @@ public class DataProcessingService { DataCollectorService dataCollectorService = applicationContext.getBean(DataCollectorService.class); Map flightNotificationCache = dataCollectorService.getFlightNotificationCache(); + // 打印缓存数量 + log.error("航班通知缓存数量: {}", flightNotificationCache.size()); + log.error("航班通知缓存内容: {}", flightNotificationCache); + if (flightNotificationCache.isEmpty()) { log.debug("航班通知缓存为空，跳过处理"); return; @@ -1268,4 +1276,30 @@ public class DataProcessingService { log.error("❌ 发布航空器路由更新事件失败: flightNo={}", flightNo, e); } } + + /** + * 触发查询航班号对应的路由 + */ + public void hello(String flightNo){ + log.error("hello666666666666666666666666666666666666"); + String key = "flight:"+flightNo; + + String time =(String) redisCache.getCacheMapValue(key, "time"); + log.error("time是{}",time); + String inRunway = "35"; + String outRunway = "34"; + String contactCross = "F1"; + String seat = "138"; + + // 获取DataCollectorDao进行路由数据查询 + DataCollectorDao dataCollectorDao = applicationContext.getBean(DataCollectorDao.class); + AircraftRouteDTO routeData = null; + // 进港路由查询 + routeData = dataCollectorDao.getArrivalRoute( + inRunway, + outRunway, + contactCross, + seat); + log.error("routeData是{}",routeData); + } } \ No newline at end of file diff --git a/qaup-collision/src/main/java/com/qaup/collision/rule/service/impl/LocationRuleQueryServiceImpl.java b/qaup-collision/src/main/java/com/qaup/collision/rule/service/impl/LocationRuleQueryServiceImpl.java index ac0c8ee3..71b12270 100644 --- a/qaup-collision/src/main/java/com/qaup/collision/rule/service/impl/LocationRuleQueryServiceImpl.java +++ b/qaup-collision/src/main/java/com/qaup/collision/rule/service/impl/LocationRuleQueryServiceImpl.java @@ -74,7 +74,7 @@ public class LocationRuleQueryServiceImpl implements LocationRuleQueryService { // ============================================ @Override - @Cacheable(value = "locationRules", key = "#location.toString() + '_' + #vehicleType + '_' + #timestamp.toString()") + //@Cacheable(value = "locationRules", key = "#location.toString() + '_' + #vehicleType + '_' + #timestamp.toString()") public List findApplicableRules(Point location, MovingObjectType vehicleType, LocalDateTime timestamp) { try { logger.debug("查询位置适用规则: location={}, vehicleType={}, timestamp={}",