[Java] Spring boot定时任务的原理及动态创建详解

@target(elementtype.type)
@retention(retentionpolicy.runtime)
@import(schedulingconfiguration.class)
@documented
public @interface enablescheduling {
 
}
 
@configuration
@role(beandefinition.role_infrastructure)
public class schedulingconfiguration {
 
 @bean(name = taskmanagementconfigutils.scheduled_annotation_processor_bean_name)
 @role(beandefinition.role_infrastructure)
 public scheduledannotationbeanpostprocessor scheduledannotationprocessor() {
 return new scheduledannotationbeanpostprocessor();
 }
 
}

@override
public object postprocessafterinitialization(object bean, string beanname) {
 if (bean instanceof aopinfrastructurebean || bean instanceof taskscheduler ||
  bean instanceof scheduledexecutorservice) {
 // ignore aop infrastructure such as scoped proxies.
 return bean;
 }
 
 class<?> targetclass = aopproxyutils.ultimatetargetclass(bean);
 if (!this.nonannotatedclasses.contains(targetclass)) {
 map<method, set<scheduled>> annotatedmethods = methodintrospector.selectmethods(targetclass,
  (methodintrospector.metadatalookup<set<scheduled>>) method -> {
   set<scheduled> scheduledmethods = annotatedelementutils.getmergedrepeatableannotations(
  method, scheduled.class, schedules.class);
   return (!scheduledmethods.isempty() ? scheduledmethods : null);
  });
 if (annotatedmethods.isempty()) {
  this.nonannotatedclasses.add(targetclass);
  if (logger.istraceenabled()) {
  logger.trace("no @scheduled annotations found on bean class: " + targetclass);
  }
 }
 else {
  // non-empty set of methods
  annotatedmethods.foreach((method, scheduledmethods) ->
   scheduledmethods.foreach(scheduled -> processscheduled(scheduled, method, bean)));
  if (logger.istraceenabled()) {
  logger.trace(annotatedmethods.size() + " @scheduled methods processed on bean '" + beanname +
   "': " + annotatedmethods);
  }
 }
 }
 return bean;
}

private void finishregistration() {
 if (this.scheduler != null) {
 this.registrar.setscheduler(this.scheduler);
 }
 
 if (this.beanfactory instanceof listablebeanfactory) {
 map<string, schedulingconfigurer> beans =
  ((listablebeanfactory) this.beanfactory).getbeansoftype(schedulingconfigurer.class);
 list<schedulingconfigurer> configurers = new arraylist<>(beans.values());
 annotationawareordercomparator.sort(configurers);
 for (schedulingconfigurer configurer : configurers) {
  configurer.configuretasks(this.registrar);
 }
 }
 
 if (this.registrar.hastasks() && this.registrar.getscheduler() == null) {
 assert.state(this.beanfactory != null, "beanfactory must be set to find scheduler by type");
 try {
  // search for taskscheduler bean...
  this.registrar.settaskscheduler(resolveschedulerbean(this.beanfactory, taskscheduler.class, false));
 }
 catch (nouniquebeandefinitionexception ex) {
  logger.trace("could not find unique taskscheduler bean", ex);
  try {
  this.registrar.settaskscheduler(resolveschedulerbean(this.beanfactory, taskscheduler.class, true));
  }
  catch (nosuchbeandefinitionexception ex2) {
  if (logger.isinfoenabled()) {
   logger.info("more than one taskscheduler bean exists within the context, and " +
  "none is named 'taskscheduler'. mark one of them as primary or name it 'taskscheduler' " +
  "(possibly as an alias); or implement the schedulingconfigurer interface and call " +
  "scheduledtaskregistrar#setscheduler explicitly within the configuretasks() callback: " +
  ex.getbeannamesfound());
  }
  }
 }
 catch (nosuchbeandefinitionexception ex) {
  logger.trace("could not find default taskscheduler bean", ex);
  // search for scheduledexecutorservice bean next...
  try {
  this.registrar.setscheduler(resolveschedulerbean(this.beanfactory, scheduledexecutorservice.class, false));
  }
  catch (nouniquebeandefinitionexception ex2) {
  logger.trace("could not find unique scheduledexecutorservice bean", ex2);
  try {
   this.registrar.setscheduler(resolveschedulerbean(this.beanfactory, scheduledexecutorservice.class, true));
  }
  catch (nosuchbeandefinitionexception ex3) {
   if (logger.isinfoenabled()) {
   logger.info("more than one scheduledexecutorservice bean exists within the context, and " +
  "none is named 'taskscheduler'. mark one of them as primary or name it 'taskscheduler' " +
  "(possibly as an alias); or implement the schedulingconfigurer interface and call " +
  "scheduledtaskregistrar#setscheduler explicitly within the configuretasks() callback: " +
  ex2.getbeannamesfound());
   }
  }
  }
  catch (nosuchbeandefinitionexception ex2) {
  logger.trace("could not find default scheduledexecutorservice bean", ex2);
  // giving up -> falling back to default scheduler within the registrar...
  logger.info("no taskscheduler/scheduledexecutorservice bean found for scheduled processing");
  }
 }
 }
 
 this.registrar.afterpropertiesset();
}

protected void scheduletasks() {
 if (this.taskscheduler == null) {
 this.localexecutor = executors.newsinglethreadscheduledexecutor();
 this.taskscheduler = new concurrenttaskscheduler(this.localexecutor);
 }
 if (this.triggertasks != null) {
 for (triggertask task : this.triggertasks) {
  addscheduledtask(scheduletriggertask(task));
 }
 }
 if (this.crontasks != null) {
 for (crontask task : this.crontasks) {
  addscheduledtask(schedulecrontask(task));
 }
 }
 if (this.fixedratetasks != null) {
 for (intervaltask task : this.fixedratetasks) {
  addscheduledtask(schedulefixedratetask(task));
 }
 }
 if (this.fixeddelaytasks != null) {
 for (intervaltask task : this.fixeddelaytasks) {
  addscheduledtask(schedulefixeddelaytask(task));
 }
 }
}

scheduledtaskregistrar.java
@nullable
public scheduledtask schedulecrontask(crontask task) {
 scheduledtask scheduledtask = this.unresolvedtasks.remove(task);
 boolean newtask = false;
 if (scheduledtask == null) {
 scheduledtask = new scheduledtask(task);
 newtask = true;
 }
 if (this.taskscheduler != null) {
 scheduledtask.future = this.taskscheduler.schedule(task.getrunnable(), task.gettrigger());
 }
 else {
 addcrontask(task);
 this.unresolvedtasks.put(task, scheduledtask);
 }
 return (newtask ? scheduledtask : null);
}
 
concurrenttaskscheduler.java
@override
@nullable
public scheduledfuture<?> schedule(runnable task, trigger trigger) {
 try {
 if (this.enterpriseconcurrentscheduler) {
  return new enterpriseconcurrenttriggerscheduler().schedule(decoratetask(task, true), trigger);
 }
 else {
  errorhandler errorhandler =
   (this.errorhandler != null ? this.errorhandler : taskutils.getdefaulterrorhandler(true));
  return new reschedulingrunnable(task, trigger, this.scheduledexecutor, errorhandler).schedule();
 }
 }
 catch (rejectedexecutionexception ex) {
 throw new taskrejectedexception("executor [" + this.scheduledexecutor + "] did not accept task: " + task, ex);
 }
}
 
reschedulingrunnable.java
@nullable
public scheduledfuture<?> schedule() {
 synchronized (this.triggercontextmonitor) {
 this.scheduledexecutiontime = this.trigger.nextexecutiontime(this.triggercontext);
 if (this.scheduledexecutiontime == null) {
  return null;
 }
 long initialdelay = this.scheduledexecutiontime.gettime() - system.currenttimemillis();
 this.currentfuture = this.executor.schedule(this, initialdelay, timeunit.milliseconds);
 return this;
 }
}
 
private scheduledfuture<?> obtaincurrentfuture() {
 assert.state(this.currentfuture != null, "no scheduled future");
 return this.currentfuture;
}
 
@override
public void run() {
 date actualexecutiontime = new date();
 super.run();
 date completiontime = new date();
 synchronized (this.triggercontextmonitor) {
 assert.state(this.scheduledexecutiontime != null, "no scheduled execution");
 this.triggercontext.update(this.scheduledexecutiontime, actualexecutiontime, completiontime);
 if (!obtaincurrentfuture().iscancelled()) {
  schedule();
 }
 }
}

scheduledthreadpoolexecutor.java
public scheduledfuture<?> schedule(runnable command,
   long delay,
   timeunit unit) {
 if (command == null || unit == null)
 throw new nullpointerexception();
 runnablescheduledfuture<?> t = decoratetask(command,
 new scheduledfuturetask<void>(command, null,
   triggertime(delay, unit)));
 delayedexecute(t);
 return t;
}
 
 
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();
 }
}

/**
 * creates with given first task and thread from threadfactory.
 * @param firsttask the first task (null if none)
 */
worker(runnable firsttask) {
 setstate(-1); // inhibit interrupts until runworker
 this.firsttask = firsttask;
 this.thread = getthreadfactory().newthread(this);
}
 
/** delegates main run loop to outer runworker */
public void run() {
 runworker(this);
}

@configuration
@enablescheduling
@role(beandefinition.role_infrastructure)
public class taskconfiguration {
 
 @bean(name = scheduledannotationbeanpostprocessor.default_task_scheduler_bean_name)
 @role(beandefinition.role_infrastructure)
 public scheduledexecutorservice scheduledannotationprocessor() {
 return executors.newscheduledthreadpool(5, new defaultthreadfactory());
 }
 
 private static class defaultthreadfactory implements threadfactory {
 private static final atomicinteger poolnumber = new atomicinteger(1);
 private final threadgroup group;
 private final atomicinteger threadnumber = new atomicinteger(1);
 private final string nameprefix;
 
 defaultthreadfactory() {
  securitymanager s = system.getsecuritymanager();
  group = (s != null) ? s.getthreadgroup() :
   thread.currentthread().getthreadgroup();
  nameprefix = "pool-" +
   poolnumber.getandincrement() +
   "-schedule-";
 }
 
 @override
 public thread newthread(runnable r) {
  thread t = new thread(group, r,
   nameprefix + threadnumber.getandincrement(),
   0);
  if (t.isdaemon()) {
  t.setdaemon(false);
  }
  if (t.getpriority() != thread.norm_priority) {
  t.setpriority(thread.norm_priority);
  }
  return t;
 }
 }
}

@configuration
public class dynamictask implements schedulingconfigurer {
 private static logger logger = loggerfactory.getlogger(dynamictask.class);
 
 private static final executorservice es = new threadpoolexecutor(10, 20,
   0l, timeunit.milliseconds,
   new linkedblockingqueue<>(10),
   new dynamictaskconsumethreadfactory());
 
 
 private volatile scheduledtaskregistrar registrar;
 private final concurrenthashmap<string, scheduledfuture<?>> scheduledfutures = new concurrenthashmap<>();
 private final concurrenthashmap<string, crontask> crontasks = new concurrenthashmap<>();
 
 private volatile list<taskconstant> taskconstants = lists.newarraylist();
 
 @override
 public void configuretasks(scheduledtaskregistrar registrar) {
  this.registrar = registrar;
  this.registrar.addtriggertask(() -> {
   if (!collectionutils.isempty(taskconstants)) {
    logger.info("检测动态定时任务列表...");
    list<timingtask> tts = new arraylist<>();
    taskconstants
    .foreach(taskconstant -> {
     timingtask tt = new timingtask();
     tt.setexpression(taskconstant.getcron());
     tt.settaskid("dynamic-task-" + taskconstant.gettaskid());
     tts.add(tt);
    });
    this.refreshtasks(tts);
   }
  }
  , triggercontext -> new periodictrigger(5l, timeunit.seconds).nextexecutiontime(triggercontext));
 }
 
 
 public list<taskconstant> gettaskconstants() {
  return taskconstants;
 }
 
 private void refreshtasks(list<timingtask> tasks) {
  //取消已经删除的策略任务
  set<string> taskids = scheduledfutures.keyset();
  for (string taskid : taskids) {
   if (!exists(tasks, taskid)) {
  scheduledfutures.get(taskid).cancel(false);
   }
  }
  for (timingtask tt : tasks) {
   string expression = tt.getexpression();
   if (stringutils.isblank(expression) || !cronsequencegenerator.isvalidexpression(expression)) {
  logger.error("定时任务dynamictask cron表达式不合法: " + expression);
  continue;
   }
   //如果配置一致，则不需要重新创建定时任务
   if (scheduledfutures.containskey(tt.gettaskid())
   && crontasks.get(tt.gettaskid()).getexpression().equals(expression)) {
  continue;
   }
   //如果策略执行时间发生了变化，则取消当前策略的任务
   if (scheduledfutures.containskey(tt.gettaskid())) {
  scheduledfutures.remove(tt.gettaskid()).cancel(false);
  crontasks.remove(tt.gettaskid());
   }
   crontask task = new crontask(tt, expression);
   scheduledfuture<?> future = registrar.getscheduler().schedule(task.getrunnable(), task.gettrigger());
   crontasks.put(tt.gettaskid(), task);
   scheduledfutures.put(tt.gettaskid(), future);
  }
 }
 
 private boolean exists(list<timingtask> tasks, string taskid) {
  for (timingtask task : tasks) {
   if (task.gettaskid().equals(taskid)) {
  return true;
   }
  }
  return false;
 }
 
 @predestroy
 public void destroy() {
  this.registrar.destroy();
 }
 
 public static class taskconstant {
  private string cron;
  private string taskid;
 
  public string getcron() {
   return cron;
  }
 
  public void setcron(string cron) {
   this.cron = cron;
  }
 
  public string gettaskid() {
   return taskid;
  }
 
  public void settaskid(string taskid) {
   this.taskid = taskid;
  }
 }
 
 private class timingtask implements runnable {
  private string expression;
 
  private string taskid;
 
  public string gettaskid() {
   return taskid;
  }
 
  public void settaskid(string taskid) {
   this.taskid = taskid;
  }
 
  @override
  public void run() {
   //设置队列大小10
   logger.error("当前crontask: " + this);
   dynamicblockingqueue queue = new dynamicblockingqueue(3);
   es.submit(() -> {
  while (!queue.isdone() || !queue.isempty()) {
   try {
    string content = queue.poll(500, timeunit.milliseconds);
    if (stringutils.isblank(content)) {
     return;
    }
    logger.info("dynamicblockingqueue 消费：" + content);
    timeunit.milliseconds.sleep(500);
   } catch (interruptedexception e) {
    e.printstacktrace();
   }
  }
   });
 
   //队列放入数据
   for (int i = 0; i < 5; ++i) {
  try {
   queue.put(string.valueof(i));
   logger.info("dynamicblockingqueue 生产：" + i);
  } catch (interruptedexception e) {
   e.printstacktrace();
  }
   }
   queue.setdone(true);
  }
 
  public string getexpression() {
   return expression;
  }
 
  public void setexpression(string expression) {
   this.expression = expression;
  }
 
  @override
  public string tostring() {
   return reflectiontostringbuilder.tostring(this
   , tostringstyle.json_style
   , false
   , false
   , timingtask.class);
  }
 
 }
 
 /**
  * 队列消费线程工厂类
  */
 private static class dynamictaskconsumethreadfactory implements threadfactory {
  private static final atomicinteger poolnumber = new atomicinteger(1);
  private final threadgroup group;
  private final atomicinteger threadnumber = new atomicinteger(1);
  private final string nameprefix;
 
  dynamictaskconsumethreadfactory() {
   securitymanager s = system.getsecuritymanager();
   group = (s != null) ? s.getthreadgroup() :
   thread.currentthread().getthreadgroup();
   nameprefix = "pool-" +
   poolnumber.getandincrement() +
   "-dynamic-task-";
  }
 
  @override
  public thread newthread(runnable r) {
   thread t = new thread(group, r,
   nameprefix + threadnumber.getandincrement(),
   0);
   if (t.isdaemon()) {
  t.setdaemon(false);
   }
   if (t.getpriority() != thread.norm_priority) {
  t.setpriority(thread.norm_priority);
   }
   return t;
  }
 }
 
 private static class dynamicblockingqueue extends linkedblockingqueue<string> {
  dynamicblockingqueue(int capacity) {
   super(capacity);
  }
 
 
  private volatile boolean done = false;
 
  public boolean isdone() {
   return done;
  }
 
  public void setdone(boolean done) {
   this.done = done;
  }
 }
}

crontask task = new crontask(tt, expression);
scheduledfuture<?> future = registrar.getscheduler().schedule(task.getrunnable(), task.gettrigger());
crontasks.put(tt.gettaskid(), task);
scheduledfutures.put(tt.gettaskid(), future);

@runwith(springrunner.class)
@springboottest
public class dynamictasktest {
 
 @autowired
 private dynamictask dynamictask;
 
 @test
 public void test() throws interruptedexception {
  list<dynamictask.taskconstant> taskconstans = dynamictask.gettaskconstants();
  dynamictask.taskconstant taskconstant = new dynamictask.taskconstant();
  taskconstant.setcron("0/5 * * * * ?");
  taskconstant.settaskid("test1");
  taskconstans.add(taskconstant);
 
 
  dynamictask.taskconstant taskconstant1 = new dynamictask.taskconstant();
  taskconstant1.setcron("0/5 * * * * ?");
  taskconstant1.settaskid("test2");
  taskconstans.add(taskconstant1);
 
  dynamictask.taskconstant taskconstant2 = new dynamictask.taskconstant();
  taskconstant2.setcron("0/5 * * * * ?");
  taskconstant2.settaskid("test3");
  taskconstans.add(taskconstant2);
 
  timeunit.seconds.sleep(40);
  //移除并添加新的配置
  taskconstans.remove(taskconstans.size() - 1);
  dynamictask.taskconstant taskconstant3 = new dynamictask.taskconstant();
  taskconstant3.setcron("0/5 * * * * ?");
  taskconstant3.settaskid("test4");
  taskconstans.add(taskconstant3);
//
  timeunit.minutes.sleep(50);
 }
}