From eb32e8b1e927b8dddadcf2fa44ca1fdb3c0ee0ea Mon Sep 17 00:00:00 2001 From: zhouxiunai <154707516@qq.com> Date: Fri, 30 Nov 2018 11:30:08 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=88=AA=E7=8F=AD=E5=8A=A8?= =?UTF-8?q?=E6=80=81=E4=BF=A1=E6=81=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../primary/entity/msg/schd/dnld/DnldMsg.java | 12 +- .../secondary/dao/FlightSchdInfoDao.java | 9 ++ .../secondary/entity/FlightSchdInfo.java | 34 +++++ .../service/ExchangeServiceImpl.java | 3 +- .../msgexchangeapi/task/ExchangeTask.java | 87 ++++++++++-- .../omms/msgexchangeapi/utils/JsonUtil.java | 18 ++- .../omms/msgexchangeapi/tools/ToolTest.java | 124 ++++++++++++------ 7 files changed, 222 insertions(+), 65 deletions(-) create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/primary/entity/msg/schd/dnld/DnldMsg.java b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/primary/entity/msg/schd/dnld/DnldMsg.java index f057ff32..2043830d 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/primary/entity/msg/schd/dnld/DnldMsg.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/primary/entity/msg/schd/dnld/DnldMsg.java @@ -3,7 +3,17 @@ package com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.schd.dnld; import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; public class DnldMsg extends Msg { - + public static final String TYPE = "SCHD"; + public static final String SUBTYPE = "DNLD"; + private DnldMsgBody schd; + + public DnldMsgBody getSchd() { + return schd; + } + + public void setSchd(DnldMsgBody schd) { + this.schd = schd; + } } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java new file mode 100644 index 00000000..e81c5c3c --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/dao/FlightSchdInfoDao.java @@ -0,0 +1,9 @@ +package com.gzzn.omms.msgexchangeapi.domain.secondary.dao; + +import org.springframework.data.repository.CrudRepository; + +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightSchdInfo; + +public interface FlightSchdInfoDao extends CrudRepository{ + +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java new file mode 100644 index 00000000..c7376790 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/domain/secondary/entity/FlightSchdInfo.java @@ -0,0 +1,34 @@ +package com.gzzn.omms.msgexchangeapi.domain.secondary.entity; + +import javax.persistence.Entity; +import javax.persistence.Id; +import javax.persistence.Table; + +/** + * 航班计划信息 + * @author + * + */ +@Entity +@Table(name="flightschdinfo") +public class FlightSchdInfo { + + @Id + private String flightId;//航班id,用于索引 + + private String context;//内容 + + + public String getFlightId() { + return flightId; + } + public void setFlightId(String flightId) { + this.flightId = flightId; + } + public String getContext() { + return context; + } + public void setContext(String context) { + this.context = context; + } +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java index 8fb00b12..3bfbb8a5 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/ExchangeServiceImpl.java @@ -22,9 +22,10 @@ public class ExchangeServiceImpl implements IExchangeService { Map map; ObjectMapper xmlMapper = new XmlMapper(); map = xmlMapper.readValue(xml, Map.class); + map.remove("noNamespaceSchemaLocation"); ObjectMapper jsonMapper = new ObjectMapper(); - String json = jsonMapper.writeValueAsString(map); + String json = jsonMapper.writer().withDefaultPrettyPrinter().writeValueAsString(map); return json; } catch (JsonParseException e) { throw new ExchangeServiceException(e.getMessage(),e); diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java b/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java index 3ce81f62..6240efb2 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java @@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchangeapi.task; import java.util.Date; import java.util.List; import java.util.Optional; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -11,12 +12,17 @@ import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import com.gzzn.omms.msgexchangeapi.domain.primary.entity.Cminmsg; +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.Msg; +import com.gzzn.omms.msgexchangeapi.domain.primary.entity.msg.schd.dnld.DnldMsg; import com.gzzn.omms.msgexchangeapi.domain.secondary.dao.ContextDao; +import com.gzzn.omms.msgexchangeapi.domain.secondary.dao.FlightSchdInfoDao; import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.Context; +import com.gzzn.omms.msgexchangeapi.domain.secondary.entity.FlightSchdInfo; import com.gzzn.omms.msgexchangeapi.service.ICminmsgService; import com.gzzn.omms.msgexchangeapi.service.IExchangeService; import com.gzzn.omms.msgexchangeapi.service.IKafkaService; import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; +import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; /** @@ -39,8 +45,11 @@ public class ExchangeTask { @Autowired private ContextDao contextDao; + + @Autowired + private FlightSchdInfoDao flightSchdInfoDao; - @Scheduled(cron="0 0/1 * * * ?") + //@Scheduled(cron="0 0/1 * * * ?") public void corn() { logger.info("定时任务启动...."); @@ -48,29 +57,79 @@ public class ExchangeTask { Context isWaitSchd = contextDao.findOne(Context.KEY_ISWAITSCHD); Context msgProgress = contextDao.findOne(Context.KEY_MSGPROGRESS); - + Long beginId = null; while (isWaitSchd.getValue().equalsIgnoreCase("true")) { - //查找回复的日计划消息 String strDate = msgProgress.getValue(); Date date = DateTimeUtil.toDate(strDate, "yyyy-MM-dd hh:mm:ss"); - List lsCminmsgs = cminmsgService.getNewMsgsAfterDate(date); - Optional opCminmsg = lsCminmsgs.stream().findFirst(); - - //如果找到，添加到内存数据库动态航班信息表 - - //更新状态 - isWaitSchd.setValue("false"); - contextDao.save(isWaitSchd); + Optional opCminmsg = lsCminmsgs + .stream() + .filter(x->{ + String msgBlob = x.getCminmsgsClobMsg(); + Msg msg = exchangeService.xmlstrToObject(msgBlob, Msg.class); + String type = msg.getMeta().getType(); + String subType = msg.getMeta().getStyp(); + return type.equals("SCHD") && subType.equals("RESP"); + }) + .findFirst(); + if(opCminmsg.isPresent()) + { + //如果找到，添加到内存数据库动态航班信息表 + String clobMsg = opCminmsg.get().getCminmsgsClobMsg(); + DnldMsg dnldMsg = exchangeService.xmlstrToObject(clobMsg, DnldMsg.class); + List flightSchdInfo = dnldMsg + .getSchd() + .getFltr() + .stream() + .map(x->{ + FlightSchdInfo node = new FlightSchdInfo(); + node.setFlightId(x.getFlid()); + node.setContext(JsonUtil.getString(x)); + return node; + }).collect(Collectors.toList()); + + flightSchdInfoDao.save(flightSchdInfo); + + // + beginId = opCminmsg.get().getCminmsgsId(); + + //更新状态 + isWaitSchd.setValue("false"); + contextDao.save(isWaitSchd); + } + else { + //等待xx秒继续循环过程 + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + logger.warn("Thread.sleep:暂停线程异常"); + } + } + } // end while 获取航班日记录 + + //获取航班动态消息 + List lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId); + for (Cminmsg cminmsg : lsCminmsgs) { + //更新动态航班消息 + String strClobMsg = cminmsg.getCminmsgsClobMsg(); + Msg msg = exchangeService.xmlstrToObject(strClobMsg, Msg.class); + String type = msg.getMeta().getType(); + String subType = msg.getMeta().getStyp(); + } + //新增或变更 航班动态信息 发送 到kafka的SCHD topic + for (Cminmsg cminmsg : lsCminmsgs) { + //更新动态航班消息 + } + //发送动态消息到kafka + for (Cminmsg cminmsg : lsCminmsgs) { + //更新动态航班消息 + } - - - // logger.info("同步完成"); } } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/utils/JsonUtil.java b/src/main/java/com/gzzn/omms/msgexchangeapi/utils/JsonUtil.java index 8f8c6fe1..56b294c7 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/utils/JsonUtil.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/utils/JsonUtil.java @@ -9,14 +9,18 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectWriter; public class JsonUtil { - public static String getString(Object object) throws JsonProcessingException + public static String getString(Object object) { - ObjectMapper mapper = new ObjectMapper(); - ObjectWriter ow = mapper.writer().withDefaultPrettyPrinter(); - java.lang.String jsonStr = ow.writeValueAsString(object); - - return jsonStr; - } + try { + ObjectMapper mapper = new ObjectMapper(); + ObjectWriter ow = mapper.writer().withDefaultPrettyPrinter(); + java.lang.String jsonStr; + jsonStr = ow.writeValueAsString(object); + return jsonStr; + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } //end function public static T getObject(String str,Class valueType) throws JsonParseException, JsonMappingException, IOException diff --git a/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java b/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java index 8f7f1285..7bce973c 100644 --- a/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java +++ b/src/test/java/com/gzzn/omms/msgexchangeapi/tools/ToolTest.java @@ -3,7 +3,10 @@ package com.gzzn.omms.msgexchangeapi.tools; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.junit.Test; import org.junit.runner.RunWith; @@ -35,56 +38,93 @@ public class ToolTest { @Autowired private IExchangeService exchangeService; + /** + * 每种消息类型导出一个xml文件即可 + */ @Test - public void test() { + public void exportOneTypeToJsonFiles() { + List lsCmin = (List) cminmsgDao.findAll(); + Map mapCminmsg = new HashMap(); + for(Cminmsg x : lsCmin) + { + Msg msg = exchangeService.xmlstrToObject(x.getCminmsgsClobMsg(),Msg.class); + mapCminmsg.putIfAbsent(msg.getMeta().getType() + "_" + msg.getMeta().getStyp(), x); + } + + List ls = new ArrayList(mapCminmsg.values()); + toJsonFiles(ls); } - @Test + /** + * 导出CMINMSGS数据中的消息到单独的Xml文件 + * @throws JsonParseException + * @throws JsonMappingException + * @throws IOException + */ public void exportToXmlFiles() throws JsonParseException, JsonMappingException, IOException { List lsCmin = (List) cminmsgDao.findAll(); - for(Cminmsg x : lsCmin) - { - String json = exchangeService.xmlToJson(x.getCminmsgsClobMsg()); - Msg msg =JsonUtil.getObject(json,Msg.class); - String Dttm = msg.getMeta().getDttm(); - - String dir = "xml/" + Dttm.substring(0, 10); - - StringBuilder path = new StringBuilder(); - path.append(dir); - path.append("/"); - path.append(msg.getMeta().getDttm().substring(10, Dttm.length()));//time - path.append("-"+x.getCminmsgsId()); - path.append("-"+msg.getMeta().getType());//Type - path.append("-"+msg.getMeta().getStyp());//Subtype - path.append(".xml"); - - //父目录是否存在 - File fileDir = new File(dir); - if(fileDir.exists() == false) - { - fileDir.mkdirs(); - } - - // - File f = new File(path.toString()); - try (FileOutputStream fop = new FileOutputStream(f)) { - byte[] contentInBytes = x.getCminmsgsClobMsg().getBytes("UTF-8"); - - fop.write(contentInBytes); - fop.flush(); - fop.close(); - } - catch (ExchangeServiceException e) { - logger.error(e.getMessage()); - } - catch (IOException e) { - logger.error(e.getMessage()); - } - } + toJsonFiles(lsCmin); } //end function + + /** + * 生成xml文件 + * @param lsCmin + */ + private void toJsonFiles(List lsCmin) + { + try { + for(Cminmsg x : lsCmin) + { + String json = exchangeService.xmlToJson(x.getCminmsgsClobMsg()); + Msg msg; + + msg = JsonUtil.getObject(json,Msg.class); + String Dttm = msg.getMeta().getDttm(); + + String dir = "json"; + + StringBuilder path = new StringBuilder(); + path.append(dir); + path.append("/"); + path.append(msg.getMeta().getType());//Type + path.append("-"+msg.getMeta().getStyp());//Subtype + path.append(".json"); + + //父目录是否存在 + File fileDir = new File(dir); + if(fileDir.exists() == false) + { + fileDir.mkdirs(); + } + + // + File f = new File(path.toString()); + try (FileOutputStream fop = new FileOutputStream(f)) { + String jsonString = exchangeService.xmlToJson(x.getCminmsgsClobMsg()); + jsonString = jsonString.replace("\"noNamespaceSchemaLocation\":\"unisysaodbsis.xsd\",", ""); + byte[] contentInBytes = jsonString.getBytes("UTF-8"); + + fop.write(contentInBytes); + fop.flush(); + fop.close(); + } + catch (ExchangeServiceException e) { + logger.error(e.getMessage()); + } + catch (IOException e) { + logger.error(e.getMessage()); + } + } + } catch (JsonParseException e1) { + throw new RuntimeException(e1.getMessage()); + } catch (JsonMappingException e1) { + throw new RuntimeException(e1); + } catch (IOException e1) { + throw new RuntimeException(e1); + } + } }