Spring5源码解析之Spring中的异步和计划任务

package org.springframework.core.task;
import java.util.concurrent.executor;
/**
 * simple task executor interface that abstracts the execution
 * of a {@link runnable}.
 *
 * <p>implementations can use all sorts of different execution strategies,
 * such as: synchronous, asynchronous, using a thread pool, and more.
 *
 * <p>equivalent to jdk 1.5's {@link java.util.concurrent.executor}
 * interface; extending it now in spring 3.0, so that clients may declare
 * a dependency on an executor and receive any taskexecutor implementation.
 * this interface remains separate from the standard executor interface
 * mainly for backwards compatibility with jdk 1.4 in spring 2.x.
 *
 * @author juergen hoeller
 * @since 2.0
 * @see java.util.concurrent.executor
 */
@functionalinterface
public interface taskexecutor extends executor {
 /**
 * execute the given {@code task}.
 * <p>the call might return immediately if the implementation uses
 * an asynchronous execution strategy, or might block in the case
 * of synchronous execution.
 * @param task the {@code runnable} to execute (never {@code null})
 * @throws taskrejectedexception if the given task was not accepted
 */
 @override
 void execute(runnable task);
}

public interface taskscheduler {
 /**
 * schedule the given {@link runnable}, invoking it whenever the trigger
 * indicates a next execution time.
 * <p>execution will end once the scheduler shuts down or the returned
 * {@link scheduledfuture} gets cancelled.
 * @param task the runnable to execute whenever the trigger fires
 * @param trigger an implementation of the {@link trigger} interface,
 * e.g. a {@link org.springframework.scheduling.support.crontrigger} object
 * wrapping a cron expression
 * @return a {@link scheduledfuture} representing pending completion of the task,
 * or {@code null} if the given trigger object never fires (i.e. returns
 * {@code null} from {@link trigger#nextexecutiontime})
 * @throws org.springframework.core.task.taskrejectedexception if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * @see org.springframework.scheduling.support.crontrigger
 */
 @nullable
 scheduledfuture<?> schedule(runnable task, trigger trigger);
 
 /**
 * schedule the given {@link runnable}, invoking it at the specified execution time.
 * <p>execution will end once the scheduler shuts down or the returned
 * {@link scheduledfuture} gets cancelled.
 * @param task the runnable to execute whenever the trigger fires
 * @param starttime the desired execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @return a {@link scheduledfuture} representing pending completion of the task
 * @throws org.springframework.core.task.taskrejectedexception if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * 使用了默认实现，值得我们学习使用的，java9中同样可以有私有实现的，从这里我们可以做到我只通过  * 一个接口你来实现，我把其他相应的功能默认实现下，最后调用你自定义实现的接口即可，使接口功能更  * 加一目了然
 * @since 5.0
 * @see #schedule(runnable, date)
 */
 default scheduledfuture<?> schedule(runnable task, instant starttime) {
 return schedule(task, date.from(starttime));
 }
 /**
 * schedule the given {@link runnable}, invoking it at the specified execution time.
 * <p>execution will end once the scheduler shuts down or the returned
 * {@link scheduledfuture} gets cancelled.
 * @param task the runnable to execute whenever the trigger fires
 * @param starttime the desired execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @return a {@link scheduledfuture} representing pending completion of the task
 * @throws org.springframework.core.task.taskrejectedexception if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 */
 scheduledfuture<?> schedule(runnable task, date starttime);
...
/**
 * schedule the given {@link runnable}, invoking it at the specified execution time
 * and subsequently with the given period.
 * <p>execution will end once the scheduler shuts down or the returned
 * {@link scheduledfuture} gets cancelled.
 * @param task the runnable to execute whenever the trigger fires
 * @param starttime the desired first execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @param period the interval between successive executions of the task
 * @return a {@link scheduledfuture} representing pending completion of the task
 * @throws org.springframework.core.task.taskrejectedexception if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 * @since 5.0
 * @see #scheduleatfixedrate(runnable, date, long)
 */
 default scheduledfuture<?> scheduleatfixedrate(runnable task, instant starttime, duration period) {
 return scheduleatfixedrate(task, date.from(starttime), period.tomillis());
 }
 /**
 * schedule the given {@link runnable}, invoking it at the specified execution time
 * and subsequently with the given period.
 * <p>execution will end once the scheduler shuts down or the returned
 * {@link scheduledfuture} gets cancelled.
 * @param task the runnable to execute whenever the trigger fires
 * @param starttime the desired first execution time for the task
 * (if this is in the past, the task will be executed immediately, i.e. as soon as possible)
 * @param period the interval between successive executions of the task (in milliseconds)
 * @return a {@link scheduledfuture} representing pending completion of the task
 * @throws org.springframework.core.task.taskrejectedexception if the given task was not accepted
 * for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
 */
 scheduledfuture<?> scheduleatfixedrate(runnable task, date starttime, long period);
...
}

public scheduledfuture<?> schedule(runnable task, final trigger trigger) {
 managedscheduledexecutorservice executor = (managedscheduledexecutorservice) scheduledexecutor;
 return executor.schedule(task, new javax.enterprise.concurrent.trigger() {
  @override
  public date getnextruntime(lastexecution le, date taskscheduledtime) {
   return trigger.nextexecutiontime(le != null ?
    new simpletriggercontext(le.getscheduledstart(), le.getrunstart(), le.getrunend()) :
    new simpletriggercontext());
  }
  @override
  public boolean skiprun(lastexecution lastexecution, date scheduledruntime) {
   return false;
  }
 });
}

/**
 * simple data holder implementation of the {@link triggercontext} interface.
 *
 * @author juergen hoeller
 * @since 3.0
 */
public class simpletriggercontext implements triggercontext {
 @nullable
 private volatile date lastscheduledexecutiontime;
 @nullable
 private volatile date lastactualexecutiontime;
 @nullable
 private volatile date lastcompletiontime;
...
 /**
 * create a simpletriggercontext with the given time values.
 * @param lastscheduledexecutiontime last <i>scheduled</i> execution time
 * @param lastactualexecutiontime last <i>actual</i> execution time
 * @param lastcompletiontime last completion time
 */
 public simpletriggercontext(date lastscheduledexecutiontime, date lastactualexecutiontime, date lastcompletiontime) {
 this.lastscheduledexecutiontime = lastscheduledexecutiontime;
 this.lastactualexecutiontime = lastactualexecutiontime;
 this.lastcompletiontime = lastcompletiontime;
 }
 ...
}

/**
 * adapter that takes a jdk {@code java.util.concurrent.executor} and
 * exposes a spring {@link org.springframework.core.task.taskexecutor} for it.
 * also detects an extended {@code java.util.concurrent.executorservice 从此解释上面的说明}, adapting
 * the {@link org.springframework.core.task.asynctaskexecutor} interface accordingly.
 *
 * @author juergen hoeller
 * @since 3.0
 * @see java.util.concurrent.executor
 * @see java.util.concurrent.executorservice 
 * @see java.util.concurrent.executors
 */
public class taskexecutoradapter implements asynclistenabletaskexecutor {
 private final executor concurrentexecutor;
 @nullable
 private taskdecorator taskdecorator;
 ...
  /**
 * create a new taskexecutoradapter,
 * using the given jdk concurrent executor.
 * @param concurrentexecutor the jdk concurrent executor to delegate to
 */
 public taskexecutoradapter(executor concurrentexecutor) {
 assert.notnull(concurrentexecutor, "executor must not be null");
 this.concurrentexecutor = concurrentexecutor;
 }
  /**
 * delegates to the specified jdk concurrent executor.
 * @see java.util.concurrent.executor#execute(runnable)
 */
 @override
 public void execute(runnable task) {
 try {
  doexecute(this.concurrentexecutor, this.taskdecorator, task);
 }
 catch (rejectedexecutionexception ex) {
  throw new taskrejectedexception(
   "executor [" + this.concurrentexecutor + "] did not accept task: " + task, ex);
 }
 }
 @override
 public void execute(runnable task, long starttimeout) {
 execute(task);
 }
 @override
 public future<?> submit(runnable task) {
 try {
  if (this.taskdecorator == null && this.concurrentexecutor instanceof executorservice) {
  return ((executorservice) this.concurrentexecutor).submit(task);
  }
  else {
  futuretask<object> future = new futuretask<>(task, null);
  doexecute(this.concurrentexecutor, this.taskdecorator, future);
  return future;
  }
 }
 catch (rejectedexecutionexception ex) {
  throw new taskrejectedexception(
   "executor [" + this.concurrentexecutor + "] did not accept task: " + task, ex);
 }
 }
 ...
}

<task:scheduler id="taskscheduler"/>
<task:executor id="taskexecutor" pool-size="2" />
<task:annotation-driven executor="taskexecutor" scheduler="taskscheduler"/>
<context:component-scan base-package="com.migo.async"/>

protected void processscheduled(scheduled scheduled, method method, object bean) {
 try {
  assert.istrue(method.getparametercount() == 0,
   "only no-arg methods may be annotated with @scheduled");
 /**
 *  之前的版本中直接把返回值非空的给拒掉了，在spring 4.3 spring5 的版本中就没那么严格了
   * assert.istrue(void.class.equals(method.getreturntype()),
   *        "only void-returning methods may be annotated with @scheduled");
   **/        
// ...

/**
 * 注释很重要
 * an annotation that marks a method to be scheduled. exactly one of
 * the {@link #cron()}, {@link #fixeddelay()}, or {@link #fixedrate()}
 * attributes must be specified.
 *
 * <p>the annotated method must expect no arguments. it will typically have
 * a {@code void} return type; if not, the returned value will be ignored
 * when called through the scheduler.
 *
 * <p>processing of {@code @scheduled} annotations is performed by
 * registering a {@link scheduledannotationbeanpostprocessor}. this can be
 * done manually or, more conveniently, through the {@code <task:annotation-driven/>}
 * element or @{@link enablescheduling} annotation.
 *
 * <p>this annotation may be used as a <em>meta-annotation</em> to create custom
 * <em>composed annotations</em> with attribute overrides.
 *
 * @author mark fisher
 * @author dave syer
 * @author chris beams
 * @since 3.0
 * @see enablescheduling
 * @see scheduledannotationbeanpostprocessor
 * @see schedules
 */
@target({elementtype.method, elementtype.annotation_type})
@retention(retentionpolicy.runtime)
@documented
@repeatable(schedules.class)
public @interface scheduled {
...
}

@runwith(springjunit4classrunner.class)
@contextconfiguration(locations={"classpath:applicationcontext-test.xml"})
@webappconfiguration
public class taskexecutorstest {
 
 @test
 public void simpeasync() throws interruptedexception {
  /**
   * simpleasynctaskexecutor creates new thread for every task and executes it asynchronously. the threads aren't reused as in 
   * native java's thread pools.
   * 
   * the number of concurrently executed threads can be specified through concurrencylimit bean property 
   * (concurrencylimit xml attribute). here it's more simple to invoke setconcurrencylimit method. 
   * here the tasks will be executed by 2 simultaneous threads. without specifying this value,
   * the number of executed threads will be indefinite.
   * 
   * you can observe that only 2 tasks are executed at a given time - even if 3 are submitted to execution (lines 40-42).
   **/
  simpleasynctaskexecutor executor = new simpleasynctaskexecutor("thread_name_prefix_____");
  executor.setconcurrencylimit(2);
  executor.execute(new simpletask("simpleasynctask-1", counters.simpleasynctask, 1000));
  executor.execute(new simpletask("simpleasynctask-2", counters.simpleasynctask, 1000));
 
  thread.sleep(1050);
  asserttrue("2 threads should be terminated, but "+counters.simpleasynctask.getnb()+" were instead", counters.simpleasynctask.getnb() == 2);
 
  executor.execute(new simpletask("simpleasynctask-3", counters.simpleasynctask, 1000));
  executor.execute(new simpletask("simpleasynctask-4", counters.simpleasynctask, 1000));
  executor.execute(new simpletask("simpleasynctask-5", counters.simpleasynctask, 2000));
   
  thread.sleep(1050);
  asserttrue("4 threads should be terminated, but "+counters.simpleasynctask.getnb()+" were instead", counters.simpleasynctask.getnb() == 4);
  executor.execute(new simpletask("simpleasynctask-6", counters.simpleasynctask, 1000));
 
  thread.sleep(1050);
  asserttrue("6 threads should be terminated, but "+counters.simpleasynctask.getnb()+" were instead", 
   counters.simpleasynctask.getnb() == 6);
 }
  
 @test
 public void synctasktest() {
  /**
   * synctask works almost as java's countdownlatch. in fact, this executor is synchronous with the calling thread. in our case,
   * synctaskexecutor tasks will be synchronous with junit thread. it means that the testing thread will sleep 5 
   * seconds after executing the third task ('synctask-3'). to prove that, we check if the total execution time is ~5 seconds.
   **/
  long start = system.currenttimemillis();
  synctaskexecutor executor = new synctaskexecutor();
  executor.execute(new simpletask("synctask-1", counters.synctask, 0));
  executor.execute(new simpletask("synctask-2", counters.synctask, 0));
  executor.execute(new simpletask("synctask-3", counters.synctask, 0));
  executor.execute(new simpletask("synctask-4", counters.synctask, 5000));
  executor.execute(new simpletask("synctask-5", counters.synctask, 0));
  long end = system.currenttimemillis();
  int exectime = math.round((end-start)/1000);
  asserttrue("execution time should be 5 seconds but was "+exectime+" seconds", exectime == 5); 
 }
  
 @test
 public void threadpooltest() throws interruptedexception {
  /**
   * this executor can be used to expose java's native threadpoolexecutor as spring bean, with the 
   * possibility to set core pool size, max pool size and queue capacity through bean properties.
   * 
   * it works exactly as threadpoolexecutor from java.util.concurrent package. it means that our pool starts 
   * with 2 threads (core pool size) and can be growth until 3 (max pool size).
   * in additionally, 1 task can be stored in the queue. this task will be treated 
   * as soon as one from 3 threads ends to execute provided task. in our case, we try to execute 5 tasks
   * in 3 places pool and 1 place queue. so the 5th task should be rejected and taskrejectedexception should be thrown.
   **/
  threadpooltaskexecutor executor = new threadpooltaskexecutor();
  executor.setcorepoolsize(2);
  executor.setmaxpoolsize(3);
  executor.setqueuecapacity(1);
  executor.initialize();
 
  executor.execute(new simpletask("threadpooltask-1", counters.threadpool, 1000));
  executor.execute(new simpletask("threadpooltask-2", counters.threadpool, 1000));
  executor.execute(new simpletask("threadpooltask-3", counters.threadpool, 1000));
  executor.execute(new simpletask("threadpooltask-4", counters.threadpool, 1000));
  boolean wastre = false;
  try {
   executor.execute(new simpletask("threadpooltask-5", counters.threadpool, 1000));
  } catch (taskrejectedexception tre) {
   wastre = true;
  }
  asserttrue("the last task should throw a taskrejectedexception but it wasn't", wastre);
 
  thread.sleep(3000);
 
  asserttrue("4 tasks should be terminated, but "+counters.threadpool.getnb()+" were instead", 
   counters.threadpool.getnb()==4);
 }
 
}
 
class simpletask implements runnable {
 private string name;
 private counters counter;
 private int sleeptime;
  
 public simpletask(string name, counters counter, int sleeptime) {
  this.name = name;
  this.counter = counter;
  this.sleeptime = sleeptime;
 }
  
 @override
 public void run() {
  try {
   thread.sleep(this.sleeptime);
  } catch (interruptedexception e) {
   e.printstacktrace();
  }
  this.counter.increment();
  system.out.println("running task '"+this.name+"' in thread "+thread.currentthread().getname());
 }
  
 @override
 public string tostring() {
     return "task {"+this.name+"}";
 }
}
 
enum counters {
     
 simpleasynctask(0),
 synctask(0),
 threadpool(0);
  
 private int nb;
  
 public int getnb() {
  return this.nb;
 }
  
 public synchronized void increment() {
  this.nb++;
 }
 
 private counters(int n) {
  this.nb = n;
 }
}

running task 'simpleasynctask-1' in thread thread_name_prefix_____1
running task 'simpleasynctask-2' in thread thread_name_prefix_____2
running task 'simpleasynctask-3' in thread thread_name_prefix_____3
running task 'simpleasynctask-4' in thread thread_name_prefix_____4
running task 'simpleasynctask-5' in thread thread_name_prefix_____5
running task 'simpleasynctask-6' in thread thread_name_prefix_____6
running task 'synctask-1' in thread main
running task 'synctask-2' in thread main
running task 'synctask-3' in thread main
running task 'synctask-4' in thread main
running task 'synctask-5' in thread main
running task 'threadpooltask-2' in thread threadpooltaskexecutor-2
running task 'threadpooltask-1' in thread threadpooltaskexecutor-1
running task 'threadpooltask-4' in thread threadpooltaskexecutor-3
running task 'threadpooltask-3' in thread threadpooltaskexecutor-2

@runwith(springjunit4classrunner.class)
@contextconfiguration(locations={"classpath:applicationcontext-test.xml"})
@webappconfiguration
public class annotationtest {
 
 @autowired
 private genericapplicationcontext context;
     
 @test
 public void testscheduled() throws interruptedexception {
 
   system.out.println("start sleeping");
   thread.sleep(6000);
   system.out.println("wake up !");
 
   testscheduledtask scheduledtask = (testscheduledtask) context.getbean("testscheduledtask");
    /**
    * test fixed delay. it's executed every 6 seconds. the first execution is registered after application's context start. 
    **/
   asserttrue("scheduled task should be executed 2 times (1 before sleep in this method, 1 after the sleep), but was "+scheduledtask.getfixeddelaycounter(), 
    scheduledtask.getfixeddelaycounter() == 2);
    
    /**
    * test fixed rate. it's executed every 6 seconds. the first execution is registered after application's context start. 
    * unlike fixed delay, a fixed rate configuration executes one task with specified time. for example, it will execute on 
    * 6 seconds delayed task at 10:30:30, 10:30:36, 10:30:42 and so on - even if the task 10:30:30 taken 30 seconds to 
    * be terminated. in teh case of fixed delay, if the first task takes 30 seconds, the next will be executed 6 seconds 
    * after the first one, so the execution flow will be: 10:30:30, 10:31:06, 10:31:12.
    **/
   asserttrue("scheduled task should be executed 2 times (1 before sleep in this method, 1 after the sleep), but was "+scheduledtask.getfixedratecounter(), 
    scheduledtask.getfixedratecounter() == 2);
    /**
    * test fixed rate with initial delay attribute. the initialdelay attribute is set to 6 seconds. it causes that 
    * scheduled method is executed 6 seconds after application's context start. in our case, it should be executed 
    * only once because of previous thread.sleep(6000) invocation.
    **/
   asserttrue("scheduled task should be executed 1 time (0 before sleep in this method, 1 after the sleep), but was "+scheduledtask.getinitialdelaycounter(), scheduledtask.getinitialdelaycounter() == 1);
    /**
    * test cron scheduled task. cron is scheduled to be executed every 6 seconds. it's executed only once, 
    * so we can deduce that it's not invoked directly before applications 
    * context start, but only after configured time (6 seconds in our case).
    **/
   asserttrue("scheduled task should be executed 1 time (0 before sleep in this method, 1 after the sleep), but was "+scheduledtask.getcroncounter(), scheduledtask.getcroncounter() == 1);
 }
     
 @test
 public void testasyc() throws interruptedexception {
    /**
    * to test @async annotation, we can create a bean in-the-fly. asynccounter bean is a
    * simple counter which value should be equals to 2 at the end of the test. a supplementary test
    * concerns threads which execute both of asynccounter methods: one which 
    * isn't annotated with @async and another one which is annotated with it. for the first one, invoking
    * thread should have the same name as the main thread. for annotated method, it can't be executed in 
    * the main thread. it must be executed asynchrounously in a new thread.
    **/
   context.registerbeandefinition("asynccounter", new rootbeandefinition(asynccounter.class));
    
   string currentname = thread.currentthread().getname();
   asynccounter asynccounter = context.getbean("asynccounter", asynccounter.class);
   asynccounter.incrementnormal();
   asserttrue("thread executing normal increment should be the same as junit thread but it wasn't (expected '"+currentname+"', got '"+asynccounter.getnormalthreadname()+"')",
           asynccounter.getnormalthreadname().equals(currentname));
   asynccounter.incrementasync();
   // sleep 50ms and give some time to asynccounter to update asyncthreadname value
   thread.sleep(50);
 
   assertfalse("thread executing @async increment shouldn't be the same as junit thread but it wasn (junit thread '"+currentname+"', @async thread '"+asynccounter.getasyncthreadname()+"')",
           asynccounter.getasyncthreadname().equals(currentname));
   system.out.println("main thread execution's name: "+currentname);
   system.out.println("asynccounter normal increment thread execution's name: "+asynccounter.getnormalthreadname());
   system.out.println("asynccounter @async increment thread execution's name: "+asynccounter.getasyncthreadname());
   asserttrue("counter should be 2, but was "+asynccounter.getcounter(), asynccounter.getcounter()==2);
 }
 
}
 
class asynccounter {
     
 private int counter = 0;
 private string normalthreadname;
 private string asyncthreadname;
  
 public void incrementnormal() {
  normalthreadname = thread.currentthread().getname();
  this.counter++;
 }
  
 @async
 public void incrementasync() {
  asyncthreadname = thread.currentthread().getname();
  this.counter++;
 }
  
 public string getasyncthreadname() {
  return asyncthreadname;
 }
  
 public string getnormalthreadname() {
  return normalthreadname;
 }
  
 public int getcounter() {
  return this.counter;
 }
     
}

<!-- imported configuration file first -->
<!-- activates various annotations to be detected in bean classes -->
<context:annotation-config />
 
<!-- scans the classpath for annotated components that will be auto-registered as spring beans.
 for example @controller and @service. make sure to set the correct base-package-->
<context:component-scan base-package="com.migo.test.schedulers" />
 
<task:scheduler id="taskscheduler"/>
<task:executor id="taskexecutor" pool-size="40" />
<task:annotation-driven executor="taskexecutor" scheduler="taskscheduler"/>

// scheduled methods after, all are executed every 6 seconds (scheduledatfixedrate and scheduledatfixeddelay start to execute at
// application context start, two other methods begin 6 seconds after application's context start)
@component
public class testscheduledtask {
 
 private int fixedratecounter = 0;
 private int fixeddelaycounter = 0;
 private int initialdelaycounter = 0;
 private int croncounter = 0;
 
 @scheduled(fixedrate = 6000)
 public void scheduledatfixedrate() {
  system.out.println("<r> increment at fixed rate");
  fixedratecounter++;
 }
  
 @scheduled(fixeddelay = 6000)
 public void scheduledatfixeddelay() {
  system.out.println("<d> incrementing at fixed delay");
  fixeddelaycounter++;
 }
  
 @scheduled(fixeddelay = 6000, initialdelay = 6000)
 public void scheduledwithinitialdelay() {
  system.out.println("<di> incrementing with initial delay");
  initialdelaycounter++;
 }
  
 @scheduled(cron = "**/6 ** ** ** ** **")
 public void scheduledwithcron() {
  system.out.println("<c> incrementing with cron");
  croncounter++;
      
 }
  
 public int getfixedratecounter() {
  return this.fixedratecounter;
 }
  
 public int getfixeddelaycounter() {
  return this.fixeddelaycounter;
 }
  
 public int getinitialdelaycounter() {
  return this.initialdelaycounter;
 }
  
 public int getcroncounter() {
  return this.croncounter;
 }
     
}

<r> increment at fixed rate
<d> incrementing at fixed delay
start sleeping
<c> incrementing with cron
<di> incrementing with initial delay
<r> increment at fixed rate
<d> incrementing at fixed delay
wake up !
main thread execution's name: main
asynccounter normal increment thread execution's name: main
asynccounter @async increment thread execution's name: taskexecutor-1