msgexchange-api/src/main/java/com/gzzn/omms/msgexchangeapi/scheduled/FlightHisScheduled.java

package com.gzzn.omms.msgexchangeapi.scheduled;

import java.util.ArrayList;
import java.util.List;
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;
import org.slf4j.LoggerFactory;
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 com.alibaba.fastjson.JSONObject;
import com.gzzn.omms.msgexchangeapi.elasticsearch.ElasticsearchUtil;
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.JsonUtil;

/**
 * 历史航班数据定时器
 * @author fhc
 *
 */
@Component
@Async
public class FlightHisScheduled {
	
	@Autowired
	private IFlightInfoService flightInfoService;
	
	
	@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);
	/**
	 * 每天凌晨  将前一天的 动态消息中 已为历史的 放入ES中
	 */
	@Scheduled(cron = "${scheduled.flightInfoCron}")
    public void scheduled(){
		logger.info("开始采集历史数据...");
		
        
		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 ; //没有要转的航班历史
		}
		
		//往 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(null!=listInEs && listInEs.size()>0){
				//获取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(null!=fltrIdForHistory && fltrIdForHistory.size()>0){
			flightInfoService.batchDeleteFltr(fltrIdForHistory);
			logger.info("本次删除转历史动态航班数据:{}条",fltrIdForHistory.size());
		}
		
		logger.info("结束采集历史数据。");
	}
}