diff --git a/pom.xml b/pom.xml index 12f63651..13ad97e0 100644 --- a/pom.xml +++ b/pom.xml @@ -93,12 +93,39 @@ spring-boot-starter-data-redis - + net.logstash.logback logstash-logback-encoder 5.2 + + + org.elasticsearch + elasticsearch + 6.4.2 + + + + + org.elasticsearch.client + transport + 6.4.2 + + + org.elasticsearch + elasticsearch + + + + + + + com.alibaba + fastjson + 1.2.15 + + diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/config/ScheduledAsyncConfig.java b/src/main/java/com/gzzn/omms/msgexchangeapi/config/ScheduledAsyncConfig.java new file mode 100644 index 00000000..32cbb633 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/config/ScheduledAsyncConfig.java @@ -0,0 +1,38 @@ +package com.gzzn.omms.msgexchangeapi.config; + +import java.util.concurrent.Executor; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/** + * 定时器多线程配置 + * @author fhc + * + */ +@Configuration +@EnableAsync +public class ScheduledAsyncConfig { + + /* + 此处成员变量应该使用@Value从配置中读取 + */ + @Value("${scheduled.corePoolSize}") + private int corePoolSize; + @Value("${scheduled.maxPoolSize}") + private int maxPoolSize; + @Value("${scheduled.queueCapacity}") + private int queueCapacity; + @Bean + public Executor taskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(corePoolSize); + executor.setMaxPoolSize(maxPoolSize); + executor.setQueueCapacity(queueCapacity); + executor.initialize(); + return executor; + } +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchConfig.java b/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchConfig.java new file mode 100644 index 00000000..11ab89d8 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchConfig.java @@ -0,0 +1,69 @@ +package com.gzzn.omms.msgexchangeapi.elasticsearch; + +import java.net.InetAddress; + +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.transport.TransportAddress; +import org.elasticsearch.transport.client.PreBuiltTransportClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class ElasticsearchConfig { + + private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfig.class); + + /** + * elk集群地址 + */ + @Value("${elasticsearch.ip}") + private String hostName; + + /** + * 端口 + */ + @Value("${elasticsearch.port}") + private String port; + + /** + * 集群名称 + */ + @Value("${elasticsearch.cluster.name}") + private String clusterName; + + /** + * 连接池 + */ + @Value("${elasticsearch.pool}") + private String poolSize; + + /** + * Bean name default 函数名字 + * + * @return + */ + @Bean(name = "transportClient") + public TransportClient transportClient() { + LOGGER.info("Elasticsearch初始化开始。。。。。"); + TransportClient transportClient = null; + try { + // 配置信息 + Settings esSetting = Settings.builder() + .put("cluster.name", clusterName) // 集群名字 +// .put("client.transport.sniff", true)// 增加嗅探机制，找到ES集群 (当前api无需嗅探，而且此处开启嗅探会导致链接失败) + .put("thread_pool.search.size", Integer.parseInt(poolSize))// 增加线程池个数，暂时设为5 + .build(); + // 配置信息Settings自定义 + transportClient = new PreBuiltTransportClient(esSetting); + TransportAddress transportAddress = new TransportAddress(InetAddress.getByName(hostName), Integer.valueOf(port)); + transportClient.addTransportAddresses(transportAddress); + } catch (Exception e) { + LOGGER.error("elasticsearch TransportClient create error!!", e); + } + return transportClient; + } +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchUtil.java b/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchUtil.java new file mode 100644 index 00000000..ac3f07e1 --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/elasticsearch/ElasticsearchUtil.java @@ -0,0 +1,300 @@ +package com.gzzn.omms.msgexchangeapi.elasticsearch; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import javax.annotation.PostConstruct; + +import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; +import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; +import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest; +import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse; +import org.elasticsearch.action.delete.DeleteResponse; +import org.elasticsearch.action.get.GetRequestBuilder; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.index.IndexResponse; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.update.UpdateRequest; +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.common.text.Text; +import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder; +import org.elasticsearch.search.sort.SortOrder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import com.alibaba.fastjson.JSONObject; + +@Component +public class ElasticsearchUtil { + + private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchUtil.class); + + @Autowired + private TransportClient transportClient; + private static TransportClient client; + + /** + * @PostContruct是spring框架的注解 spring容器初始化的时候执行该方法 + */ + @PostConstruct + public void init() { + client = this.transportClient; + } + + /** + * 创建索引 + * + * @param index + * @return + */ + public static boolean createIndex(String index) { + if (!isIndexExist(index)) { + LOGGER.info("Index is not exits!"); + } + CreateIndexResponse indexresponse = client.admin().indices().prepareCreate(index).execute().actionGet(); + LOGGER.info("执行建立成功？" + indexresponse.isAcknowledged()); + return indexresponse.isAcknowledged(); + } + + /** + * 删除索引 + * + * @param index + * @return + */ + public static boolean deleteIndex(String index) { + if (!isIndexExist(index)) { + LOGGER.info("Index is not exits!"); + } + DeleteIndexResponse dResponse = client.admin().indices().prepareDelete(index).execute().actionGet(); + if (dResponse.isAcknowledged()) { + LOGGER.info("delete index " + index + " successfully!"); + } else { + LOGGER.info("Fail to delete index " + index); + } + return dResponse.isAcknowledged(); + } + + /** + * 判断索引是否存在 + * + * @param index + * @return + */ + public static boolean isIndexExist(String index) { + IndicesExistsResponse inExistsResponse = client.admin().indices().exists(new IndicesExistsRequest(index)).actionGet(); + if (inExistsResponse.isExists()) { + LOGGER.info("Index [" + index + "] is exist!"); + } else { + LOGGER.info("Index [" + index + "] is not exist!"); + } + return inExistsResponse.isExists(); + } + + /** + * @Author: LX + * @Description: 判断inde下指定type是否存在 + * @Date: 2018/11/6 14:46 + * @Modified by: + */ + public boolean isTypeExist(String index, String type) { + return isIndexExist(index) + ? client.admin().indices().prepareTypesExists(index).setTypes(type).execute().actionGet().isExists() + : false; + } + + /** + * 数据添加，正定ID + * + * @param jsonObject 要增加的数据 + * @param index 索引，类似数据库 + * @param type 类型，类似表 + * @param id 数据ID + * @return + */ + public static String addData(JSONObject jsonObject, String index, String type, String id) { + IndexResponse response = client.prepareIndex(index, type, id).setSource(jsonObject).get(); + LOGGER.info("addData response status:{},id:{}", response.status().getStatus(), response.getId()); + return response.getId(); + } + + /** + * 数据添加 + * + * @param jsonObject 要增加的数据 + * @param index 索引，类似数据库 + * @param type 类型，类似表 + * @return + */ + public static String addData(JSONObject jsonObject, String index, String type) { + return addData(jsonObject, index, type, ElasticsearchUtil.getUUID32()); + } + + /** + * 通过ID删除数据 + * + * @param index 索引，类似数据库 + * @param type 类型，类似表 + * @param id 数据ID + */ + public static void deleteDataById(String index, String type, String id) { + + DeleteResponse response = client.prepareDelete(index, type, id).execute().actionGet(); + + LOGGER.info("deleteDataById response status:{},id:{}", response.status().getStatus(), response.getId()); + } + + /** + * 通过ID 更新数据 + * + * @param jsonObject 要增加的数据 + * @param index 索引，类似数据库 + * @param type 类型，类似表 + * @param id 数据ID + * @return + */ + public static void updateDataById(JSONObject jsonObject, String index, String type, String id) { + + UpdateRequest updateRequest = new UpdateRequest(); + + updateRequest.index(index).type(type).id(id).doc(jsonObject); + + client.update(updateRequest); + + } + + /** + * 通过ID获取数据 + * + * @param index 索引，类似数据库 + * @param type 类型，类似表 + * @param id 数据ID + * @param fields 需要显示的字段，逗号分隔（缺省为全部字段） + * @return + */ + public static Map searchDataById(String index, String type, String id, String fields) { + + GetRequestBuilder getRequestBuilder = client.prepareGet(index, type, id); + + if (!StringUtils.isEmpty(fields)) { + getRequestBuilder.setFetchSource(fields.split(","), null); + } + + GetResponse getResponse = getRequestBuilder.execute().actionGet(); + + return getResponse.getSource(); + } + + + + + /** + * 使用分词查询 + * + * @param index 索引名称 + * @param type 类型名称,可传入多个type逗号分隔 + * @param query 查询条件 + * @param size 文档大小限制 + * @param fields 需要显示的字段，逗号分隔（缺省为全部字段） + * @param sortField 排序字段 + * @param sortOrder + * @param highlightField 高亮字段 + * @return + */ + public static List> searchListData( + String index, String type, QueryBuilder query, Integer size, + String fields, String sortField, SortOrder sortOrder, String highlightField) { + + SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index); + if (!StringUtils.isEmpty(type)) { + searchRequestBuilder.setTypes(type.split(",")); + } + + if (!StringUtils.isEmpty(highlightField)) { + HighlightBuilder highlightBuilder = new HighlightBuilder(); + // 设置高亮字段 + highlightBuilder.field(highlightField); + searchRequestBuilder.highlighter(highlightBuilder); + } + + searchRequestBuilder.setQuery(query); + + if (!StringUtils.isEmpty(fields)) { + searchRequestBuilder.setFetchSource(fields.split(","), null); + } + searchRequestBuilder.setFetchSource(true); + + if (!StringUtils.isEmpty(sortField)) { + if(null==sortOrder){ + sortOrder = SortOrder.DESC; + } + searchRequestBuilder.addSort(sortField, sortOrder); + } + + if (size != null && size > 0) { + searchRequestBuilder.setSize(size); + } + + //打印的内容 可以在 Elasticsearch head 和 Kibana 上执行查询 + LOGGER.info("\n{}", searchRequestBuilder); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + + long totalHits = searchResponse.getHits().totalHits; + long length = searchResponse.getHits().getHits().length; + + LOGGER.info("共查询到[{}]条数据,处理数据条数[{}]", totalHits, length); + + if (searchResponse.status().getStatus() == 200) { + // 解析对象 + return setSearchResponse(searchResponse, highlightField); + } + return null; + + } + + + /** + * 高亮结果集 特殊处理 + * + * @param searchResponse + * @param highlightField + */ + private static List> setSearchResponse(SearchResponse searchResponse, String highlightField) { + List> sourceList = new ArrayList>(); + StringBuffer stringBuffer = new StringBuffer(); + for (SearchHit searchHit : searchResponse.getHits().getHits()) { + searchHit.getSourceAsMap().put("id", searchHit.getId()); + + if (!StringUtils.isEmpty(highlightField)) { + + System.out.println("遍历 高亮结果集，覆盖 正常结果集" + searchHit.getSourceAsMap()); + Text[] text = searchHit.getHighlightFields().get(highlightField).getFragments(); + + if (text != null) { + for (Text str : text) { + stringBuffer.append(str.string()); + } + //遍历 高亮结果集，覆盖 正常结果集 + searchHit.getSourceAsMap().put(highlightField, stringBuffer.toString()); + } + } + sourceList.add(searchHit.getSourceAsMap()); + } + return sourceList; + } + + private static String getUUID32(){ + String uuid = UUID.randomUUID().toString().replace("-", "").toLowerCase(); + return uuid; + } + +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java b/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java new file mode 100644 index 00000000..32909abc --- /dev/null +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java @@ -0,0 +1,137 @@ +package com.gzzn.omms.msgexchangeapi.scheduled; + +import java.math.BigInteger; +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.springframework.beans.factory.annotation.Autowired; +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.MOVEMENTINDICATOR; +import com.gzzn.omms.msgexchangeapi.entity.msg.SCHD.FLTR; +import com.gzzn.omms.msgexchangeapi.service.flightInfo.IFlightInfoService; +import com.gzzn.omms.msgexchangeapi.utils.DateTimeUtil; +import com.gzzn.omms.msgexchangeapi.utils.JsonUtil; + +/** + * 历史航班数据定时器 + * @author fhc + * + */ +@Component +@Async +public class FlightHisScheduled { + + @Autowired + private IFlightInfoService flightInfoService; + + /** + * 航班到达超过多久则成为历史航班数据（单位：秒） + */ + @Value("${hstCondition.ARRIVE_HST_TIME}") + private Integer ARRIVE_HST_TIME ; + /** + * 历史航班数据索引 + */ + private final String INDEX_NAME = "flight_hts"; + /** + * 历史航班数据类型 + */ + private final String ES_TYPE = "_doc"; + + + /** + * 每天凌晨 将前一天的 动态消息中 已为历史的 放入ES中 + */ + @Scheduled(cron = "${scheduled.flightInfoCron}") + public void scheduled(){ + System.out.println("开始采集历史数据..."); + //1.从redis中获取 当前的动态消息 + List fltrs = flightInfoService.findAll(); + List fltrIdForHistory = new ArrayList(); + List fltrsHistory = new ArrayList(); + + Long todayMixTime = DateTimeUtil.getTodayStartTime(); + //遍历 查找出已经 可以作为历史数据的动态消息 + //条件为 ：完成运营的航班是指已经落地超过2小时或已经起飞的 + if(null!=fltrs && fltrs.size()>0){ + for(FLTR fltr : fltrs){ + String ACTT = fltr.getACTT(); //航班实际时间 ddMMMyyHHmm + if(StringUtils.isEmpty(ACTT)){ + continue; + } + Long ACTT_longTime = DateTimeUtil.toDate(ACTT, "ddMMMyyHHmm",Locale.ENGLISH).getTime(); + if(ACTT_longTime>=todayMixTime){ + //计划时间为凌晨0点(包含)后的 不计入历史数据 + continue; + } + + //离港航班 + if(fltr.getMVIN().equals(MOVEMENTINDICATOR.D)){ + //实际时间小于当前时间则为已离港 + if(ACTT_longTime<=System.currentTimeMillis()){ + //转为历史数据 + fltrsHistory.add(fltr); + } + } + + //到达航班 + if(fltr.getMVIN().equals(MOVEMENTINDICATOR.A)){ + //到达实际时间超过规定时间的 则视为历史数据 + if((System.currentTimeMillis()-ACTT_longTime)>ARRIVE_HST_TIME){ + //转为历史数据 + fltrsHistory.add(fltr); + } + } + } + } + + + //往 elasticsearch中放入历史数据 通过FLID + ACTT 判断唯一 + 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 ACTT = fltr.getACTT(); + Long FLID = fltr.getFLID().longValue(); + BoolQueryBuilder boolQuery = QueryBuilders.boolQuery() + .must(QueryBuilders.termQuery("ACTT",ACTT)) + .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()); + }else{ + //新增历史数据 + ElasticsearchUtil.addData(jsondata, INDEX_NAME, ES_TYPE); + fltrIdForHistory.add(fltr.getFLID()); + } + } + } + + //删除redis中已成为历史数据的动态消息 + if(null!=fltrIdForHistory && fltrIdForHistory.size()>0){ + flightInfoService.batchDeleteFltr(fltrIdForHistory); + } + + System.out.println("结束采集历史数据..."); + } + +} diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightInfoServiceImpl.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightInfoServiceImpl.java index 663a2377..86208b8b 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightInfoServiceImpl.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/FlightInfoServiceImpl.java @@ -1,7 +1,9 @@ package com.gzzn.omms.msgexchangeapi.service.flightInfo; +import java.math.BigInteger; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -84,4 +86,16 @@ public class FlightInfoServiceImpl implements IFlightInfoService { redisService.hdel(redisKey, fltr.getFLID()); return true; } + + @Override + public boolean batchDeleteFltr(List flids) { + if(null!=flids){ + flids.removeAll(Collections.singleton(null)); + } + if(null==flids || flids.size()<=0){ + return true; + } + redisService.hdel(redisKey, flids.toArray()); + return true; + } } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/IFlightInfoService.java b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/IFlightInfoService.java index af53065b..3b044383 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/IFlightInfoService.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/service/flightInfo/IFlightInfoService.java @@ -1,5 +1,6 @@ package com.gzzn.omms.msgexchangeapi.service.flightInfo; +import java.math.BigInteger; import java.util.List; import com.gzzn.omms.msgexchangeapi.entity.Cminmsg; @@ -38,4 +39,11 @@ public interface IFlightInfoService { * @return */ public boolean deleteFltr(SCHD.FLTR fltr); + + /** + * 批量物理删除动态航班信息 + * @param flids + * @return + */ + public boolean batchDeleteFltr(List flids); } diff --git a/src/main/java/com/gzzn/omms/msgexchangeapi/utils/DateTimeUtil.java b/src/main/java/com/gzzn/omms/msgexchangeapi/utils/DateTimeUtil.java index 86184963..877b8b99 100644 --- a/src/main/java/com/gzzn/omms/msgexchangeapi/utils/DateTimeUtil.java +++ b/src/main/java/com/gzzn/omms/msgexchangeapi/utils/DateTimeUtil.java @@ -38,4 +38,33 @@ public class DateTimeUtil { throw new RuntimeException("日期转换失败:"+e.getMessage()); } } + + /** + * 字符串转日期 + * @param strDate 字符串的日期 + * @param pattern 日期格式 + * @param locale + * @return + * @throws ParseException + */ + public static Date toDate(String strDate,String pattern,Locale locale) + { + try { + SimpleDateFormat sdf = new SimpleDateFormat(pattern,locale); + return sdf.parse(strDate); + } + catch (ParseException e) + { + throw new RuntimeException("日期转换失败:"+e.getMessage()); + } + } + + /** + * 获取今天最小时间 单位秒 + * @return + */ + public static Long getTodayStartTime(){ + String time = DateTimeUtil.gtFormatStr(new Date(), "yyyy-MM-dd",Locale.ENGLISH)+" 00:00:00"; + return DateTimeUtil.toDate(time,"yyyy-MM-dd HH:mm:ss").getTime(); + } } diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index 81b34ad0..f57457bf 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -53,4 +53,24 @@ msgExchange: intervalSeconds: 5 logstash: host: 130.120.3.233:5000 - \ No newline at end of file +scheduled: + #每日凌晨2点采集历史数据 + #flightInfoCron: 0 14 18 * * ? + flightInfoCron: 0 0 2 * * ? + corePoolSize: 10 + maxPoolSize: 50 + queueCapacity: 10 +#计入历史航班数据的条件 +hstCondition: + #航班到达超过多久则成为历史航班数据（单位：秒） + ARRIVE_HST_TIME: 7200 + +# Elasticsearch +# 9200端口是用来让HTTP REST API来访问ElasticSearch，而9300端口是传输层监听的默认端口 +elasticsearch: + ip: 130.120.3.233 + port: 9300 + pool: 5 + #注意cluster.name需要与config/elasticsearch.yml中的cluster.name一致 + cluster: + name: docker-cluster \ No newline at end of file