网站怎么做实名认证,视觉设计包括哪些,已矣seo排名点击软件,免费做外贸的网站平台一、摘要在很多业务的系统中#xff0c;我们常常需要定时的执行一些任务#xff0c;例如定时发短信、定时变更数据、定时发起促销活动等等。在上篇文章中#xff0c;我们简单的介绍了定时任务的使用方式#xff0c;不同的架构对应的解决方案也有所不同#xff0c;总结起来… 一、摘要在很多业务的系统中我们常常需要定时的执行一些任务例如定时发短信、定时变更数据、定时发起促销活动等等。在上篇文章中我们简单的介绍了定时任务的使用方式不同的架构对应的解决方案也有所不同总结起来主要分单机和分布式两大类本文会重点分析下单机的定时任务实现原理以及优缺点分布式框架的实现原理会在后续文章中进行分析。从单机角度定时任务实现主要有以下 3 种方案while sleep 组合最小堆实现时间轮实现二、whilesleep组合whilesleep 方案简单的说就是定义一个线程然后 while 循环通过 sleep 延迟时间来达到周期性调度任务。简单示例如下public static void main(String[] args) {final long timeInterval  5000;new Thread(new Runnable() {Overridepublic void run() {while (true) {System.out.println(Thread.currentThread().getName()  每隔5秒执行一次);try {Thread.sleep(timeInterval);} catch (InterruptedException e) {e.printStackTrace();}}}}).start(); } 实现上非常简单如果我们想在创建一个每隔3秒钟执行一次任务怎么办呢同样的也可以在创建一个线程然后间隔性的调度方法但是如果创建了大量这种类型的线程这个时候会发现大量的定时任务线程在调度切换时性能消耗会非常大而且整体效率低面对这种在情况大佬们也想到了于是想出了用一个线程将所有的定时任务存起来事先排好序按照一定的规则来调度这样不就可以极大的减少每个线程的切换消耗吗正因此JDK 中的 Timer 定时器由此诞生了三、最小堆实现所谓最小堆方案正如我们上面所说的每当有新任务加入的时候会把需要即将要执行的任务排到前面同时会有一个线程不断的轮询判断如果当前某个任务已经到达执行时间点就会立即执行具体实现代表就是 JDK 中的 Timer 定时器3.1、Timer首先我们来一个简单的 Timer 定时器例子public static void main(String[] args) {Timer timer  new Timer();//每隔1秒调用一次timer.schedule(new TimerTask() {Overridepublic void run() {System.out.println(test1);}}, 1000, 1000);//每隔3秒调用一次timer.schedule(new TimerTask() {Overridepublic void run() {System.out.println(test2);}}, 3000, 3000);} 实现上好像跟我们上面介绍的 whilesleep 方案差不多同样也是起一个TimerTask线程任务只不过共用一个Timer调度器。下面我们一起来打开源码看看里面到底有些啥进入Timer.schedule()方法从方法上可以看出这里主要做参数验证其中TimerTask是一个线程任务delay表示延迟多久执行单位毫秒period表示多久执行一次单位毫秒public void schedule(TimerTask task, long delay, long period) {if (delay  0)throw new IllegalArgumentException(Negative delay.);if (period  0)throw new IllegalArgumentException(Non-positive period.);sched(task, System.currentTimeMillis()delay, -period); } 接着看sched()方法这步操作中可以很清晰的看到在同步代码块里会将task对象加入到queueprivate void sched(TimerTask task, long time, long period) {if (time  0)throw new IllegalArgumentException(Illegal execution time.);// Constrain value of period sufficiently to prevent numeric// overflow while still being effectively infinitely large.if (Math.abs(period)  (Long.MAX_VALUE  1))period  1;synchronized(queue) {if (!thread.newTasksMayBeScheduled)throw new IllegalStateException(Timer already cancelled.);synchronized(task.lock) {if (task.state ! TimerTask.VIRGIN)throw new IllegalStateException(Task already scheduled or cancelled);task.nextExecutionTime  time;task.period  period;task.state  TimerTask.SCHEDULED;}queue.add(task);if (queue.getMin()  task)queue.notify();} } 我们继续来看queue对象任务会将入到TaskQueue队列中同时在Timer初始化阶段会将TaskQueue作为参数传入到TimerThread线程中并且起到线程public class Timer {private final TaskQueue queue  new TaskQueue();private final TimerThread thread  new TimerThread(queue);public Timer() {this(Timer-  serialNumber());}public Timer(String name) {thread.setName(name);thread.start();}//... } 而TaskQueue其实是一个最小堆的数据实体类源码如下每当有新元素加入的时候会对原来的数组进行重排会将即将要执行的任务排在数组的前面class TaskQueue {private TimerTask[] queue  new TimerTask[128];private int size  0;void add(TimerTask task) {// Grow backing store if necessaryif (size  1  queue.length)queue  Arrays.copyOf(queue, 2*queue.length);queue[size]  task;fixUp(size);}private void fixUp(int k) {while (k  1) {int j  k  1;if (queue[j].nextExecutionTime  queue[k].nextExecutionTime)break;TimerTask tmp  queue[j];queue[j]  queue[k];queue[k]  tmp;k  j;}}//.... } 最后我们来看看TimerThreadTimerThread其实就是一个任务调度线程首先从TaskQueue里面获取排在最前面的任务然后判断它是否到达任务执行时间点如果已到达就会立刻执行任务class TimerThread extends Thread {boolean newTasksMayBeScheduled  true;private TaskQueue queue;TimerThread(TaskQueue queue) {this.queue  queue;}public void run() {try {mainLoop();} finally {// Someone killed this Thread, behave as if Timer cancelledsynchronized(queue) {newTasksMayBeScheduled  false;queue.clear();  // Eliminate obsolete references}}}/*** The main timer loop.  (See class comment.)*/private void mainLoop() {while (true) {try {TimerTask task;boolean taskFired;synchronized(queue) {// Wait for queue to become non-emptywhile (queue.isEmpty()  newTasksMayBeScheduled)queue.wait();if (queue.isEmpty())break; // Queue is empty and will forever remain; die// Queue nonempty; look at first evt and do the right thinglong currentTime, executionTime;task  queue.getMin();synchronized(task.lock) {if (task.state  TimerTask.CANCELLED) {queue.removeMin();continue;  // No action required, poll queue again}currentTime  System.currentTimeMillis();executionTime  task.nextExecutionTime;if (taskFired  (executionTimecurrentTime)) {if (task.period  0) { // Non-repeating, removequeue.removeMin();task.state  TimerTask.EXECUTED;} else { // Repeating task, reschedulequeue.rescheduleMin(task.period0 ? currentTime   - task.period: executionTime  task.period);}}}if (!taskFired) // Task hasnt yet fired; waitqueue.wait(executionTime - currentTime);}if (taskFired)  // Task fired; run it, holding no lockstask.run();} catch(InterruptedException e) {}}} } 总结这个利用最小堆实现的方案相比 while sleep 方案多了一个线程来管理所有的任务优点就是减少了线程之间的性能开销提升了执行效率但是同样也带来的了一些缺点整体的新加任务写入效率变成了 O(log(n))。同时细心的发现这个方案还有以下几个缺点串行阻塞调度线程只有一个长任务会阻塞短任务的执行例如A任务跑了一分钟B任务至少需要等1分钟才能跑容错能力差没有异常处理能力一旦一个任务执行故障后续任务都无法执行3.2、ScheduledThreadPoolExecutor鉴于 Timer 的上述缺陷从 Java 5 开始推出了基于线程池设计的 ScheduledThreadPoolExecutor 。其设计思想是每一个被调度的任务都会由线程池来管理执行因此任务是并发执行的相互之间不会受到干扰。需要注意的是只有当任务的执行时间到来时ScheduledThreadPoolExecutor 才会真正启动一个线程其余时间 ScheduledThreadPoolExecutor 都是在轮询任务的状态。简单的使用示例public static void main(String[] args) {ScheduledThreadPoolExecutor executor  new ScheduledThreadPoolExecutor(3);//启动1秒之后每隔1秒执行一次executor.scheduleAtFixedRate((new Runnable() {Overridepublic void run() {System.out.println(test3);}}),1,1, TimeUnit.SECONDS);//启动1秒之后每隔3秒执行一次executor.scheduleAtFixedRate((new Runnable() {Overridepublic void run() {System.out.println(test4);}}),1,3, TimeUnit.SECONDS); } 同样的我们首先打开源码看看里面到底做了啥进入scheduleAtFixedRate()方法首先是校验基本参数然后将任务作为封装到ScheduledFutureTask线程中ScheduledFutureTask继承自RunnableScheduledFuture并作为参数调用delayedExecute()方法进行预处理public ScheduledFuture? scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {if (command  null || unit  null)throw new NullPointerException();if (period  0)throw new IllegalArgumentException();ScheduledFutureTaskVoid sft new ScheduledFutureTaskVoid(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));RunnableScheduledFutureVoid t  decorateTask(command, sft);sft.outerTask  t;delayedExecute(t);return t; } 继续看delayedExecute()方法可以很清晰的看到当线程池没有关闭的时候会通过super.getQueue().add(task)操作将任务加入到队列同时调用ensurePrestart()方法做预处理private void delayedExecute(RunnableScheduledFuture? task) {if (isShutdown())reject(task);else {super.getQueue().add(task);if (isShutdown() !canRunInCurrentRunState(task.isPeriodic()) remove(task))task.cancel(false);else//预处理ensurePrestart();} } 其中super.getQueue()得到的是一个自定义的new DelayedWorkQueue()阻塞队列数据存储方面也是一个最小堆结构的队列这一点在初始化new ScheduledThreadPoolExecutor()的时候可以看出public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue()); } 打开源码可以看到DelayedWorkQueue其实是ScheduledThreadPoolExecutor中的一个静态内部类在添加的时候会将任务加入到RunnableScheduledFuture数组中同时线程池中的Woker线程会通过调用任务队列中的take()方法获取对应的ScheduledFutureTask线程任务接着执行对应的任务线程static class DelayedWorkQueue extends AbstractQueueRunnableimplements BlockingQueueRunnable {private static final int INITIAL_CAPACITY  16;private RunnableScheduledFuture?[] queue new RunnableScheduledFuture?[INITIAL_CAPACITY];private final ReentrantLock lock  new ReentrantLock();private int size  0;   //....public boolean add(Runnable e) {return offer(e);}public boolean offer(Runnable x) {if (x  null)throw new NullPointerException();RunnableScheduledFuture? e  (RunnableScheduledFuture?)x;final ReentrantLock lock  this.lock;lock.lock();try {int i  size;if (i  queue.length)grow();size  i  1;if (i  0) {queue[0]  e;setIndex(e, 0);} else {siftUp(i, e);}if (queue[0]  e) {leader  null;available.signal();}} finally {lock.unlock();}return true;}public RunnableScheduledFuture? take() throws InterruptedException {final ReentrantLock lock  this.lock;lock.lockInterruptibly();try {for (;;) {RunnableScheduledFuture? first  queue[0];if (first  null)available.await();else {long delay  first.getDelay(NANOSECONDS);if (delay  0)return finishPoll(first);first  null; // dont retain ref while waitingif (leader ! null)available.await();else {Thread thisThread  Thread.currentThread();leader  thisThread;try {available.awaitNanos(delay);} finally {if (leader  thisThread)leader  null;}}}}} finally {if (leader  null  queue[0] ! null)available.signal();lock.unlock();}} } 回到我们最开始说到的ScheduledFutureTask任务线程类最终执行任务的其实就是它ScheduledFutureTask任务线程才是真正执行任务的线程类只是绕了一圈做了很多包装run()方法就是真正执行定时任务的方法。private class ScheduledFutureTaskVextends FutureTaskV implements RunnableScheduledFutureV {/** Sequence number to break ties FIFO */private final long sequenceNumber;/** The time the task is enabled to execute in nanoTime units */private long time;/*** Period in nanoseconds for repeating tasks.  A positive* value indicates fixed-rate execution.  A negative value* indicates fixed-delay execution.  A value of 0 indicates a* non-repeating task.*/private final long period;/** The actual task to be re-enqueued by reExecutePeriodic */RunnableScheduledFutureV outerTask  this;/*** Overrides FutureTask version so as to reset/requeue if periodic.*/public void run() {boolean periodic  isPeriodic();if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();else if (ScheduledFutureTask.super.runAndReset()) {setNextRunTime();reExecutePeriodic(outerTask);}}//... } 3.3、小结ScheduledExecutorService 相比 Timer 定时器完美的解决上面说到的 Timer 存在的两个缺点在单体应用里面使用 ScheduledExecutorService 可以解决大部分需要使用定时任务的业务需求但是这是否意味着它是最佳的解决方案呢我们发现线程池中 ScheduledExecutorService 的排序容器跟 Timer 一样都是采用最小堆的存储结构新任务加入排序效率是O(log(n))执行取任务是O(1)。这里的写入排序效率其实是有空间可提升的有可能优化到O(1)的时间复杂度也就是我们下面要介绍的时间轮实现四、时间轮实现所谓时间轮RingBuffer实现从数据结构上看简单的说就是循环队列从名称上看可能感觉很抽象。它其实就是一个环形的数组如图所示假设我们创建了一个长度为 8 的时间轮。插入、取值流程1.当我们需要新建一个 1s 延时任务的时候则只需要将它放到下标为 1 的那个槽中2、3、...、7也同样如此。2.而如果是新建一个 10s 的延时任务则需要将它放到下标为 2 的槽中但同时需要记录它所对应的圈数也就是 1 圈不然就和 2 秒的延时消息重复了3.当创建一个 21s 的延时任务时它所在的位置就在下标为 5 的槽中同样的需要为他加上圈数为 2依次类推...因此总结起来有两个核心的变量数组下标表示某个任务延迟时间从数据操作上对执行时间点进行取余圈数表示需要循环圈数通过这张图可以更直观的理解当我们需要取出延时任务时只需要每秒往下移动这个指针然后取出该位置的所有任务即可取任务的时间消耗为O(1)。当我们需要插入任务式也只需要计算出对应的下表和圈数即可将任务插入到对应的数组位置中插入任务的时间消耗为O(1)。如果时间轮的槽比较少会导致某一个槽上的任务非常多那么效率也比较低这就和 HashMap 的 hash 冲突是一样的因此在设计槽的时候不能太大也不能太小。4.1、代码实现首先创建一个RingBufferWheel时间轮定时任务管理器public class RingBufferWheel {private Logger logger  LoggerFactory.getLogger(RingBufferWheel.class);/*** default ring buffer size*/private static final int STATIC_RING_SIZE  64;private Object[] ringBuffer;private int bufferSize;/*** business thread pool*/private ExecutorService executorService;private volatile int size  0;/**** task stop sign*/private volatile boolean stop  false;/*** task start sign*/private volatile AtomicBoolean start  new AtomicBoolean(false);/*** total tick times*/private AtomicInteger tick  new AtomicInteger();private Lock lock  new ReentrantLock();private Condition condition  lock.newCondition();private AtomicInteger taskId  new AtomicInteger();private MapInteger, Task taskMap  new ConcurrentHashMap(16);/*** Create a new delay task ring buffer by default size** param executorService the business thread pool*/public RingBufferWheel(ExecutorService executorService) {this.executorService  executorService;this.bufferSize  STATIC_RING_SIZE;this.ringBuffer  new Object[bufferSize];}/*** Create a new delay task ring buffer by custom buffer size** param executorService the business thread pool* param bufferSize      custom buffer size*/public RingBufferWheel(ExecutorService executorService, int bufferSize) {this(executorService);if (!powerOf2(bufferSize)) {throw new RuntimeException(bufferSize[  bufferSize  ] must be a power of 2);}this.bufferSize  bufferSize;this.ringBuffer  new Object[bufferSize];}/*** Add a task into the ring buffer(thread safe)** param task business task extends {link Task}*/public int addTask(Task task) {int key  task.getKey();int id;try {lock.lock();int index  mod(key, bufferSize);task.setIndex(index);SetTask tasks  get(index);int cycleNum  cycleNum(key, bufferSize);if (tasks ! null) {task.setCycleNum(cycleNum);tasks.add(task);} else {task.setIndex(index);task.setCycleNum(cycleNum);SetTask sets  new HashSet();sets.add(task);put(key, sets);}id  taskId.incrementAndGet();task.setTaskId(id);taskMap.put(id, task);size;} finally {lock.unlock();}start();return id;}/*** Cancel task by taskId* param id unique id through {link #addTask(Task)}* return*/public boolean cancel(int id) {boolean flag  false;SetTask tempTask  new HashSet();try {lock.lock();Task task  taskMap.get(id);if (task  null) {return false;}SetTask tasks  get(task.getIndex());for (Task tk : tasks) {if (tk.getKey()  task.getKey()  tk.getCycleNum()  task.getCycleNum()) {size--;flag  true;taskMap.remove(id);} else {tempTask.add(tk);}}//update origin dataringBuffer[task.getIndex()]  tempTask;} finally {lock.unlock();}return flag;}/*** Thread safe** return the size of ring buffer*/public int taskSize() {return size;}/*** Same with method {link #taskSize}* return*/public int taskMapSize(){return taskMap.size();}/*** Start background thread to consumer wheel timer, it will always run until you call method {link #stop}*/public void start() {if (!start.get()) {if (start.compareAndSet(start.get(), true)) {logger.info(Delay task is starting);Thread job  new Thread(new TriggerJob());job.setName(consumer RingBuffer thread);job.start();start.set(true);}}}/*** Stop consumer ring buffer thread** param force True will force close consumer thread and discard all pending tasks*              otherwise the consumer thread waits for all tasks to completes before closing.*/public void stop(boolean force) {if (force) {logger.info(Delay task is forced stop);stop  true;executorService.shutdownNow();} else {logger.info(Delay task is stopping);if (taskSize()  0) {try {lock.lock();condition.await();stop  true;} catch (InterruptedException e) {logger.error(InterruptedException, e);} finally {lock.unlock();}}executorService.shutdown();}}private SetTask get(int index) {return (SetTask) ringBuffer[index];}private void put(int key, SetTask tasks) {int index  mod(key, bufferSize);ringBuffer[index]  tasks;}/*** Remove and get task list.* param key* return task list*/private SetTask remove(int key) {SetTask tempTask  new HashSet();SetTask result  new HashSet();SetTask tasks  (SetTask) ringBuffer[key];if (tasks  null) {return result;}for (Task task : tasks) {if (task.getCycleNum()  0) {result.add(task);size2Notify();} else {// decrement 1 cycle number and update origin datatask.setCycleNum(task.getCycleNum() - 1);tempTask.add(task);}// remove task, and free the memory.taskMap.remove(task.getTaskId());}//update origin dataringBuffer[key]  tempTask;return result;}private void size2Notify() {try {lock.lock();size--;if (size  0) {condition.signal();}} finally {lock.unlock();}}private boolean powerOf2(int target) {if (target  0) {return false;}int value  target  (target - 1);if (value ! 0) {return false;}return true;}private int mod(int target, int mod) {// equals target % modtarget  target  tick.get();return target  (mod - 1);}private int cycleNum(int target, int mod) {//equals target/modreturn target  Integer.bitCount(mod - 1);}/*** An abstract class used to implement business.*/public abstract static class Task extends Thread {private int index;private int cycleNum;private int key;/*** The unique ID of the task*/private int taskId ;Overridepublic void run() {}public int getKey() {return key;}/**** param key Delay time(seconds)*/public void setKey(int key) {this.key  key;}public int getCycleNum() {return cycleNum;}private void setCycleNum(int cycleNum) {this.cycleNum  cycleNum;}public int getIndex() {return index;}private void setIndex(int index) {this.index  index;}public int getTaskId() {return taskId;}public void setTaskId(int taskId) {this.taskId  taskId;}}private class TriggerJob implements Runnable {Overridepublic void run() {int index  0;while (!stop) {try {SetTask tasks  remove(index);for (Task task : tasks) {executorService.submit(task);}if (index  bufferSize - 1) {index  0;}//Total tick number of recordstick.incrementAndGet();TimeUnit.SECONDS.sleep(1);} catch (Exception e) {logger.error(Exception, e);}}logger.info(Delay task has stopped);}} }接着编写一个客户端测试客户端public static void main(String[] args) {RingBufferWheel ringBufferWheel  new RingBufferWheel( Executors.newFixedThreadPool(2));for (int i  0; i  3; i) {RingBufferWheel.Task job  new Job();job.setKey(i);ringBufferWheel.addTask(job);} }public static class Job extends RingBufferWheel.Task{Overridepublic void run() {System.out.println(test5);} }运行结果test5 test5 test5 如果要周期性执行任务可以在任务执行完成之后再重新加入到时间轮中。详细源码分析地址[https://crossoverjie.top/2019/09/27/algorithm/time%20wheel/]4.2、应用时间轮的应用还是非常广的例如在 Disruptor 项目中就运用到了 RingBuffer还有Netty中的HashedWheelTimer工具原理也差不多等等有兴趣的同学可以阅读一下官方对应的源码五、小结本文主要围绕单体应用中的定时任务原理进行分析可能也有理解不对的地方欢迎批评指出六、参考1、简书 - 谈谈定时任务解决方案原理2、crossoverJies Blog - 延时消息之时间轮 往期推荐 史上最全的延迟任务实现方式汇总附代码强烈推荐2020-04-14 定时任务最简单的3种实现方法超好用2020-08-18 文件写入的6种方法这种方法性能最好2020-12-22 关注我每天陪你进步一点点