Java线程池和Spring异步处理高级篇

/**
     * Creates a new {@code ThreadPoolExecutor}
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

public class ThreadPoolTest {

    public static void main(String[] args) {
        ThreadPoolExecutor pools = createPool();
        int activeCount = -1;
        int queueSize = -1;
        while (true) {
            if (activeCount != pools.getActiveCount() || queueSize != pools.getQueue().size()) {
                System.out.println("活跃的线程的个数：" + pools.getActiveCount());
                System.out.println("队列中线程的个数：" + pools.getQueue().size());
                System.out.println("最大的线程的个数" + pools.getMaximumPoolSize());
                activeCount = pools.getActiveCount();
                queueSize = pools.getQueue().size();
                System.out.println("=========================================");
            }
        }
    }

     // 创建线程池,通过更改线程池的参数方便你更好的理解线程池
     // 其中第六个参数你也可以改成ThreadPoolExecutor默认的：Executors.defaultThreadFactory()
    private static ThreadPoolExecutor createPool() {
        ThreadPoolExecutor pools = new ThreadPoolExecutor(1, 2, 30,
                TimeUnit.SECONDS, new ArrayBlockingQueue(1),
                r -> {
                    Thread t = new Thread(r);
                    return t;
                }, new ThreadPoolExecutor.AbortPolicy());
        System.out.println("The PoolExecutor is create done");
        pools.execute(() -> {
            sleep(100);
        });
        //这个里面就可以写自己的业务
        pools.execute(() -> {
            sleep(10);
        });
        pools.execute(() -> {
            sleep(10);
        });
        return pools;
    }
    private static void sleep(int seconds) {
        try {
            System.out.println("  " + Thread.currentThread().getName() + "   ");
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(AsyncConfigurationSelector.class)
public @interface EnableAsync {
  /**
   * Indicate the 'async' annotation type to be detected at either class
   * or method level.
   * 默认情况下，要开启异步操作，要在相应的方法或者类上加上@Async注解
   */
  Class extends Annotation> annotation() default Annotation.class;

  /**
   * Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
   * to standard Java interface-based proxies.
   *   true表示启用CGLIB代理
   */
  boolean proxyTargetClass() default false;

  /**
   * Indicate the order in which the {@link AsyncAnnotationBeanPostProcessor}
   * should be applied.
   * 直接定义：它的执行顺序（因为可能有多个@EnableXXX）
   */
  int order() default Ordered.LOWEST_PRECEDENCE;

}

@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Async {

  /**
   * A qualifier value for the specified asynchronous operation(s).
   *   这个value值是用来指定执行器的
   */
  String value() default "";
}

@Override
  @Nullable
  // 见名知意就知道这个是获取线程池的方法，
  // 这个厉害了。如果父类返回的defaultExecutor 为null，
  // 那就new一个SimpleAsyncTaskExecutor作为默认的执行器,所以我们上文中
  // 如果没有指定线程池，那么就默认给我们一个默认的：SimpleAsyncTaskExecuto
protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
    Executor defaultExecutor = super.getDefaultExecutor(beanFactory);
    return (defaultExecutor != null ? defaultExecutor : new SimpleAsyncTaskExecutor());
  }

protected Executor getDefaultExecutor(@Nullable BeanFactory beanFactory) {
    if (beanFactory != null) {
      // 这个处理很有意思，它是用用的try catch的技巧去处理的
      try {
        // 如果容器内存在唯一的TaskExecutor（子类），就直接返回了
        return beanFactory.getBean(TaskExecutor.class);
      }
      catch (NoUniqueBeanDefinitionException ex) {
        // 这是出现了多个TaskExecutor类型的话，那就按照名字去拿  `taskExecutor`且是Executor类型
        try {
          return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);
        }
        // 如果再没有找到，也不要报错，而是接下来创建一个默认的处理器
        // 这里输出一个info信息
        catch (NoSuchBeanDefinitionException ex2) {
        }
      }
      catch (NoSuchBeanDefinitionException ex) {
        try {
          return beanFactory.getBean(DEFAULT_TASK_EXECUTOR_BEAN_NAME, Executor.class);
        }
        catch (NoSuchBeanDefinitionException ex2) {
        }
        // 这里还没有获取到，就放弃。用本地默认的executor吧~~~
        // 子类可以去复写此方法，发现为null的话可议给一个默认值~~~~比如`AsyncExecutionInterceptor`默认给的就是`SimpleAsyncTaskExecutor`作为执行器的
        // Giving up -> either using local default executor or none at all...
      }
    }
    return null;
  }

/**
   * Determine the specific executor to use when executing the given method.
   * Should preferably return an {@link AsyncListenableTaskExecutor} implementation.
   * @return the executor to use (or {@code null}, but just if no default executor is available)
   */
  @Nullable
  protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
  // 如果缓存中能够找到该方法对应的执行器，就立马返回了
    AsyncTaskExecutor executor = this.executors.get(method);
    if (executor == null) {
      Executor targetExecutor;
      // 抽象方法：AnnotationAsyncExecutionInterceptor有实现。
      // 就是@Async注解的value值
      String qualifier = getExecutorQualifier(method);
      // 现在知道@Async直接的value值的作用了吧。就是制定执行此方法的执行器的（容器内执行器的Bean的名称）
      // 当然有可能为null。注意此处是支持@Qualified注解标注在类上来区分Bean的
      // 注意：此处targetExecutor仍然可能为null
      // 使用自定义线程池d额时候Async注解的value值最好加上线程池的名称
      if (StringUtils.hasLength(qualifier)) {
        targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);
      }
      else {
        targetExecutor = this.defaultExecutor.get();
      }
      if (targetExecutor == null) {
        return null;
      }
      executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?
          (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));
      this.executors.put(method, executor);
    }
    return executor;
  }

/**
   * Return the qualifier or bean name of the executor to be used when executing the
   * given method, specified via {@link Async#value} at the method or declaring
   * class level. If {@code @Async} is specified at both the method and class level, the
   * method's {@code #value} takes precedence (even if empty string, indicating that
   * the default executor should be used preferentially).
   */
  @Override
  @Nullable
  protected String getExecutorQualifier(Method method) {
    // Maintainer's note: changes made here should also be made in
    // AnnotationAsyncExecutionAspect#getExecutorQualifier
    // 可以见它就是去方法拿到@Async的value值。
    // 这下知道配置这个注解的作用了吧。
    Async async = AnnotatedElementUtils.findMergedAnnotation(method, Async.class);
    if (async == null) {
      async = AnnotatedElementUtils.findMergedAnnotation(method.getDeclaringClass(), Async.class);
    }
    return (async != null ? async.value() : null);
  }

/**
   * Delegate for actually executing the given task with the chosen executor.
   *  用选定的执行者实际执行给定任务
   */
  @Nullable
  protected Object doSubmit(Callable