|
|
|
|
@ -6,6 +6,7 @@ import com.xxl.job.admin.core.model.XxlJobInfo;
|
|
|
|
|
import com.xxl.job.admin.core.scheduler.MisfireStrategyEnum;
|
|
|
|
|
import com.xxl.job.admin.core.scheduler.ScheduleTypeEnum;
|
|
|
|
|
import com.xxl.job.admin.core.trigger.TriggerTypeEnum;
|
|
|
|
|
import com.xxl.job.core.util.IpUtil;
|
|
|
|
|
import org.slf4j.Logger;
|
|
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
|
|
|
|
|
@ -15,6 +16,7 @@ import java.sql.SQLException;
|
|
|
|
|
import java.util.*;
|
|
|
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* @author xuxueli 2019-05-21
|
|
|
|
|
@ -70,6 +72,7 @@ public class JobScheduleHelper {
|
|
|
|
|
connAutoCommit = conn.getAutoCommit();
|
|
|
|
|
conn.setAutoCommit(false);
|
|
|
|
|
|
|
|
|
|
//TODO 通过这个来控制多节点只有一个节点会执行 low的一笔
|
|
|
|
|
preparedStatement = conn.prepareStatement( "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
|
|
|
|
|
preparedStatement.execute();
|
|
|
|
|
|
|
|
|
|
@ -77,18 +80,37 @@ public class JobScheduleHelper {
|
|
|
|
|
|
|
|
|
|
// 1、pre read
|
|
|
|
|
long nowTime = System.currentTimeMillis();
|
|
|
|
|
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
|
|
|
|
|
//TODO 下次执行时间 在5秒内的 整个框架的核心
|
|
|
|
|
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao()
|
|
|
|
|
.scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount, XxlJobAdminConfig.getAdminConfig().getHostName(),
|
|
|
|
|
XxlJobAdminConfig.getAdminConfig().isClusterEnable());
|
|
|
|
|
if (scheduleList!=null && scheduleList.size()>0) {
|
|
|
|
|
|
|
|
|
|
if(XxlJobAdminConfig.getAdminConfig().isClusterEnable()) {
|
|
|
|
|
XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().updateStatusById(scheduleList.stream()
|
|
|
|
|
.map(XxlJobInfo::getId).collect(Collectors.toList()));
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
conn.commit();
|
|
|
|
|
} catch (SQLException e) {
|
|
|
|
|
if (!scheduleThreadToStop) {
|
|
|
|
|
logger.error(e.getMessage(), e);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
scheduleList.forEach(x->logger.info(x.getId()+""));
|
|
|
|
|
// 2、push time-ring
|
|
|
|
|
for (XxlJobInfo jobInfo: scheduleList) {
|
|
|
|
|
|
|
|
|
|
// time-ring jump
|
|
|
|
|
// TODO 当前时间>下次执行时间+5秒 任务触发过期策略 5秒内不算过期
|
|
|
|
|
if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
|
|
|
|
|
// 2.1、trigger-expire > 5s：pass && make next-trigger-time
|
|
|
|
|
logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
|
|
|
|
|
|
|
|
|
|
// 1、misfire match
|
|
|
|
|
MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
|
|
|
|
|
//如果过期策略是立即执行一次则 执行
|
|
|
|
|
if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
|
|
|
|
|
// FIRE_ONCE_NOW 》 trigger
|
|
|
|
|
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
|
|
|
|
|
@ -96,25 +118,33 @@ public class JobScheduleHelper {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 2、fresh next
|
|
|
|
|
//刷新下次执行时间
|
|
|
|
|
refreshNextValidTime(jobInfo, new Date());
|
|
|
|
|
|
|
|
|
|
} else if (nowTime > jobInfo.getTriggerNextTime()) {
|
|
|
|
|
}
|
|
|
|
|
//TODO 当前时间>下次执行时间 过期了但是控制在5秒内不算过期
|
|
|
|
|
else if (nowTime > jobInfo.getTriggerNextTime()) {
|
|
|
|
|
// 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
|
|
|
|
|
|
|
|
|
|
// 1、trigger
|
|
|
|
|
//TODO 执行
|
|
|
|
|
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
|
|
|
|
|
logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
|
|
|
|
|
|
|
|
|
|
// 2、fresh next
|
|
|
|
|
//TODO 刷新下次执行时间
|
|
|
|
|
refreshNextValidTime(jobInfo, new Date());
|
|
|
|
|
|
|
|
|
|
// next-trigger-time in 5s, pre-read again
|
|
|
|
|
//TODO 如果任务在运行状态 且 下次执行时间在5秒内 则将 任务 缓存到 ringData
|
|
|
|
|
if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
|
|
|
|
|
|
|
|
|
|
// 1、make ring second
|
|
|
|
|
//TODO 获取下次执行时间 是哪一秒 作为缓存的key
|
|
|
|
|
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
|
|
|
|
|
|
|
|
|
|
// 2、push time ring
|
|
|
|
|
//TODO 缓存至 ringData
|
|
|
|
|
pushTimeRing(ringSecond, jobInfo.getId());
|
|
|
|
|
|
|
|
|
|
// 3、fresh next
|
|
|
|
|
@ -122,7 +152,9 @@ public class JobScheduleHelper {
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
}
|
|
|
|
|
//还没到下次执行时间的 缓存到 ringData
|
|
|
|
|
else {
|
|
|
|
|
// 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
|
|
|
|
|
|
|
|
|
|
// 1、make ring second
|
|
|
|
|
@ -140,6 +172,7 @@ public class JobScheduleHelper {
|
|
|
|
|
|
|
|
|
|
// 3、update trigger info
|
|
|
|
|
for (XxlJobInfo jobInfo: scheduleList) {
|
|
|
|
|
jobInfo.setLockStatus(0);
|
|
|
|
|
XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -199,6 +232,7 @@ public class JobScheduleHelper {
|
|
|
|
|
if (cost < 1000) { // scan-overtime, not wait
|
|
|
|
|
try {
|
|
|
|
|
// pre-read period: success > scan each second; fail > skip this period;
|
|
|
|
|
//随机休眠 没任务则休眠5秒内 有任务则休眠1秒内
|
|
|
|
|
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
|
|
|
|
|
} catch (InterruptedException e) {
|
|
|
|
|
if (!scheduleThreadToStop) {
|
|
|
|
|
@ -236,7 +270,8 @@ public class JobScheduleHelper {
|
|
|
|
|
try {
|
|
|
|
|
// second data
|
|
|
|
|
List<Integer> ringItemData = new ArrayList<>();
|
|
|
|
|
int nowSecond = Calendar.getInstance().get(Calendar.SECOND); // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
|
|
|
|
|
//TODO 避免处理耗时太长，跨过刻度，向前校验一个刻度；
|
|
|
|
|
int nowSecond = Calendar.getInstance().get(Calendar.SECOND);
|
|
|
|
|
for (int i = 0; i < 2; i++) {
|
|
|
|
|
List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
|
|
|
|
|
if (tmpData != null) {
|
|
|
|
|
@ -249,7 +284,7 @@ public class JobScheduleHelper {
|
|
|
|
|
if (ringItemData.size() > 0) {
|
|
|
|
|
// do trigger
|
|
|
|
|
for (int jobId: ringItemData) {
|
|
|
|
|
// do trigger
|
|
|
|
|
//TODO do trigger
|
|
|
|
|
JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
|
|
|
|
|
}
|
|
|
|
|
// clear
|
|
|
|
|
@ -355,6 +390,14 @@ public class JobScheduleHelper {
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// ---------------------- tools ----------------------
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* 获取下次执行时间
|
|
|
|
|
* @param jobInfo
|
|
|
|
|
* @param fromTime
|
|
|
|
|
* @return
|
|
|
|
|
* @throws Exception
|
|
|
|
|
*/
|
|
|
|
|
public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
|
|
|
|
|
ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
|
|
|
|
|
if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
|
|
|
|
|
|
lixiaoranxuhuaiyun
2 years ago
committed by
Gitee