Skip to content

X
XXL-JOB
提交 18162e75 authored 作者: xuxueli's avatar xuxueli
浏览文件
操作

update document

上级 2f640f13
隐藏空白字符变更
内嵌 并排
正在显示 包含 59 行增加 和 33 行删除
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobScheduleHelper.java
浏览文件 @ 18162e75
...@@ -25,6 +25,8 @@ public class JobScheduleHelper { ...@@ -25,6 +25,8 @@ public class JobScheduleHelper {
return instance; return instance;
} }
public static final long PRE_READ_MS = 5000; // pre read
private Thread scheduleThread; private Thread scheduleThread;
private Thread ringThread; private Thread ringThread;
private volatile boolean scheduleThreadToStop = false; private volatile boolean scheduleThreadToStop = false;
...@@ -47,11 +49,11 @@ public class JobScheduleHelper { ...@@ -47,11 +49,11 @@ public class JobScheduleHelper {
} }
logger.info(">>>>>>>>> init xxl-job admin scheduler success."); logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
Connection conn = null;
while (!scheduleThreadToStop) { while (!scheduleThreadToStop) {
// 扫描任务 // 扫描任务
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
Connection conn = null;
PreparedStatement preparedStatement = null; PreparedStatement preparedStatement = null;
try { try {
if (conn==null || conn.isClosed()) { if (conn==null || conn.isClosed()) {
...@@ -65,16 +67,16 @@ public class JobScheduleHelper { ...@@ -65,16 +67,16 @@ public class JobScheduleHelper {
// tx start // tx start
// 1、预读5s内调度任务 // 1、预读5s内调度任务
long maxNextTime = System.currentTimeMillis() + 5000; List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(System.currentTimeMillis() + PRE_READ_MS);
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(maxNextTime);
if (scheduleList!=null && scheduleList.size()>0) { if (scheduleList!=null && scheduleList.size()>0) {
// 2、推送时间轮 // 2、推送时间轮
for (XxlJobInfo jobInfo: scheduleList) { for (XxlJobInfo jobInfo: scheduleList) {
// 时间轮刻度计算 // 时间轮刻度计算
if (System.currentTimeMillis() > jobInfo.getTriggerNextTime() + 5000) { if (System.currentTimeMillis() > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
// 过期超5s：本地忽略，当前时间开始计算下次触发时间 // 过期超5s：本地忽略，当前时间开始计算下次触发时间
// fresh next
jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime()); jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
jobInfo.setTriggerNextTime( jobInfo.setTriggerNextTime(
new CronExpression(jobInfo.getJobCron()) new CronExpression(jobInfo.getJobCron())
...@@ -82,44 +84,55 @@ public class JobScheduleHelper { ...@@ -82,44 +84,55 @@ public class JobScheduleHelper {
.getTime() .getTime()
); );
// pass
continue;
} else if (System.currentTimeMillis() > jobInfo.getTriggerNextTime()) { } else if (System.currentTimeMillis() > jobInfo.getTriggerNextTime()) {
// 过期5s内 ：立即触发一次，当前时间开始计算下次触发时间 // 过期5s内 ：立即触发一次，当前时间开始计算下次触发时间；一旦过期，预读一次；
jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime()); CronExpression cronExpression = new CronExpression(jobInfo.getJobCron());
jobInfo.setTriggerNextTime( long nextTime = cronExpression.getNextValidTimeAfter(new Date()).getTime();
new CronExpression(jobInfo.getJobCron())
.getNextValidTimeAfter(new Date())
.getTime()
);
// do trigger // 1、trigger
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null); JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
// 2、fresh next
jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
jobInfo.setTriggerNextTime(nextTime);
// 3、check pre read
if (jobInfo.getTriggerNextTime() - System.currentTimeMillis() < PRE_READ_MS) {
// 1、make ring second
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
// 2、push time ring
pushTimeRing(ringSecond, jobInfo.getId());
// 3、fresh next
jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
jobInfo.setTriggerNextTime(
new CronExpression(jobInfo.getJobCron())
.getNextValidTimeAfter(new Date(jobInfo.getTriggerNextTime()))
.getTime()
);
}
logger.debug(">>>>>>>>>>> xxl-job, push trigger : jobId = " + jobInfo.getId() ); logger.debug(">>>>>>>>>>> xxl-job, push trigger : jobId = " + jobInfo.getId() );
} else { } else {
// 未过期：正常触发，递增计算下次触发时间 // 未过期：正常触发，递增计算下次触发时间
// 1、make ring second
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60); int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
// 2、push time ring
pushTimeRing(ringSecond, jobInfo.getId());
// 3、fresh next
jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime()); jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
jobInfo.setTriggerNextTime( jobInfo.setTriggerNextTime(
new CronExpression(jobInfo.getJobCron()) new CronExpression(jobInfo.getJobCron())
.getNextValidTimeAfter(new Date(jobInfo.getTriggerNextTime())) .getNextValidTimeAfter(new Date(jobInfo.getTriggerNextTime()))
.getTime() .getTime()
); );
// push async ring
List<Integer> ringItemData = ringData.get(ringSecond);
if (ringItemData == null) {
ringItemData = new ArrayList<Integer>();
ringData.put(ringSecond, ringItemData);
}
ringItemData.add(jobInfo.getId());
logger.debug(">>>>>>>>>>> xxl-job, push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) );
} }
} }
...@@ -139,12 +152,6 @@ public class JobScheduleHelper { ...@@ -139,12 +152,6 @@ public class JobScheduleHelper {
logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e); logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
} }
} finally { } finally {
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
}
}
if (null != preparedStatement) { if (null != preparedStatement) {
try { try {
preparedStatement.close(); preparedStatement.close();
...@@ -166,6 +173,12 @@ public class JobScheduleHelper { ...@@ -166,6 +173,12 @@ public class JobScheduleHelper {
} }
} }
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
}
}
logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop"); logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
} }
}); });
...@@ -235,6 +248,18 @@ public class JobScheduleHelper { ...@@ -235,6 +248,18 @@ public class JobScheduleHelper {
ringThread.start(); ringThread.start();
} }
private void pushTimeRing(int ringSecond, int jobId){
// push async ring
List<Integer> ringItemData = ringData.get(ringSecond);
if (ringItemData == null) {
ringItemData = new ArrayList<Integer>();
ringData.put(ringSecond, ringItemData);
}
ringItemData.add(jobId);
logger.debug(">>>>>>>>>>> xxl-job, push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) );
}
public void toStop(){ public void toStop(){
// 1、stop schedule // 1、stop schedule
......
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/XxlJobServiceImpl.java
浏览文件 @ 18162e75
...@@ -4,6 +4,7 @@ import com.xxl.job.admin.core.model.XxlJobGroup; ...@@ -4,6 +4,7 @@ import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobInfo; import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.cron.CronExpression; import com.xxl.job.admin.core.cron.CronExpression;
import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum; import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.core.thread.JobScheduleHelper;
import com.xxl.job.admin.core.util.I18nUtil; import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.admin.dao.XxlJobGroupDao; import com.xxl.job.admin.dao.XxlJobGroupDao;
import com.xxl.job.admin.dao.XxlJobInfoDao; import com.xxl.job.admin.dao.XxlJobInfoDao;
...@@ -193,7 +194,7 @@ public class XxlJobServiceImpl implements XxlJobService { ...@@ -193,7 +194,7 @@ public class XxlJobServiceImpl implements XxlJobService {
long nextTriggerTime = exists_jobInfo.getTriggerNextTime(); long nextTriggerTime = exists_jobInfo.getTriggerNextTime();
if (exists_jobInfo.getTriggerStatus() == 1 && !jobInfo.getJobCron().equals(exists_jobInfo.getJobCron()) ) { if (exists_jobInfo.getTriggerStatus() == 1 && !jobInfo.getJobCron().equals(exists_jobInfo.getJobCron()) ) {
try { try {
nextTriggerTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(new Date(System.currentTimeMillis() + 5000)).getTime(); nextTriggerTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(new Date(System.currentTimeMillis() + JobScheduleHelper.PRE_READ_MS)).getTime();
} catch (ParseException e) { } catch (ParseException e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_unvalid")+" | "+ e.getMessage()); return new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_unvalid")+" | "+ e.getMessage());
...@@ -239,7 +240,7 @@ public class XxlJobServiceImpl implements XxlJobService { ...@@ -239,7 +240,7 @@ public class XxlJobServiceImpl implements XxlJobService {
// next trigger time (5s后生效，避开预读周期) // next trigger time (5s后生效，避开预读周期)
long nextTriggerTime = 0; long nextTriggerTime = 0;
try { try {
nextTriggerTime = new CronExpression(xxlJobInfo.getJobCron()).getNextValidTimeAfter(new Date(System.currentTimeMillis() + 5000)).getTime(); nextTriggerTime = new CronExpression(xxlJobInfo.getJobCron()).getNextValidTimeAfter(new Date(System.currentTimeMillis() + JobScheduleHelper.PRE_READ_MS)).getTime();
} catch (ParseException e) { } catch (ParseException e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_unvalid")+" | "+ e.getMessage()); return new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobinfo_field_cron_unvalid")+" | "+ e.getMessage());
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论