从源码理解线程池

Executor接口

源码非常简单，只有一个execute（Runnable command）回调接口

public interface Executor {
    void execute(Runnable command);
}

执行已提交的

Runnable

任务对象。此接口提供一种将任务提交与每个任务将如何运行的机制（包括线程使用的细节、调度等）分离开来的方法。通常使用 Executor 而不是显式地创建线程。例如，可能会使用以下方法

Executor executor = anExecutor;
executor.execute(new RunnableTask1());
 ...

不过，Executor 接口并没有严格地要求执行是异步的。在最简单的情况下，执行程序可以在调用方的线程中立即运行已提交的任务：

class DirectExecutor implements Executor {
     public void execute(Runnable r) {
         r.run();
     }
 }

更常见的是，任务是在某个不是调用方线程的线程中执行的。以下执行程序将为每个任务生成一个新线程。

class ThreadPerTaskExecutor implements Executor {
     public void execute(Runnable r) {
         new Thread(r).start();
     }
 }

许多 Executor 实现都对调度任务的方式和时间强加了某种限制。以下执行程序使任务提交与第二个执行程序保持连续，这展示了一个复合执行程序。

class SerialExecutor implements Executor {
     final Queue<Runnable> tasks = new LinkedBlockingQueue<Runnable>();
     final Executor executor;
     Runnable active;

     SerialExecutor(Executor executor) {
         this.executor = executor;
     }

     public synchronized void execute(final Runnable r) {
         tasks.offer(new Runnable() {
             public void run() {
                 try {
                     r.run();
                 } finally {
                     scheduleNext();
                 }
             }
         });
         if (active == null) {
             scheduleNext();
         }
     }

     protected synchronized void scheduleNext() {
         if ((active = tasks.poll()) != null) {
             executor.execute(active);
         }
     }
 }

ExecutorService接口

该接口提供了管理终止的方法,以及可为跟踪一个或多个异步任务执行状况而生成 Future 的方法。先看下源码

public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Executors类为创建ExecutorService提供了便捷的工厂方法。它只有一个直接实现类ThreadPoolExecutor和间接实现类ScheduledThreadPoolExecutor。

ExecutorService在Executor的基础上增加了一些方法，其中有两个核心的方法：

1、Future<?> submit(Runnable task)

2、<T> Future<T> submit(Callable<T> task)

通过创建并返回一个可用于取消执行和/或等待完成的 Future，方法submit扩展了基本方法

Executor.execute(java.lang.Runnable)。

下面对ExecutorService的函数进行一下简单介绍：

void shutdown()：启动一个关闭命令，不再接受新任务，当所有已提交任务执行完后，就关闭。
List<Runnable> shutdownNow()：试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。它无法保证能够停止正在处理的活动执行任务，但是会尽力尝试。例如，通过 Thread.interrupt() 来取消典型的实现，所以任何任务无法响应中断都可能永远无法终止。应该关闭未使用的 ExecutorService以允许回收其资源。
boolean isShutdown()：如果此执行程序已关闭，则返回 true。
boolean isTerminated()：如果关闭后所有任务都已完成，则返回 true。注意，除非首先调用 shutdown 或 shutdownNow，否则 isTerminated 永不为 true。
boolean awaitTermination(long timeout,TimeUnit unit) ：如果此执行程序终止，则返回 true；如果终止前超时期满，则返回 false
<T> Future<T> submit(Callable<T> task)：提交一个有返回值的任务用于执行，返回一个表示任务结果的 Future。该 Future 的 get 方法在成功完成时将会返回该任务的结果。
<T> Future<T> submit(Runnable task,T result)：提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。该 Future 的 get 方法在成功完成时将会返回给定的结果。
Future<?> submit(Runnable task)：提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。该 Future 的 get 方法在成功完成时将会返回 null。
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)：执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。返回列表的所有元素的 Future.isDone() 为 true。注意，该方法会一直阻塞直到所有任务完成。可以正常地或通过抛出异常来终止已完成任务。如果正在进行此操作时修改了给定的collection，则此方法的结果是不确定的。

ThreadPoolExecutor

ThreadPoolExecutor是ExecutorService的一个实现类，它使用可能的几个线程池之一执行每个提交的任务，通常使用 Executors 工厂方法配置。

线程池可以解决两个不同问题：由于减少了每个任务调用的开销，它们通常可以在执行大量异步任务时提供增强的性能，并且还可以提供绑定和管理资源（包括执行任务集时使用的线程）的方法。

每个 ThreadPoolExecutor 还维护着一些基本的统计数据，如完成的任务数。为了便于跨大量上下文使用，此类提供了很多可调整的参数和扩展钩子 (hook)。

对于核心的几个线程池,无论是 newFixedThreadPool（）方法、 newSingleThreadExecutor()还是 newCachedThreadPool（）方法, 虽然看起来创建的线程有着完全不同的功能特点, 但其内部实现均使用了 ThreadPoolExecutor实现。下面给出了这三个线程池的实现方式:

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
//使用一个基于FIFO排序的阻塞队列，在所有corePoolSize线程都忙时新任务将在队列中等待
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

由以上线程池的实现代码可以看到, 它们都只是 ThreadPoolExecutor 类的封装。为何ThreadPoolExecutor有如此强大的功能呢? 来看一下 ThreadPoolExecutor最重要的构造函数:

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) //后两个参数为可选参数

参数说明：

corePoolSize：指定了线程池中的核心线程数
maximumPoolSize:最大可允许创建的线程数，corePoolSize和maximumPoolSize设置的边界自动调整池大小：corePoolSize=运行的线程数= maximumPoolSize：创建固定大小的线程池
keepAliveTime:如果线程数多于corePoolSize,则这些多余的线程的空闲时间超过keepAliveTime时将被终止
unit:keepAliveTime参数的时间单位
workQueue:保存任务的阻塞队列，被提交但尚未执行的任务，与线程池的大小有关：当运行的线程数少于corePoolSize时，在有新任务时直接创建新线程来执行任务而无需再进队列；当运行的线程数等于或多于corePoolSize，在有新任务添加时则先加入队列，不直接创建线程；当队列满时，在有新任务时就创建新线程
threadFactory:线程工厂，用于创建新线程，默认使用defaultThreadFactory创建线程
handle:定义处理被拒绝任务的策略，默认使用ThreadPoolExecutor.AbortPolicy,任务被拒绝时将抛出RejectExecutorException

ThreadPoolExecutor将根据corePoolSize和 maximumPoolSize设置的边界自动调整线程池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时，如果运行的线程少于corePoolSize，则创建新线程来执行新任务，即使线程池中的其他线程是空闲的；如果运行的线程多于corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程；如果设置的corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池；如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许线程池适应任意数量的并发任务。

在大多数情况下，核心和最大池大小仅基于构造来设置，不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

参考：

https://www.cnblogs.com/MOBIN/p/5436482.html http://cmsblogs.com/?p=2444 https://blog.csdn.net/linghu_java/article/details/17123057

从源码理解线程池

Executor接口

ExecutorService接口

ThreadPoolExecutor

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method