From 749df8ef24e770427f238b20a2194eb6df35f68b Mon Sep 17 00:00:00 2001 From: islandempty <2667114200@qq.com> Date: Sat, 9 Oct 2021 22:30:32 +0800 Subject: [PATCH] =?UTF-8?q?fix[scheduler]:=E4=BF=AE=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E5=9F=BA=E4=BA=8E=E6=97=B6=E9=97=B4=E8=BD=AE=E7=AE=97=E6=B3=95?= =?UTF-8?q?=E7=9A=84=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E8=B0=83=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../zfoo/scheduler/manager/SchedulerBus.java | 109 ++++------------- .../zfoo/scheduler/timeWheelUtils/Bucket.java | 98 +++++++++++++++ .../scheduler/timeWheelUtils/TimeWheel.java | 113 ++++++++++++++++++ .../zfoo/scheduler/timeWheelUtils/Timer.java | 79 ++++++++++++ .../scheduler/timeWheelUtils/TimerTask.java | 51 ++++++++ .../com/zfoo/scheduler/util/TimeUtils.java | 2 +- 6 files changed, 367 insertions(+), 85 deletions(-) create mode 100644 scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Bucket.java create mode 100644 scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimeWheel.java create mode 100644 scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Timer.java create mode 100644 scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimerTask.java diff --git a/scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerBus.java b/scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerBus.java index 88048f3d..320c54f8 100644 --- a/scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerBus.java +++ b/scheduler/src/main/java/com/zfoo/scheduler/manager/SchedulerBus.java @@ -16,6 +16,8 @@ package com.zfoo.scheduler.manager; import com.zfoo.protocol.collection.CollectionUtils; import com.zfoo.scheduler.SchedulerContext; import com.zfoo.scheduler.model.vo.SchedulerDefinition; +import com.zfoo.scheduler.timeWheelUtils.Timer; +import com.zfoo.scheduler.timeWheelUtils.TimerTask; import com.zfoo.scheduler.util.TimeUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,21 +36,7 @@ public abstract class SchedulerBus { private static final Logger logger = LoggerFactory.getLogger(SchedulerBus.class); - private static final List schedulerDefList = new CopyOnWriteArrayList<>(); - - /** - * 上一次trigger触发时间 - */ - private static long lastTriggerTimestamp = 0L; - - - /** - * 在scheduler中，最小的triggerTimestamp - */ - private static long minTriggerTimestamp = 0L; - - - public static final long TRIGGER_MILLIS_INTERVAL = TimeUtils.MILLIS_PER_SECOND; + private static Timer timer = new Timer(); /** * scheduler默认只有一个单线程的线程池 @@ -56,75 +44,12 @@ public abstract class SchedulerBus { private static final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new SchedulerThreadFactory(1)); - static { - executor.scheduleAtFixedRate(() -> { - try { - triggerPerSecond(); - } catch (Exception e) { - logger.error("scheduler triggers an error.", e); - } - }, 0, TRIGGER_MILLIS_INTERVAL, TimeUnit.MILLISECONDS); - } - - - public static void refreshMinTriggerTimestamp() { - var minTimestamp = Long.MAX_VALUE; - for (var scheduler : schedulerDefList) { - if (scheduler.getTriggerTimestamp() < minTimestamp) { - minTimestamp = scheduler.getTriggerTimestamp(); - } - } - minTriggerTimestamp = minTimestamp; - } - - /** - * 每一秒执行一次，如果这个任务执行时间过长超过，比如10秒，执行完成后，不会再执行10次 - */ - private static void triggerPerSecond() { - var timestamp = TimeUtils.currentTimeMillis(); - - if (CollectionUtils.isEmpty(schedulerDefList)) { - return; - } - - - // 有人向前调整过机器时间，重新计算scheduler里的triggerTimestamp - // var diff = timestamp - lastTriggerTimestamp; - if (timestamp < lastTriggerTimestamp) { - for (SchedulerDefinition schedulerDef : schedulerDefList) { - var nextTriggerTimestamp = TimeUtils.getNextTimestampByCronExpression(schedulerDef.getCronExpression(), timestamp); - schedulerDef.setTriggerTimestamp(nextTriggerTimestamp); - } - refreshMinTriggerTimestamp(); - } - - // diff > 0, 没有人调整时间或者有人向后调整过机器时间，可以忽略，因为向后调整时间时间戳一定会大于triggerTimestamp，所以一定会触发 - lastTriggerTimestamp = timestamp; - - // 如果minSchedulerTriggerTimestamp大于timestamp，说明没有可执行的scheduler - if (timestamp < minTriggerTimestamp) { - return; - } - - var minTimestamp = Long.MAX_VALUE; - for (var scheduler : schedulerDefList) { - if (timestamp >= scheduler.getTriggerTimestamp()) { - // 到达触发时间，则执行runnable方法 - scheduler.getScheduler().invoke(); - // 重新设置下一次的触发时间戳 - var nextTriggerTimestamp = TimeUtils.getNextTimestampByCronExpression(scheduler.getCronExpression(), timestamp); - scheduler.setTriggerTimestamp(nextTriggerTimestamp); - } - if (scheduler.getTriggerTimestamp() < minTimestamp) { - minTimestamp = scheduler.getTriggerTimestamp(); - } - } - minTriggerTimestamp = minTimestamp; - } - public static void registerScheduler(SchedulerDefinition scheduler) { - schedulerDefList.add(scheduler); - refreshMinTriggerTimestamp(); + var timerTask = new TimerTask(scheduler.getTriggerTimestamp(), () -> { + scheduler.getScheduler().invoke(); + refreshTask(scheduler); + }); + timer.addTask(timerTask); } @@ -181,6 +106,22 @@ public abstract class SchedulerBus { return; } - schedulerDefList.add(SchedulerDefinition.valueOf(cron, runnable)); + SchedulerDefinition scheduler = SchedulerDefinition.valueOf(cron, runnable); + var timerTask = new TimerTask(scheduler.getTriggerTimestamp(), () -> { + scheduler.getScheduler().invoke(); + refreshTask(scheduler); + }); + timer.addTask(timerTask); + } + + public static void refreshTask(SchedulerDefinition schedulerDefinition) { + var timestamp = TimeUtils.currentTimeMillis(); + var nextTriggerTimestamp = TimeUtils.getNextTimestampByCronExpression(schedulerDefinition.getCronExpression(), timestamp); + schedulerDefinition.setTriggerTimestamp(nextTriggerTimestamp); + var timerTask = new TimerTask(schedulerDefinition.getTriggerTimestamp(), () -> { + schedulerDefinition.getScheduler().invoke(); + refreshTask(schedulerDefinition); + }); + timer.addTask(timerTask); } } diff --git a/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Bucket.java b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Bucket.java new file mode 100644 index 00000000..b3aaf474 --- /dev/null +++ b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Bucket.java @@ -0,0 +1,98 @@ +package com.zfoo.scheduler.timeWheelUtils; + +import java.util.concurrent.Delayed; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; + +/** + * 时间槽 + */ +public class Bucket implements Delayed { + + /** + * 过期时间 + */ + private AtomicLong expiration = new AtomicLong(-1L); + + /** + * 傀儡节点 + */ + private TimerTask root = new TimerTask(-1L, null); + + { + root.pre = root; + root.next = root; + } + + /** + * 设置过期时间 + */ + public boolean setExpiration(long expire) { + return expiration.getAndSet(expire) != expire; + } + + /** + * 获取过期时间 + */ + public long getExpiration() { + return expiration.get(); + } + + /** + * 新增任务，加入链表尾部 + */ + public void addTask(TimerTask timerTask) { + synchronized (this) { + if (timerTask.bucket == null) { + timerTask.bucket = this; + TimerTask tail = root.pre; + timerTask.next = root; + timerTask.pre = tail; + tail.next = timerTask; + root.pre = timerTask; + } + } + } + + /** + * 移除任务 + */ + public void removeTask(TimerTask timerTask) { + synchronized (this) { + if (timerTask.bucket.equals(this)) { + timerTask.next.pre = timerTask.pre; + timerTask.pre.next = timerTask.next; + timerTask.bucket = null; + timerTask.next = null; + timerTask.pre = null; + } + } + } + + /** + * 重新分配 + */ + public synchronized void flush(Consumer flush) { + TimerTask timerTask = root.next; + while (!timerTask.equals(root)) { + this.removeTask(timerTask); + flush.accept(timerTask); + timerTask = root.next; + } + expiration.set(-1L); + } + + @Override + public long getDelay(TimeUnit unit) { + return Math.max(0, unit.convert(expiration.get() - System.currentTimeMillis(), TimeUnit.MILLISECONDS)); + } + + @Override + public int compareTo(Delayed o) { + if (o instanceof Bucket) { + return Long.compare(expiration.get(), ((Bucket) o).expiration.get()); + } + return 0; + } +} diff --git a/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimeWheel.java b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimeWheel.java new file mode 100644 index 00000000..a18e9477 --- /dev/null +++ b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimeWheel.java @@ -0,0 +1,113 @@ +package com.zfoo.scheduler.timeWheelUtils; + +import java.util.concurrent.DelayQueue; + +/** + * 时间轮 + */ +public class TimeWheel { + + /** + * 一个时间槽的范围 + */ + private long tickMs; + + /** + * 时间轮大小 + */ + private int wheelSize; + + /** + * 时间轮的范围 + */ + private long interval; + + /** + * 时间槽 + */ + private Bucket[] buckets; + + /** + * 当前时间 + */ + private long currentTime; + + /** + * 上层时间轮 + */ + private volatile TimeWheel overflowWheel; + + /** + * 一个Timer只有一个delayQueue + */ + private DelayQueue delayQueue; + + public TimeWheel(long tickMs, int wheelSize, long currentTime, DelayQueue delayQueue) { + this.currentTime = currentTime; + this.tickMs = tickMs; + this.wheelSize = wheelSize; + this.interval = tickMs * wheelSize; + this.buckets = new Bucket[wheelSize]; + //currentTime为tickMs的整数倍 这里做取整操作 + this.currentTime = currentTime - (currentTime % tickMs); + this.delayQueue = delayQueue; + for (int i = 0; i < wheelSize; i++) { + buckets[i] = new Bucket(); + } + } + + /** + * 创建或者获取上层时间轮 + */ + private TimeWheel getOverflowWheel() { + if (overflowWheel == null) { + synchronized (this) { + if (overflowWheel == null) { + overflowWheel = new TimeWheel(interval, wheelSize, currentTime, delayQueue); + } + } + } + return overflowWheel; + } + + /** + * 添加任务到时间轮 + */ + public boolean addTask(TimerTask timerTask) { + long expiration = timerTask.getDelayMs(); + //过期任务直接执行 + if (expiration < currentTime + tickMs) { + //TODO这里可以直接执行定时任务 + return false; + } else if (expiration < currentTime + interval) { + //当前时间轮可以容纳该任务 加入时间槽 + Long bound = expiration / tickMs; + int bucketIndex = (int) (bound % wheelSize); + //System.out.println("tickMs:" + tickMs + "------bucketIndex:" + bucketIndex + "------expiration:" + expiration); + Bucket bucket = buckets[bucketIndex]; + bucket.addTask(timerTask); + if (bucket.setExpiration(bound * tickMs)) { + //添加到delayQueue中 + delayQueue.offer(bucket); + } + } else { + //放到上一层的时间轮 + TimeWheel timeWheel = getOverflowWheel(); + timeWheel.addTask(timerTask); + } + return true; + } + + /** + * 推进时间 + */ + public void advanceClock(long timestamp) { + if (timestamp >= currentTime + tickMs) { + currentTime = timestamp - (timestamp % tickMs); + if (overflowWheel != null) { + //推进上层时间轮时间 + this.getOverflowWheel().advanceClock(timestamp); + } + } + } +} diff --git a/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Timer.java b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Timer.java new file mode 100644 index 00000000..290914e8 --- /dev/null +++ b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/Timer.java @@ -0,0 +1,79 @@ +package com.zfoo.scheduler.timeWheelUtils; + +import com.zfoo.scheduler.util.TimeUtils; + +import java.util.concurrent.DelayQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +/** + * 定时器 + */ +public class Timer { + + /** + * 底层时间轮 + */ + private TimeWheel timeWheel; + + /** + * 一个Timer只有一个delayQueue + */ + private DelayQueue delayQueue = new DelayQueue<>(); + + /** + * 过期任务执行线程 + */ + private ExecutorService workerThreadPool; + + /** + * 轮询delayQueue获取过期任务线程 + */ + private ExecutorService bossThreadPool; + + /** + * 构造函数 + */ + public Timer() { + timeWheel = new TimeWheel(1000, 20, TimeUtils.currentTimeMillis(), delayQueue); + bossThreadPool = Executors.newFixedThreadPool(1); + workerThreadPool = Executors.newFixedThreadPool(10); + + //20ms获取一次过期任务 + bossThreadPool.submit(() -> { + while (true) { + this.advanceClock(20); + } + }); + } + + /** + * 添加任务 + */ + public void addTask(TimerTask timerTask) { + //添加失败任务直接执行 + if (!timeWheel.addTask(timerTask)) { + workerThreadPool.submit(timerTask.getTask()); + } + } + + /** + * 获取过期任务 + */ + private void advanceClock(long timestamp) { + try { + //阻塞获取队头元素 + Bucket bucket = delayQueue.poll(timestamp, TimeUnit.MILLISECONDS); + + if (bucket != null) { + //推进时间 + timeWheel.advanceClock(bucket.getExpiration()); + //执行过期任务（包含降级操作） + bucket.flush(this::addTask); + } + } catch (Exception e) { + e.printStackTrace(); + } + } +} diff --git a/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimerTask.java b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimerTask.java new file mode 100644 index 00000000..79794efb --- /dev/null +++ b/scheduler/src/main/java/com/zfoo/scheduler/timeWheelUtils/TimerTask.java @@ -0,0 +1,51 @@ +package com.zfoo.scheduler.timeWheelUtils; + +/** + * 任务 + */ +public class TimerTask { + + /** + * 延迟时间 + */ + private long delayMs; + + /** + * 任务 + */ + private Runnable task; + + /** + * 时间槽 + */ + protected Bucket bucket; + + /** + * 下一个节点 + */ + protected TimerTask next; + + /** + * 上一个节点 + */ + protected TimerTask pre; + + + public TimerTask(long delayMs, Runnable task) { + this.delayMs = delayMs; + this.task = task; + this.bucket = null; + this.next = null; + this.pre = null; + } + + public Runnable getTask() { + return task; + } + + public long getDelayMs() { + return delayMs; + } + + +} diff --git a/scheduler/src/main/java/com/zfoo/scheduler/util/TimeUtils.java b/scheduler/src/main/java/com/zfoo/scheduler/util/TimeUtils.java index ce45e270..32c25bfe 100644 --- a/scheduler/src/main/java/com/zfoo/scheduler/util/TimeUtils.java +++ b/scheduler/src/main/java/com/zfoo/scheduler/util/TimeUtils.java @@ -82,7 +82,7 @@ public abstract class TimeUtils { static { currentTimeMillis(); // 调用一下静态方法，使SchedulerBus静态代码块初始化 - SchedulerBus.refreshMinTriggerTimestamp(); + //SchedulerBus.refreshMinTriggerTimestamp(); } /**