From 5d96d4f2b62599a5f43ab61c1f2125918a9d115e Mon Sep 17 00:00:00 2001 From: zhouxiunai <154707516@qq.com> Date: Fri, 25 Jan 2019 05:08:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E5=8A=A8=E6=80=81=E8=88=AA?= =?UTF-8?q?=E7=8F=AD=E8=BD=AC=E5=8E=86=E5=8F=B2=E5=8A=9F=E8=83=BD=EF=BC=8C?= =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=8F=92=E5=85=A5es=E7=9A=84=E5=BC=82?= =?UTF-8?q?=E5=B8=B8=E6=8D=95=E8=8E=B7=E5=8F=8A=E6=97=A5=E5=BF=97=E6=89=93?= =?UTF-8?q?=E5=8D=B0.=E4=BF=AE=E5=A4=8DFDIV=E7=BB=93=E6=9E=84=E5=92=8CES?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E4=B8=8D=E7=AC=A6=E5=AF=BC=E8=87=B4=E7=9A=84?= =?UTF-8?q?=E6=8F=92=E5=85=A5=E5=BC=82=E5=B8=B8=E9=97=AE=E9=A2=98=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scheduled/FlightHisScheduled.java | 83 ++++++------------- .../flightInfo/FlightHisServiceImpl.java | 57 ++++++++++++- src/main/resources/application-dev.yml | 14 ++-- src/main/resources/application-pro.yml | 8 +- .../flightInfo/FlightHisServiceImplTest.java | 19 +++++ 5 files changed, 113 insertions(+), 68 deletions(-) diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java b/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java index 394ce8bd..ffb48868 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java @@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchangeapi.scheduled; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; @@ -36,14 +37,7 @@ public class FlightHisScheduled { @Autowired private FlightHisServiceImpl flightHisServiceImpl; - /** - * 历史航班数据索引 - */ - private final String INDEX_NAME = "flight_hts"; - /** - * 历史航班数据类型 - */ - private final String ES_TYPE = "_doc"; + private final Logger logger = LoggerFactory.getLogger(FlightHisScheduled.class); /** @@ -53,56 +47,33 @@ public class FlightHisScheduled { public void scheduled(){ logger.info("开始采集历史数据..."); - - List fltrs = flightInfoService.findAll(); - if( CollectionUtils.isEmpty(fltrs) ){ - return ; + try { + List fltrs = flightInfoService.findAll(); + if( CollectionUtils.isEmpty(fltrs) ){ + return ; + } + + List fltrsHistory = new ArrayList(); + fltrsHistory = flightHisServiceImpl.getHisToBeTran(fltrs); + if(CollectionUtils.isEmpty(fltrsHistory)){ + logger.info("没有航班历史需要转换"); + return ; //没有要转的航班历史 + } + + logger.info("获取到历史动态航班数据:{}条,准备转历史",fltrsHistory.size()); + + //航班历史写入ELK + flightHisServiceImpl.save(fltrsHistory); + + //从redis中删除历史航班 + List fltrIdForHistory = fltrsHistory.stream().map(node->{ + return node.getFLID().toString(); + }).collect(Collectors.toList()); + flightInfoService.batchDeleteFltr(fltrIdForHistory); + } catch (Exception e) { + logger.error("动态航班转历史错误:{}",e.getMessage()); } - List fltrsHistory = new ArrayList(); - fltrsHistory = flightHisServiceImpl.getHisToBeTran(fltrs); - if(CollectionUtils.isEmpty(fltrsHistory)){ - logger.info("没有航班历史需要转换"); - return ; //没有要转的航班历史 - } - - logger.info("获取到历史动态航班数据:{}条,准备转历史",fltrsHistory.size()); - - //往 elasticsearch中放入历史数据 通过FLID + SODT 判断唯一 (航班ID+计划时间) - if(!ElasticsearchUtil.isIndexExist(INDEX_NAME)){ - ElasticsearchUtil.createIndex(INDEX_NAME); - } - - //航班历史写入ELK - List fltrIdForHistory = new ArrayList(); - for(FLTR fltr:fltrsHistory){ - JSONObject jsondata = JSONObject.parseObject(JsonUtil.getString(fltr)); - //判断是否已经拥有该历史数据 因es类型原因请注意类型转换 - String SODT = fltr.getSODT(); - Long FLID = fltr.getFLID().longValue(); - BoolQueryBuilder boolQuery = QueryBuilders.boolQuery() - .must(QueryBuilders.termQuery("SODT",SODT)) - .must(QueryBuilders.termQuery("FLID",FLID)); - List> listInEs = ElasticsearchUtil.searchListData(INDEX_NAME, ES_TYPE, boolQuery, null, "FLID", null, null, null); - if(!CollectionUtils.isEmpty(listInEs)){ - //获取es中的id 并更新当前历史数据 - //唯一 获取第一个 - Map map = listInEs.get(0); - String id = (String) map.get("id"); - ElasticsearchUtil.updateDataById(jsondata, INDEX_NAME, ES_TYPE, id); - fltrIdForHistory.add(fltr.getFLID().toString()); - }else{ - //新增历史数据 - ElasticsearchUtil.addData(jsondata, INDEX_NAME, ES_TYPE); - fltrIdForHistory.add(fltr.getFLID().toString()); - } - } - - //删除redis中已成为历史数据的动态消息 - if(!CollectionUtils.isEmpty(fltrIdForHistory)){ - flightInfoService.batchDeleteFltr(fltrIdForHistory); - logger.info("本次删除转历史动态航班数据:{}条",fltrIdForHistory.size()); - } logger.info("结束采集历史数据。"); } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java index 91f47034..bc2781de 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java @@ -3,16 +3,24 @@ package com.gzzn.omms.msgexchangeapi.service.flightInfo; import java.util.ArrayList; import java.util.List; import java.util.Locale; +import java.util.Map; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import com.alibaba.fastjson.JSONObject; +import com.gzzn.omms.msgexchangeapi.elasticsearch.ElasticsearchUtil; import com.gzzn.omms.msgexchangeapi.entity.msg.DIVERSIONDATA; import com.gzzn.omms.msgexchangeapi.entity.msg.MOVEMENTINDICATOR; import com.gzzn.omms.msgexchangeapi.entity.msg.SCHD.FLTR; import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; +import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; /** * 航班历史服务 @@ -21,7 +29,8 @@ import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; */ @Service public class FlightHisServiceImpl { - + private final Logger logger = LoggerFactory.getLogger(FlightHisServiceImpl.class); + /** * 航班到达超过多久则成为历史航班数据（单位：秒） */ @@ -45,6 +54,16 @@ public class FlightHisServiceImpl { @Value("${hstCondition.SODT_HST_TIME}") private Integer SODT_HST_TIME ; + + /** + * 历史航班数据索引 + */ + private final String INDEX_NAME = "flight_hts"; + /** + * 历史航班数据类型 + */ + private final String ES_TYPE = "_doc"; + /** * 获取可以进行转历史操作的航班 * 遍历 查找出已经 可以作为历史数据的动态消息 @@ -97,6 +116,42 @@ public class FlightHisServiceImpl { } //end function + /** + * 保存航班历史到ELK + * @param fltrs + * @return + */ + public boolean save(List fltrs) + { + for(FLTR fltr : fltrs){ + JSONObject jsondata = JSONObject.parseObject(JsonUtil.getString(fltr)); + //判断是否已经拥有该历史数据 因es类型原因请注意类型转换 + String SODT = fltr.getSODT(); + Long FLID = fltr.getFLID().longValue(); + BoolQueryBuilder boolQuery = QueryBuilders.boolQuery() + .must(QueryBuilders.termQuery("SODT",SODT)) + .must(QueryBuilders.termQuery("FLID",FLID)); + List> listInEs = ElasticsearchUtil.searchListData(INDEX_NAME, ES_TYPE, boolQuery, null, "FLID", null, null, null); + + try { + if(!CollectionUtils.isEmpty(listInEs)){ + //更新 + Map map = listInEs.get(0); + String id = (String) map.get("id"); + ElasticsearchUtil.updateDataById(jsondata, INDEX_NAME, ES_TYPE, id); + }else{ + //新增 + ElasticsearchUtil.addData(jsondata, INDEX_NAME, ES_TYPE); + } + } catch (Exception e) { + logger.error("插入或更新动态航班历史错误:{}",e.getMessage()); + } + } + return false; + } + + + /** * 计划时间是否满足转历史要求 * @param fltr 动态航班信息 diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 3ac4fff1..bc31c038 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -69,14 +69,14 @@ scheduled: flightSendCron: "*/3 * * * * ?" #计入历史航班数据的条件 hstCondition: - #航班到达超过多久则成为历史航班数据（单位：秒） - ARRIVE_HST_TIME: 3600 + #航班到达超过多久则成为历史航班数据（单位：毫秒） + ARRIVE_HST_TIME: 3600000 #航班取消超过多久则成为历史航班数据（单位：秒） - CANCEL_HST_TIME: 3600 - #备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：秒） - FDIV_HST_TIME: 59400 - #计划时间超过多久则成为历史航班数据（单位：秒） - SODT_HST_TIME: 259200 + CANCEL_HST_TIME: 3600000 + #备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：毫秒） + FDIV_HST_TIME: 59400000 + #计划时间超过多久则成为历史航班数据（单位：毫秒） + SODT_HST_TIME: 259200000 # Elasticsearch # 9200端口是用来让HTTP REST API来访问ElasticSearch，而9300端口是传输层监听的默认端口 diff --git a/src/main/resources/application-pro.yml b/src/main/resources/application-pro.yml index 53c7ca1e..732fdbd3 100644 --- a/src/main/resources/application-pro.yml +++ b/src/main/resources/application-pro.yml @@ -70,13 +70,13 @@ scheduled: #计入历史航班数据的条件 hstCondition: #航班到达超过多久则成为历史航班数据（单位：秒） - ARRIVE_HST_TIME: 3600 + ARRIVE_HST_TIME: 3600000 #航班取消超过多久则成为历史航班数据（单位：秒） - CANCEL_HST_TIME: 3600 + CANCEL_HST_TIME: 3600000 #备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：秒） - FDIV_HST_TIME: 59400 + FDIV_HST_TIME: 59400000 #计划时间超过多久则成为历史航班数据（单位：秒） - SODT_HST_TIME: 259200 + SODT_HST_TIME: 259200000 # Elasticsearch # 9200端口是用来让HTTP REST API来访问ElasticSearch，而9300端口是传输层监听的默认端口 diff --git a/src/test/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImplTest.java b/src/test/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImplTest.java index 75253fca..68995967 100644 --- a/src/test/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImplTest.java +++ b/src/test/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImplTest.java @@ -2,6 +2,8 @@ package com.gzzn.omms.msgexchangeapi.service.flightInfo; import static org.assertj.core.api.Assertions.assertThat; +import java.util.List; + import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; @@ -18,6 +20,9 @@ public class FlightHisServiceImplTest { @Autowired private FlightHisServiceImpl flightHisServiceImpl; + @Autowired + private IFlightInfoService flightInfoService; + @Test public void testIsCNCLMatch() { @@ -28,4 +33,18 @@ public class FlightHisServiceImplTest { //flightHisServiceImpl.isAMatch(fltr); } + + @Test + public void testGetHisToBeTran() + { + List fltrs = flightInfoService.findAll(); + System.out.println("所有航班动态条数:"+fltrs.size()); + + List fltrsHis = flightHisServiceImpl.getHisToBeTran(fltrs); + System.out.println("等待转历史航班动态条数:"+fltrsHis.size()); + + flightHisServiceImpl.save(fltrsHis); + + assertThat(fltrsHis).isNotEmpty(); + } }