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

重构完善消息收取机制。修改到MsgExchangeRunner中收取。

Browse Source
This commit is contained in:
zhouxiunai
2018-12-06 16:38:47 +08:00
parent 47e473d9b1
commit 367f610db4
5 changed files with 170 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/gzzn/omms/msgexchangeapi/msghandler/MsgHandlerDispatcher.java
+1 -1
View File
@@ -12,7 +12,7 @@ import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
import com.gzzn.omms.msgexchangeapi.entity.msg.Msg; import com.gzzn.omms.msgexchangeapi.entity.msg.Msg;
import com.gzzn.omms.msgexchangeapi.service.IExchangeService; import com.gzzn.omms.msgexchangeapi.service.IExchangeService;
@Service @Service("msgHandlerDispatcher")
public class MsgHandlerDispatcher { public class MsgHandlerDispatcher {
private Logger logger = LoggerFactory.getLogger(MsgHandlerDispatcher.class); private Logger logger = LoggerFactory.getLogger(MsgHandlerDispatcher.class);
src/main/java/com/gzzn/omms/msgexchangeapi/runner/AppRunner.java
+6
View File
@@ -38,5 +38,11 @@ public class AppRunner implements CommandLineRunner {
redisService.set(RedisKeyConstant.KEY_WAITSCHDDATE,nowDate); //初始化消息进度 redisService.set(RedisKeyConstant.KEY_WAITSCHDDATE,nowDate); //初始化消息进度
logger.info("发送获取动态航班日计划的消息完成"); logger.info("发送获取动态航班日计划的消息完成");
//启动线程等待日航班计划消息返回
Thread t = new Thread(new SchdWaitRunner());
t.run();
}//end function run }//end function run
} }
src/main/java/com/gzzn/omms/msgexchangeapi/runner/MsgExchangeRunner.java
+64
View File
@@ -0,0 +1,64 @@
package com.gzzn.omms.msgexchangeapi.runner;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
import com.gzzn.omms.msgexchangeapi.msghandler.MsgHandlerDispatcher;
import com.gzzn.omms.msgexchangeapi.redis.RedisKeyConstant;
import com.gzzn.omms.msgexchangeapi.redis.RedisService;
import com.gzzn.omms.msgexchangeapi.service.ICminmsgService;
import com.gzzn.omms.msgexchangeapi.utils.SpringUtil;
/**
* 消息获取分派类
* @author zhouxiunai
*
*/
public class MsgExchangeRunner implements Runnable{
private static Logger logger = LoggerFactory.getLogger(MsgExchangeRunner.class);
private MsgHandlerDispatcher msgHandlerDispatcher;
private RedisService redisService;
private ICminmsgService cminmsgService;
public MsgExchangeRunner() {
redisService = (RedisService) SpringUtil.getBean("redisService");
cminmsgService = (ICminmsgService) SpringUtil.getBean("cminmsgService");
msgHandlerDispatcher = (MsgHandlerDispatcher) SpringUtil.getBean("msgHandlerDispatcher");
}
@Override
public void run() {
logger.info("进入获取动态航班消息...");
//获取上次的beginId
Integer beginIdInt = (Integer)redisService.get(RedisKeyConstant.KEY_LASTBEGINID);
Long beginId = beginIdInt.longValue();
logger.info("从上次获取的消息位置{}开始",beginId);
//获取航班动态消息
List<Cminmsg> lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId);
if(null == lsCminmsgs || lsCminmsgs.size() <= 0)
{
logger.info("本周期没有获取到最新消息，直接退出本次循环");
return;
}
logger.info("获取到{}条最新信息,准备更新动态航班信息",lsCminmsgs.size());
//分发消息进行处理
for (Cminmsg cminmsg : lsCminmsgs) {
msgHandlerDispatcher.dispatch(cminmsg);
}//end for
logger.info("本次定时获取新消息完成...");
}//end run
}
src/main/java/com/gzzn/omms/msgexchangeapi/runner/SchdWaitRunner.java
+99
View File
@@ -0,0 +1,99 @@
package com.gzzn.omms.msgexchangeapi.runner;
import java.util.Date;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
import com.gzzn.omms.msgexchangeapi.redis.RedisKeyConstant;
import com.gzzn.omms.msgexchangeapi.redis.RedisService;
import com.gzzn.omms.msgexchangeapi.service.ICminmsgService;
import com.gzzn.omms.msgexchangeapi.service.flightInfo.IFlightInfoService;
import com.gzzn.omms.msgexchangeapi.utils.SpringUtil;
/**
* 等待航班日计划返回线程
* @author zhouxiunai
*
*/
public class SchdWaitRunner implements Runnable {
private static Logger logger = LoggerFactory.getLogger(SchdWaitRunner.class);
private RedisService redisService;
private ICminmsgService cminmsgService;
private IFlightInfoService flightInfoService;
public SchdWaitRunner()
{
flightInfoService = (IFlightInfoService) SpringUtil.getBean("flightInfoService");
redisService = (RedisService) SpringUtil.getBean("redisService");
cminmsgService = (ICminmsgService) SpringUtil.getBean("cminmsgService");
}
@Override
public void run() {
logger.info("获取日计划航班信息返回结果开始...");
Boolean isRun = true;
Long intervalMs = 3000L;
Long tryTimes = 0L;
while (isRun) {
tryTimes++;
logger.info("尝试第{}次获取返回日计划消息",tryTimes);
//获取
Date waitSchdDate = (Date) redisService.get(RedisKeyConstant.KEY_WAITSCHDDATE);
Optional<Cminmsg> opCminmsg = cminmsgService.getRespSchdCminmsg(waitSchdDate,10,3000L);
if(!opCminmsg.isPresent())
{
sleepInMs(intervalMs);
continue ;
}
//更新
logger.info("找到返回的动态航班计划,开始更新动态航班信息...");
Boolean updateResult = flightInfoService.updateByDaySchd(opCminmsg.get());
if(false == updateResult)
{
logger.warn("更新动态航班信息失败,稍后重试...");
sleepInMs(intervalMs);
continue ;
}
//状态记录
Long beginId = opCminmsg.get().getCminmsgsId();
redisService.set(RedisKeyConstant.KEY_LASTBEGINID,beginId);
Boolean isWaitSchd = false;
redisService.set(RedisKeyConstant.KEY_ISWAITSCHD,isWaitSchd);
//启动消息采集线程
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
scheduledExecutorService.scheduleAtFixedRate(new MsgExchangeRunner(), 0, 30 , TimeUnit.SECONDS);
isRun = false; //退出
}
logger.info("获取日计划航班信息返回结果结束...");
} //end run
/**
* 休眠指定时间间隔-毫秒
* @param intervalMs
*/
private void sleepInMs(Long intervalMs)
{
try {
Thread.sleep(intervalMs);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
src/main/java/com/gzzn/omms/msgexchangeapi/task/ExchangeTask.java
-108
View File
@@ -1,108 +0,0 @@
package com.gzzn.omms.msgexchangeapi.task;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import com.gzzn.omms.msgexchangeapi.entity.Cminmsg;
import com.gzzn.omms.msgexchangeapi.msghandler.MsgHandlerDispatcher;
import com.gzzn.omms.msgexchangeapi.redis.RedisKeyConstant;
import com.gzzn.omms.msgexchangeapi.redis.RedisService;
import com.gzzn.omms.msgexchangeapi.service.ICminmsgService;
import com.gzzn.omms.msgexchangeapi.service.flightInfo.IFlightInfoService;
/**
* 转换任务，获取cim的消息，转换成json，写入到kafka队列中
* @author Administrator
*
*/
@Component
public class ExchangeTask {
private static Logger logger = LoggerFactory.getLogger(ExchangeTask.class);
@Autowired
MsgHandlerDispatcher msgHandlerDispatcher;
@Autowired
RedisService redisService;
@Autowired
private ICminmsgService cminmsgService;
@Autowired
private IFlightInfoService flightInfoService;
@Scheduled(cron="0 0/1 * * * ?")
public void corn()
{
logger.info("进入获取动态航班消息...");
Boolean isWaitSchd = (Boolean)redisService.get(RedisKeyConstant.KEY_ISWAITSCHD);
if (isWaitSchd) {
Date waitSchdDate = (Date)redisService.get(RedisKeyConstant.KEY_WAITSCHDDATE);
Optional<Cminmsg> opCminmsg = cminmsgService.getRespSchdCminmsg(waitSchdDate,10,3000L);
if(opCminmsg.isPresent()) {
//如果找到，添加到内存数据库动态航班信息表
logger.info("找到返回的动态航班计划,开始更新动态航班信息...");
Boolean updateResult = flightInfoService.updateByDaySchd(opCminmsg.get());
if(false == updateResult)
{
logger.error("更新动态航班信息失败");
return ;
}
//
Long beginId = opCminmsg.get().getCminmsgsId();
redisService.set(RedisKeyConstant.KEY_LASTBEGINID,beginId);
//更新状态
isWaitSchd = false;
redisService.set(RedisKeyConstant.KEY_ISWAITSCHD,isWaitSchd);
logger.info("动态航班计划更新完成...");
}
else {
logger.error("获取动态航班消息失败,退出本次周期...");
return ;
}
} // end if 获取航班日记录
//获取上次的beginId
Integer beginIdInt = (Integer)redisService.get(RedisKeyConstant.KEY_LASTBEGINID);
Long beginId = beginIdInt.longValue();
logger.info("从上次获取的消息位置{}开始",beginId);
//获取航班动态消息
List<Cminmsg> lsCminmsgs = cminmsgService.getNewMsgsAfterId(beginId);
if(null == lsCminmsgs || lsCminmsgs.size() <= 0)
{
logger.info("本周期没有获取到最新消息，直接退出本次循环");
return;
}
logger.info("获取到{}条最新信息,准备更新动态航班信息",lsCminmsgs.size());
//分发消息进行处理
for (Cminmsg cminmsg : lsCminmsgs) {
msgHandlerDispatcher.dispatch(cminmsg);
}//end for
logger.info("本次定时获取新消息完成...");
} //end function
}