windy/msgexchange-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

完善动态航班转历史功能，完善插入es的异常捕获及日志打印.修复FDIV结构和ES结构不符导致的插入异常问题。

Browse Source
This commit is contained in:
zhouxiunai
2019-01-25 05:08:43 +08:00
parent c715a135de
commit 5d96d4f2b6
5 changed files with 113 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java
+27 -56
View File
@@ -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<FLTR> fltrs = flightInfoService.findAll();
if( CollectionUtils.isEmpty(fltrs) ){
return ;
try {
List<FLTR> fltrs = flightInfoService.findAll();
if( CollectionUtils.isEmpty(fltrs) ){
return ;
}
List<FLTR> fltrsHistory = new ArrayList<FLTR>();
fltrsHistory = flightHisServiceImpl.getHisToBeTran(fltrs);
if(CollectionUtils.isEmpty(fltrsHistory)){
logger.info("没有航班历史需要转换");
return ; //没有要转的航班历史
}
logger.info("获取到历史动态航班数据:{}条,准备转历史",fltrsHistory.size());
//航班历史写入ELK
flightHisServiceImpl.save(fltrsHistory);
//从redis中删除历史航班
List<String> fltrIdForHistory = fltrsHistory.stream().map(node->{
return node.getFLID().toString();
}).collect(Collectors.toList());
flightInfoService.batchDeleteFltr(fltrIdForHistory);
} catch (Exception e) {
logger.error("动态航班转历史错误:{}",e.getMessage());
}
List<FLTR> fltrsHistory = new ArrayList<FLTR>();
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<String> fltrIdForHistory = new ArrayList<String>();
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<Map<String, Object>> listInEs = ElasticsearchUtil.searchListData(INDEX_NAME, ES_TYPE, boolQuery, null, "FLID", null, null, null);
if(!CollectionUtils.isEmpty(listInEs)){
//获取es中的id 并更新当前历史数据
//唯一 获取第一个
Map<String, Object> 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("结束采集历史数据。");
}
src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImpl.java
+56 -1
View File
@@ -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<FLTR> 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<Map<String, Object>> listInEs = ElasticsearchUtil.searchListData(INDEX_NAME, ES_TYPE, boolQuery, null, "FLID", null, null, null);
try {
if(!CollectionUtils.isEmpty(listInEs)){
//更新
Map<String, Object> 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 动态航班信息
src/main/resources/application-dev.yml
+7 -7
View File
@@ -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端口是传输层监听的默认端口
src/main/resources/application-pro.yml
+4 -4
View File
@@ -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端口是传输层监听的默认端口
src/test/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightHisServiceImplTest.java
+19
View File
@@ -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<FLTR> fltrs = flightInfoService.findAll();
System.out.println("所有航班动态条数:"+fltrs.size());
List<FLTR> fltrsHis = flightHisServiceImpl.getHisToBeTran(fltrs);
System.out.println("等待转历史航班动态条数:"+fltrsHis.size());
flightHisServiceImpl.save(fltrsHis);
assertThat(fltrsHis).isNotEmpty();
}
}