From a98fe11556e15ca1f8da7e67158714ed4ece0b71 Mon Sep 17 00:00:00 2001 From: zhouxiunai <154707516@qq.com> Date: Mon, 21 Jan 2019 18:01:10 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=E6=8F=90=E5=8F=96=E8=BD=AC?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E8=88=AA=E7=8F=AD=E7=9A=84=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scheduled/FlightHisScheduled.java | 179 +++++---------- .../flightInfo/FlightHisServiceImpl.java | 215 ++++++++++++++++++ .../redis/RedisServiceTest.java | 2 +- .../service/KafkaServiceImplTest.java | 43 ++++ 4 files changed, 312 insertions(+), 127 deletions(-) create mode 100644 src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java create mode 100644 src/test/java/com/gzzn/omms/msgexchangeapi/service/KafkaServiceImplTest.java 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 d4c89ee4..d4ad55aa 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java @@ -2,9 +2,9 @@ package com.gzzn.omms.msgexchangeapi.scheduled; import java.util.ArrayList; import java.util.List; -import java.util.Locale; import java.util.Map; +import org.apache.commons.collections.CollectionUtils; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.slf4j.Logger; @@ -14,15 +14,12 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -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.service.flightInfo.FlightHisServiceImpl; import com.gzzn.omms.msgexchangeapi.service.flightInfo.IFlightInfoService; -import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; /** @@ -37,28 +34,9 @@ public class FlightHisScheduled { @Autowired private IFlightInfoService flightInfoService; - /** - * 航班到达超过多久则成为历史航班数据（单位：秒） - */ - @Value("${hstCondition.ARRIVE_HST_TIME}") - private Integer ARRIVE_HST_TIME ; - /** - * 航班取消超过多久则成为历史航班数据（单位：秒） - */ - @Value("${hstCondition.CANCEL_HST_TIME}") - private Integer CANCEL_HST_TIME ; - /** - * 备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：秒） - */ - @Value("${hstCondition.FDIV_HST_TIME}") - private Integer FDIV_HST_TIME ; - - /** - * 计划时间超过多久则成为历史航班数据（单位：秒） - */ - @Value("${hstCondition.SODT_HST_TIME}") - private Integer SODT_HST_TIME ; + @Autowired + private FlightHisServiceImpl flightHisServiceImpl; /** * 历史航班数据索引 */ @@ -75,107 +53,56 @@ public class FlightHisScheduled { @Scheduled(cron = "${scheduled.flightInfoCron}") public void scheduled(){ logger.info("开始采集历史数据..."); - //1.从redis中获取 当前的动态消息 + + List fltrs = flightInfoService.findAll(); - List fltrIdForHistory = new ArrayList(); - List fltrsHistory = new ArrayList(); - - //遍历 查找出已经 可以作为历史数据的动态消息 - //条件为 ：完成运营的航班是指已经落地超过2小时或已经起飞的 - if(null!=fltrs && fltrs.size()>0){ - for(FLTR fltr : fltrs){ - String SODT = fltr.getSODT(); //航班计划时间 ddMMMyyHHmm - Long SODT_long = DateTimeUtil.toDate(SODT, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); - /**===================1.计划时间超时的===================**/ - if((System.currentTimeMillis()-SODT_long)>SODT_HST_TIME){ - //转为历史数据 - fltrsHistory.add(fltr); - continue; - } - /**===================2.航班取消===================**/ - String CNCL = fltr.getCNCL(); //航班取消时间 ddMMMyyHHmm - if(!StringUtils.isEmpty(CNCL)){ //航班被取消 - Long CNCL_longTime = DateTimeUtil.toDate(CNCL, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); - if((System.currentTimeMillis()-CNCL_longTime)>CANCEL_HST_TIME){ - //转为历史数据 - fltrsHistory.add(fltr); - continue; - } - } - /**===================3.备降===================**/ - DIVERSIONDATA FDIV = fltr.getFDIV(); //备降信息 - - if(null!=FDIV && FDIV.getDDES().equals("CTU") && FDIV.getDDIR().equals("FROM")){ //本应降落到成都机场 备降到其他机场 - if((System.currentTimeMillis()-SODT_long)>FDIV_HST_TIME){ - //转为历史数据 - fltrsHistory.add(fltr); - continue; - } - } - - Long todayMixTime = DateTimeUtil.getTodayStartTime(); - String ACTT = fltr.getACTT(); //航班实际时间 ddMMMyyHHmm - if(!StringUtils.isEmpty(ACTT) && SODT_longARRIVE_HST_TIME){ - //转为历史数据 - fltrsHistory.add(fltr); - continue; - } - } - } - } - - - //往 elasticsearch中放入历史数据 通过FLID + SODT 判断唯一 (航班ID+计划时间) - if(!ElasticsearchUtil.isIndexExist(INDEX_NAME)){ - ElasticsearchUtil.createIndex(INDEX_NAME); - } - if(null!=fltrsHistory && fltrsHistory.size()>0){ - 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(null!=listInEs && listInEs.size()>0){ - //获取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(null!=fltrIdForHistory && fltrIdForHistory.size()>0){ - flightInfoService.batchDeleteFltr(fltrIdForHistory); - logger.info("本次删除转历史动态航班数据:{}条",fltrIdForHistory.size()); - } - - logger.info("结束采集历史数据。"); + if( CollectionUtils.isEmpty(fltrs) ){ + return ; } + + List fltrsHistory = new ArrayList(); + fltrsHistory = flightHisServiceImpl.getHisToBeTran(fltrs); + if(CollectionUtils.isEmpty(fltrsHistory)){ + logger.info("没有航班历史需要转换"); + return ; //没有要转的航班历史 + } + + //往 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(null!=listInEs && listInEs.size()>0){ + //获取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(null!=fltrIdForHistory && fltrIdForHistory.size()>0){ + 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 new file mode 100644 index 00000000..0cde60f8 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java @@ -0,0 +1,215 @@ +package com.gzzn.omms.msgexchangeapi.service.flightInfo; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; + +import org.apache.commons.collections.CollectionUtils; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; + +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; + +/** + * 航班历史服务 + * @author zhouxiunai + * + */ +@Service +public class FlightHisServiceImpl { + + /** + * 航班到达超过多久则成为历史航班数据（单位：秒） + */ + @Value("${hstCondition.ARRIVE_HST_TIME}") + private Integer ARRIVE_HST_TIME ; + /** + * 航班取消超过多久则成为历史航班数据（单位：秒） + */ + @Value("${hstCondition.CANCEL_HST_TIME}") + private Integer CANCEL_HST_TIME ; + + /** + * 备降超过多久则成为历史航班数据 FROM CTU 本应降落到成都备降到其他地方（单位：秒） + */ + @Value("${hstCondition.FDIV_HST_TIME}") + private Integer FDIV_HST_TIME ; + + /** + * 计划时间超过多久则成为历史航班数据（单位：秒） + */ + @Value("${hstCondition.SODT_HST_TIME}") + private Integer SODT_HST_TIME ; + + /** + * 获取可以进行转历史操作的航班 + * 遍历 查找出已经 可以作为历史数据的动态消息 + * 条件为 ：完成运营的航班是指已经落地超过2小时或已经起飞的 + * @param fltrs + * @return + */ + public List getHisToBeTran(List fltrs) + { + List fltrsHistory = new ArrayList(); + + if(CollectionUtils.isEmpty(fltrs)){ + return fltrsHistory; + } //end if + + + for(FLTR fltr : fltrs){ + if(isSODTMatch(fltr)) + { + fltrsHistory.add(fltr); + continue; + } + + if(isCNCLMatch(fltr)) + { + fltrsHistory.add(fltr); + continue; + } + + if(isFDIVMatch(fltr)) + { + fltrsHistory.add(fltr); + continue; + } + + if(isDMatch(fltr)) + { + fltrsHistory.add(fltr); + continue; + } + + if(isAMatch(fltr)) + { + fltrsHistory.add(fltr); + continue; + } + } //end for + + return fltrsHistory; + } //end function + + + /** + * 计划时间是否满足转历史要求 + * @param fltr 动态航班信息 + * @return + */ + private boolean isSODTMatch(FLTR fltr) + { + String SODT = fltr.getSODT(); //航班计划时间 ddMMMyyHHmm + Long SODT_long = DateTimeUtil.toDate(SODT, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); + + /**===================1.计划时间超时的===================**/ + if((System.currentTimeMillis()-SODT_long)>SODT_HST_TIME){ + //转为历史数据 + return true; + } + return false; + } + + /** + * 已取消的航班是否满足转历史要求 + * @param fltr + * @return + */ + private boolean isCNCLMatch(FLTR fltr) + { + /**===================2.航班取消===================**/ + String CNCL = fltr.getCNCL(); //航班取消时间 ddMMMyyHHmm + if(!StringUtils.isEmpty(CNCL)){ //航班被取消 + Long CNCL_longTime = DateTimeUtil.toDate(CNCL, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); + if((System.currentTimeMillis()-CNCL_longTime)>CANCEL_HST_TIME){ + //转为历史数据 + return true; + } + } + + return false; + } + + /** + * 备降航班是否满足转历史要求 + * @param fltr + * @return + */ + private boolean isFDIVMatch(FLTR fltr) + { + String SODT = fltr.getSODT(); //航班计划时间 ddMMMyyHHmm + Long SODT_long = DateTimeUtil.toDate(SODT, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); + + /**===================3.备降===================**/ + DIVERSIONDATA FDIV = fltr.getFDIV(); //备降信息 + + if(null!=FDIV && FDIV.getDDES().equals("CTU") && FDIV.getDDIR().equals("FROM")){ //本应降落到成都机场 备降到其他机场 + if((System.currentTimeMillis() - SODT_long)>FDIV_HST_TIME){ + //转为历史数据 + return true; + } + } + + return false; + } + + /** + * 离港航班是否满足转历史要求 + * @param fltr + * @return + */ + private boolean isDMatch(FLTR fltr) + { + String SODT = fltr.getSODT(); //航班计划时间 ddMMMyyHHmm + Long SODT_long = DateTimeUtil.toDate(SODT, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); + + Long todayMixTime = DateTimeUtil.getTodayStartTime(); + String ACTT = fltr.getACTT(); //航班实际时间 ddMMMyyHHmm + if(!StringUtils.isEmpty(ACTT) && SODT_longARRIVE_HST_TIME){ + //转为历史数据 + return true; + } + } + } + + return false; + } +} //end class diff --git a/src/test/java/com/gzzn/omms/msgexchangeapi/redis/RedisServiceTest.java b/src/test/java/com/gzzn/omms/msgexchangeapi/redis/RedisServiceTest.java index 958faf52..fc94932b 100644 --- a/src/test/java/com/gzzn/omms/msgexchangeapi/redis/RedisServiceTest.java +++ b/src/test/java/com/gzzn/omms/msgexchangeapi/redis/RedisServiceTest.java @@ -18,6 +18,6 @@ public class RedisServiceTest { @Test public void removeFlightInfo() { - redisService.del(Arrays.asList("flightInfo:")); + redisService.del(Arrays.asList("flightInfo")); } } diff --git a/src/test/java/com/gzzn/omms/msgexchangeapi/service/KafkaServiceImplTest.java b/src/test/java/com/gzzn/omms/msgexchangeapi/service/KafkaServiceImplTest.java new file mode 100644 index 00000000..c4df7ebe --- /dev/null +++ b/src/test/java/com/gzzn/omms/msgexchangeapi/service/KafkaServiceImplTest.java @@ -0,0 +1,43 @@ +package com.gzzn.omms.msgexchangeapi.service; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.test.context.junit4.SpringRunner; + +@RunWith(SpringRunner.class) +@SpringBootTest +public class KafkaServiceImplTest { + @Autowired + private KafkaTemplate kafkaTemplate; + + /*** + * @param n 字符串长度 + * @return 返回n长度的随机字符串 + ***/ + private String randomString(Integer n) { + String str = "abcdefghijklmnopqrstuvwxyz9876543210"; + + StringBuffer tmp = new StringBuffer(); + for (int i = 0; i < n; i++) { + int beginIndex = (int) Math.floor(Math.random() * str.length()); + String randomStr = str.substring(beginIndex, beginIndex + 1); + tmp.append(randomStr); + } + + return tmp.toString(); + } //end funciont + + + + @Test + public void msgSendTest() { + String topic = "schd"; + String msg = randomString(4*1024*1024); + + + kafkaTemplate.send(topic, msg); + } +}