Cron4j调度框架实现原理

妖狐艹你老母 2022-05-08 00:40 173阅读 0赞

之前有篇博客我们介绍了Cron4j改造与学习的内容 [Cron4j调度框架学习与改造][Cron4j]，这篇博客我们从源码上看看Cron4j的实现机制。

示例：

public class TestMain {
    
    	public static void main(String[] args) {
    		
    		// Creates a Scheduler instance.
    		Scheduler s = new Scheduler();
    		// Schedule a once-a-minute task.
    		
    		s.schedule("*/3 * * * * *", new Runnable() {
    					int i = 0;
    					public void run() {
    						i++;
    						System.out.println("Another minute ticked away..."+System.currentTimeMillis()/1000);
    					}
    				});
    		// Starts the scheduler.
    		s.start();
    		// Will run for ten minutes.
    		try {
    			Thread.sleep(1000L * 60L * 10L);
    		} catch (InterruptedException e) {
    			;
    		}
    		// Stops the scheduler.
    		s.stop();
    
    	}
    
    }

Scheduler调用schedule方法添加一个线程，在Scheduler掉头start方法是会创建一个一直执行的线程TimerThread

public void start() throws IllegalStateException {
    		synchronized (lock) {
    			if (started) {
    				throw new IllegalStateException("Scheduler already started");
    			}
    			// Initializes required lists.
    			launchers = new ArrayList();
    			executors = new ArrayList();
    			// Starts the timer thread.
                //创建线程执行
    			timer = new TimerThread(this);
    			timer.setDaemon(daemon);
    			timer.start();
    			// Change the state of the scheduler.
    			started = true;
    		}
    	}

在TimerThread线程run方法中会一直执行线程，每隔一秒会计算检查cron表达式对应的定时任务规则，如果匹配则执行任务。

（1）首先会获取下一秒时间

（2）计算当前时间与下一秒时间差

（3）如果时间差超过1秒，则可能存在时间跳变情况，忽略重新计算

（4）睡眠时间差时间

（5）调用scheduler.spawnLauncher(millis);方法计算是否存在任务在这个时间点需要执行（每隔一秒计算一次）

public void run() {
    		// What time is it?
    		long millis = System.currentTimeMillis();
    		// Work until the scheduler is started.
    		//保持线程存货
    		for (;;) {
    			// Calculating next seconds.
    			//获取当前时间的下一秒
    			long nextSecond = ((System.currentTimeMillis() / 1000) + 1) * 1000;
    			// Coffee break 'till next seconds comes!
    			//获取当前时间
    			long sleepTime = (nextSecond - System.currentTimeMillis());
    			//time is changed
    			//如果存在时间跳变情况则直接忽略
    			if(sleepTime > 1000) {
    				continue;
    			}
    			//进行睡眠
    			if (sleepTime > 0) {
    				try {
    					safeSleep(sleepTime);
    				} catch (InterruptedException e) {
    					// Must exit!
    					break;
    				}
    			}
    			millis = System.currentTimeMillis();
    			// Launching the launching thread!
    			//计算是否要进行任务执行
    			scheduler.spawnLauncher(millis);
    		}
    		scheduler = null;
    	}

在Scheduler的spawnLauncher中会新起一个线程LauncherThread去分别计算是否定时任务与当时时间匹配，如果匹配则执行

LauncherThread spawnLauncher(long referenceTimeInMillis) {
    		TaskCollector[] nowCollectors;
    		synchronized (collectors) {
    			int size = collectors.size();
    			nowCollectors = new TaskCollector[size];
    			for (int i = 0; i < size; i++) {
    				nowCollectors[i] = (TaskCollector) collectors.get(i);
    			}
    		}
    		//新起任务去进行规制计算并执行
    		LauncherThread l = new LauncherThread(this, nowCollectors,
    				referenceTimeInMillis);
    		synchronized (launchers) {
    			launchers.add(l);
    		}
    		l.setDaemon(daemon);
    		l.start();
    		return l;
    	}

在LauncherThread中会根据cron表达式与当前时间进行匹配，如果匹配成功则将相应的任务添加到线程池中去执行

public void run() {
    		outer: for (int i = 0; i < collectors.length; i++) {
    			TaskTable taskTable = collectors[i].getTasks();
    			int size = taskTable.size();
    			for (int j = 0; j < size; j++) {
    				if (isInterrupted()) {
    					break outer;
    				}
    				SchedulingPattern pattern = taskTable.getSchedulingPattern(j);
    				//cron表达式与当前时间进行匹配
    				if (pattern.match(scheduler.getTimeZone(), referenceTimeInMillis)) {
    					Task task = taskTable.getTask(j);
    					//任务添加到线程池中执行
    					scheduler.spawnExecutor(task);
    				}
    			}
    		}
    		// Notifies completed.
    		scheduler.notifyLauncherCompleted(this);
    	}

任务添加到线程池中进行执行

TaskExecutor spawnExecutor(Task task) {
    		TaskExecutor e = new TaskExecutor(this, task);
    		synchronized (executors) {
    			executors.add(e);
    		}
    		e.start(daemon);
    		return e;
    	}

总结：

cron4j目前支持最小的时间单位为分钟，这样在线程TimerThread中其实每分钟才会对任务进行一次时间校验并执行，当然也没有对时间跳变进行处理，可能会导致时间跳变之后很长一段时间内任务不再执行。

改造：目前个人对cron4j进行改造，支持最小时间单位为妙，这样TimerThread执行的频率会变高，同样对于时间跳变也进行了处理，当时间间隔超过1秒时则重新进行校验处理。

[Cron4j]: https://blog.csdn.net/qq924862077/article/details/80970149