完成新消息采集，更新动态航班信息，发送动态航班信息，发送动态航班消息。功能

src/main/java/com/gzzn/omms/msgexchangeapi/redis/RedisKeyConstant.java

@@ -7,5 +7,6 @@ package com.gzzn.omms.msgexchangeapi.redis;
  */
 public class RedisKeyConstant {
   public static String  KEY_ISWAITSCHD="isWaitSchd";
-  public static String  KEY_MSGPROGRESS="msgProgress";
+  public static String  KEY_WAITSCHDDATE="waitSchdDate";
+  public static String KEY_LASTBEGINID="lastBeginId";
 }

src/main/java/com/gzzn/omms/msgexchangeapi/runner/AppRunner.java

@@ -55,6 +55,6 @@ public class AppRunner implements CommandLineRunner  {
 		
 		//
 		redisService.set(RedisKeyConstant.KEY_ISWAITSCHD, true);  //等待返回日计划状态
-		redisService.set(RedisKeyConstant.KEY_MSGPROGRESS,nowDate); //初始化消息进度
+		redisService.set(RedisKeyConstant.KEY_WAITSCHDDATE,nowDate); //初始化消息进度
 	}//end function run
 }

src/main/java/com/gzzn/omms/msgexchangeapi/service/CminmsgServiceImpl.java

@@ -4,6 +4,7 @@ import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -70,4 +71,21 @@ public class CminmsgServiceImpl implements ICminmsgService {
 	    		                       .findFirst();
 	    return opCminmsg;
 	}//end function
+
+
+	@Override
+	public List<Cminmsg>  updateBatchProcessed(List<Cminmsg> lsCminmsgs) {
+		//
+		List<Long> ids = lsCminmsgs.stream().map(node->{
+			return node.getCminmsgsId();
+		}).collect(Collectors.toList());
+		
+		List<Cminmsg>  lsCmins = (List<Cminmsg>) cminmsgDao.findAll(ids);
+		List<Cminmsg>  lsCminsProcessed = lsCmins.stream().map(node->{
+			node.setCminmsgsDateReceived(new Date());
+			return node;
+		}).collect(Collectors.toList());
+		
+		return (List<Cminmsg>) cminmsgDao.save(lsCminsProcessed);
+	}//end function
 }

src/main/java/com/gzzn/omms/msgexchangeapi/service/ICminmsgService.java

@@ -9,6 +9,13 @@ import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
 
 public interface ICminmsgService {
 	
+  /**
+   * 批量更新处理时间	
+   * @return
+   */
+  public List<Cminmsg>  updateBatchProcessed(List<Cminmsg> lsCminmsgs);
+	
+	
  /**
  * 发送xml格式的消息到cminmsgs表
  * @param xmlMsg

src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java

@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchangeapi.task;
 import java.util.Date;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -10,7 +11,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
-import com.gzzn.omms.msgexchangeapi.entity.msg.schd.dnld.FLTR;
 import com.gzzn.omms.msgexchangeapi.redis.RedisKeyConstant;
 import com.gzzn.omms.msgexchangeapi.redis.RedisService;
 import com.gzzn.omms.msgexchangeapi.service.ICminmsgService;
@@ -51,7 +51,7 @@ public class ExchangeTask {
 		logger.info("定时任务启动....");
 		
 		Boolean isWaitSchd =  (Boolean)redisService.get(RedisKeyConstant.KEY_ISWAITSCHD);
-		Date msgProgress = (Date)redisService.get(RedisKeyConstant.KEY_MSGPROGRESS);
+		
 		
 		Long beginId = null;
 		Integer tryTimes = 0;
@@ -59,7 +59,8 @@ public class ExchangeTask {
 			tryTimes++;
 			
 			//查找回复的日计划消息
-			Optional<Cminmsg> opCminmsg = cminmsgService.getRespSchdCminmsg(msgProgress);
+			Date waitSchdDate = (Date)redisService.get(RedisKeyConstant.KEY_WAITSCHDDATE);
+			Optional<Cminmsg> opCminmsg = cminmsgService.getRespSchdCminmsg(waitSchdDate);
 		    if(opCminmsg.isPresent())
 		    {
 		    	//如果找到，添加到内存数据库动态航班信息表
@@ -84,29 +85,66 @@ public class ExchangeTask {
 			}
 		} // end while  获取航班日记录
 		
-		if(beginId == null)
+		if(isWaitSchd == true)
 		{
 			//获取返回的航班计划回复失败,发送报警信息退出整个程序
+			logger.error("获取动态航班计划失败...");
 			return ;
 		}
 		
+		if(null == beginId) 
+		{
+			//获取上次的beginId
+			beginId = (Long)redisService.get(RedisKeyConstant.KEY_LASTBEGINID);
+		}
+		
 		//获取航班动态消息
 		List<Cminmsg> lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId);
 		List<UpdateByCminmsgsResult> lsUpdatedFlightInfo  = flightInfoService.updateByCminmsgs(lsCminmsgs);
 		
 		//新增或变更 航班动态信息  发送 到kafka的SCHD topic
 		for (UpdateByCminmsgsResult updateResult : lsUpdatedFlightInfo) {
-			
-		}
+			if(updateResult.getIsSuccess())
+			{
+				updateResult.getFltrs().forEach(node->{
+					String jsonMsg = exchangeService.objectToXmlstr(node);
+					kafkaservice.msgSend("schd", jsonMsg);
+				});
+			}
+		}//end for
 		
 		//发送动态消息到kafka
-		for (Cminmsg cminmsg : lsCminmsgs) {
-			
-		}
+		for (UpdateByCminmsgsResult updateResult : lsUpdatedFlightInfo) {
+			if(updateResult.getIsSuccess())
+			{
+				String jsonMsg = exchangeService.xmlToJson(
+						                         updateResult
+						                         .getCminmsg()
+						                         .getCminmsgsClobMsg()
+						                         );
+				
+				kafkaservice.msgSend("msg", jsonMsg);
+			}//end if
+		}//end for
 		
 		
 		//更新消息处理状态
+		List<Cminmsg> lsCminmsgsSuccess  =  lsUpdatedFlightInfo.stream().filter(node->{
+			return node.getIsSuccess();
+		}).map(node->{
+			return node.getCminmsg();
+		}).collect(Collectors.toList());
+		
+		
+		cminmsgService.updateBatchProcessed(lsCminmsgsSuccess);
+		
+		Long newBeginId = lsUpdatedFlightInfo
+				          .get(lsUpdatedFlightInfo.size()-1)
+				          .getCminmsg()
+				          .getCminmsgsId();
+		
+		redisService.set(RedisKeyConstant.KEY_LASTBEGINID,newBeginId);
 		
 		logger.info("同步完成");
-	}
+	} //end function
 }