content: package com.xxl.job.admin.core.thread;
  
  import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
  import com.xxl.job.admin.core.cron.CronExpression;
  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 org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.sql.Connection;
  import java.sql.PreparedStatement;
  import java.sql.SQLException;
  import java.util.*;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.TimeUnit;
  
  /**
   * @author xuxueli 2019-05-21
   */
  public class JobScheduleHelper {
  
  	private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
  
  	private static JobScheduleHelper instance = new JobScheduleHelper();
  
  	public static JobScheduleHelper getInstance() {
  		return instance;
  	}
  
  	public static final long PRE_READ_MS = 5000; // pre read
  
  	private Thread scheduleThread;
  
  	private Thread ringThread;
  
  	private volatile boolean scheduleThreadToStop = false;
  
  	private volatile boolean ringThreadToStop = false;
  
  	private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();
  
  	public void start() {
  
  		// schedule thread
  		scheduleThread = new Thread(new Runnable() {
  			@Override
  			public void run() {
  
  				try {
  					TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
  				}
  				catch (InterruptedException e) {
  					if (!scheduleThreadToStop) {
  						logger.error(e.getMessage(), e);
  					}
  				}
  				logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
  
  				// pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms,
  				// qps = 1000/50 = 20)
  				int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax()
  						+ XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
  
  				while (!scheduleThreadToStop) {
  
  					// Scan Job
  					long start = System.currentTimeMillis();
  
  					Connection conn = null;
  					Boolean connAutoCommit = null;
  					PreparedStatement preparedStatement = null;
  
  					boolean preReadSuc = true;
  					try {
  
  						conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
  						connAutoCommit = conn.getAutoCommit();
  						conn.setAutoCommit(false);
  
  						preparedStatement = conn.prepareStatement(
  								"select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
  						preparedStatement.execute();
  
  						// tx start
  
  						// 1、pre read
  						long nowTime = System.currentTimeMillis();
  						List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao()
  								.scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
  						if (scheduleList != null && scheduleList.size() > 0) {
  							// 2、push time-ring
  							for (XxlJobInfo jobInfo : scheduleList) {
  
  								// time-ring jump
  								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);
  										logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = "
  												+ jobInfo.getId());
  									}
  
  									// 2、fresh next
  									refreshNextValidTime(jobInfo, new Date());
  
  								}
  								else if (nowTime > jobInfo.getTriggerNextTime()) {
  									// 2.2、trigger-expire < 5s：direct-trigger && make
  									// next-trigger-time
  
  									// 1、trigger
  									JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null,
  											null);
  									logger.debug(
  											">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
  
  									// 2、fresh next
  									refreshNextValidTime(jobInfo, new Date());
  
  									// next-trigger-time in 5s, pre-read again
  									if (jobInfo.getTriggerStatus() == 1
  											&& nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
  
  										// 1、make ring second
  										int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
  
  										// 2、push time ring
  										pushTimeRing(ringSecond, jobInfo.getId());
  
  										// 3、fresh next
  										refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
  
  									}
  
  								}
  								else {
  									// 2.3、trigger-pre-read：time-ring trigger && make
  									// next-trigger-time
  
  									// 1、make ring second
  									int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
  
  									// 2、push time ring
  									pushTimeRing(ringSecond, jobInfo.getId());
  
  									// 3、fresh next
  									refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
  
  								}
  
  							}
  
  							// 3、update trigger info
  							for (XxlJobInfo jobInfo : scheduleList) {
  								XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
  							}
  
  						}
  						else {
  							preReadSuc = false;
  						}
  
  						// tx stop
  
  					}
  					catch (Exception e) {
  						if (!scheduleThreadToStop) {
  							logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
  						}
  					}
  					finally {
  
  						// commit
  						if (conn != null) {
  							try {
  								conn.commit();
  							}
  							catch (SQLException e) {
  								if (!scheduleThreadToStop) {
  									logger.error(e.getMessage(), e);
  								}
  							}
  							try {
  								conn.setAutoCommit(connAutoCommit);
  							}
  							catch (SQLException e) {
  								if (!scheduleThreadToStop) {
  									logger.error(e.getMessage(), e);
  								}
  							}
  							try {
  								conn.close();
  							}
  							catch (SQLException e) {
  								if (!scheduleThreadToStop) {
  									logger.error(e.getMessage(), e);
  								}
  							}
  						}
  
  						// close PreparedStatement
  						if (null != preparedStatement) {
  							try {
  								preparedStatement.close();
  							}
  							catch (SQLException e) {
  								if (!scheduleThreadToStop) {
  									logger.error(e.getMessage(), e);
  								}
  							}
  						}
  					}
  					long cost = System.currentTimeMillis() - start;
  
  					// Wait seconds, align second
  					if (cost < 1000) { // scan-overtime, not wait
  						try {
  							// pre-read period: success > scan each second; fail > skip
  							// this period;
  							TimeUnit.MILLISECONDS
  									.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
  						}
  						catch (InterruptedException e) {
  							if (!scheduleThreadToStop) {
  								logger.error(e.getMessage(), e);
  							}
  						}
  					}
  
  				}
  
  				logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
  			}
  		});
  		scheduleThread.setDaemon(true);
  		scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
  		scheduleThread.start();
  
  		// ring thread
  		ringThread = new Thread(new Runnable() {
  			@Override
  			public void run() {
  
  				while (!ringThreadToStop) {
  
  					// align second
  					try {
  						TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
  					}
  					catch (InterruptedException e) {
  						if (!ringThreadToStop) {
  							logger.error(e.getMessage(), e);
  						}
  					}
  
  					try {
  						// second data
  						List<Integer> ringItemData = new ArrayList<>();
  						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) {
  								ringItemData.addAll(tmpData);
  							}
  						}
  
  						// ring trigger
  						logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = "
  								+ Arrays.asList(ringItemData));
  						if (ringItemData.size() > 0) {
  							// do trigger
  							for (int jobId : ringItemData) {
  								// do trigger
  								JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
  							}
  							// clear
  							ringItemData.clear();
  						}
  					}
  					catch (Exception e) {
  						if (!ringThreadToStop) {
  							logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
  						}
  					}
  				}
  				logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
  			}
  		});
  		ringThread.setDaemon(true);
  		ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
  		ringThread.start();
  	}
  
  	private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
  		Date nextValidTime = generateNextValidTime(jobInfo, fromTime);
  		if (nextValidTime != null) {
  			jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
  			jobInfo.setTriggerNextTime(nextValidTime.getTime());
  		}
  		else {
  			jobInfo.setTriggerStatus(0);
  			jobInfo.setTriggerLastTime(0);
  			jobInfo.setTriggerNextTime(0);
  			logger.warn(
  					">>>>>>>>>>> xxl-job, refreshNextValidTime fail for job: jobId={}, scheduleType={}, scheduleConf={}",
  					jobInfo.getId(), jobInfo.getScheduleType(), jobInfo.getScheduleConf());
  		}
  	}
  
  	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, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData));
  	}
  
  	public void toStop() {
  
  		// 1、stop schedule
  		scheduleThreadToStop = true;
  		try {
  			TimeUnit.SECONDS.sleep(1); // wait
  		}
  		catch (InterruptedException e) {
  			logger.error(e.getMessage(), e);
  		}
  		if (scheduleThread.getState() != Thread.State.TERMINATED) {
  			// interrupt and wait
  			scheduleThread.interrupt();
  			try {
  				scheduleThread.join();
  			}
  			catch (InterruptedException e) {
  				logger.error(e.getMessage(), e);
  			}
  		}
  
  		// if has ring data
  		boolean hasRingData = false;
  		if (!ringData.isEmpty()) {
  			for (int second : ringData.keySet()) {
  				List<Integer> tmpData = ringData.get(second);
  				if (tmpData != null && tmpData.size() > 0) {
  					hasRingData = true;
  					break;
  				}
  			}
  		}
  		if (hasRingData) {
  			try {
  				TimeUnit.SECONDS.sleep(8);
  			}
  			catch (InterruptedException e) {
  				logger.error(e.getMessage(), e);
  			}
  		}
  
  		// stop ring (wait job-in-memory stop)
  		ringThreadToStop = true;
  		try {
  			TimeUnit.SECONDS.sleep(1);
  		}
  		catch (InterruptedException e) {
  			logger.error(e.getMessage(), e);
  		}
  		if (ringThread.getState() != Thread.State.TERMINATED) {
  			// interrupt and wait
  			ringThread.interrupt();
  			try {
  				ringThread.join();
  			}
  			catch (InterruptedException e) {
  				logger.error(e.getMessage(), e);
  			}
  		}
  
  		logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
  	}
  
  	// ---------------------- tools ----------------------
  	public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
  		ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
  		if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {
  			Date nextValidTime = new CronExpression(jobInfo.getScheduleConf()).getNextValidTimeAfter(fromTime);
  			return nextValidTime;
  		}
  		else if (ScheduleTypeEnum.FIX_RATE == scheduleTypeEnum /*
  																 * || ScheduleTypeEnum.
  																 * FIX_DELAY ==
  																 * scheduleTypeEnum
  																 */) {
  			return new Date(fromTime.getTime() + Integer.valueOf(jobInfo.getScheduleConf()) * 1000);
  		}
  		return null;
  	}
  
  }
  