`

Java多线程系列-JUC线程池

jucjava并发多线程 
阅读更多

线程池架构图

线程池的架构图如下：

 

 

1. Executor

它是"执行者"接口，它是来执行任务的。准确的说，Executor提供了execute()接口来执行已提交的 Runnable 任务的对象。Executor存在的目的是提供一种将"任务提交"与"任务如何运行"分离开来的机制。
它只包含一个函数接口：

void execute(Runnable command)

2. ExecutorService

ExecutorService继承于Executor。它是"执行者服务"接口，它是为"执行者接口Executor"服务而存在的；准确的话，ExecutorService提供了"将任务提交给执行者的接口(submit方法)"，"让执行者执行任务(invokeAll, invokeAny方法)"的接口等等。

ExecutorService的函数列表

 
 1 // 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。
 2 boolean awaitTermination(long timeout, TimeUnit unit)
 3 // 执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。
 4 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
 5 // 执行给定的任务，当所有任务完成或超时期满时（无论哪个首先发生），返回保持任务状态和结果的 Future 列表。
 6 <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
 7 // 执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。
 8 <T> T invokeAny(Collection<? extends Callable<T>> tasks)
 9 // 执行给定的任务，如果在给定的超时期满前某个任务已成功完成（也就是未抛出异常），则返回其结果。
10 <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
11 // 如果此执行程序已关闭，则返回 true。
12 boolean isShutdown()
13 // 如果关闭后所有任务都已完成，则返回 true。
14 boolean isTerminated()
15 // 启动一次顺序关闭，执行以前提交的任务，但不接受新任务。
16 void shutdown()
17 // 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。
18 List<Runnable> shutdownNow()
19 // 提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。
20 <T> Future<T> submit(Callable<T> task)
21 // 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
22 Future<?> submit(Runnable task)
23 // 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
24 <T> Future<T> submit(Runnable task, T result)

3. AbstractExecutorService

AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。
AbstractExecutorService存在的目的是为ExecutorService中的函数接口提供了默认实现。

 

4. ThreadPoolExecutor

ThreadPoolExecutor就是大名鼎鼎的"线程池"。它继承于AbstractExecutorService抽象类。

ThreadPoolExecutor函数列表

 
 1 // 用给定的初始参数和默认的线程工厂及被拒绝的执行处理程序创建新的 ThreadPoolExecutor。
 2 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)
 3 // 用给定的初始参数和默认的线程工厂创建新的 ThreadPoolExecutor。
 4 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)
 5 // 用给定的初始参数和默认被拒绝的执行处理程序创建新的 ThreadPoolExecutor。
 6 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory)
 7 // 用给定的初始参数创建新的 ThreadPoolExecutor。
 8 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
 9 
10 // 基于完成执行给定 Runnable 所调用的方法。
11 protected void afterExecute(Runnable r, Throwable t)
12 // 如果在保持活动时间内没有任务到达，新任务到达时正在替换（如果需要），则设置控制核心线程是超时还是终止的策略。
13 void allowCoreThreadTimeOut(boolean value)
14 // 如果此池允许核心线程超时和终止，如果在 keepAlive 时间内没有任务到达，新任务到达时正在替换（如果需要），则返回 true。
15 boolean allowsCoreThreadTimeOut()
16 // 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行。
17 boolean awaitTermination(long timeout, TimeUnit unit)
18 // 在执行给定线程中的给定 Runnable 之前调用的方法。
19 protected void beforeExecute(Thread t, Runnable r)
20 // 在将来某个时间执行给定任务。
21 void execute(Runnable command)
22 // 当不再引用此执行程序时，调用 shutdown。
23 protected void finalize()
24 // 返回主动执行任务的近似线程数。
25 int getActiveCount()
26 // 返回已完成执行的近似任务总数。
27 long getCompletedTaskCount()
28 // 返回核心线程数。
29 int getCorePoolSize()
30 // 返回线程保持活动的时间，该时间就是超过核心池大小的线程可以在终止前保持空闲的时间值。
31 long getKeepAliveTime(TimeUnit unit)
32 // 返回曾经同时位于池中的最大线程数。
33 int getLargestPoolSize()
34 // 返回允许的最大线程数。
35 int getMaximumPoolSize()
36 // 返回池中的当前线程数。
37 int getPoolSize()
38 // 返回此执行程序使用的任务队列。
39 BlockingQueue<Runnable> getQueue()
40 // 返回用于未执行任务的当前处理程序。
41 RejectedExecutionHandler getRejectedExecutionHandler()
42 // 返回曾计划执行的近似任务总数。
43 long getTaskCount()
44 // 返回用于创建新线程的线程工厂。
45 ThreadFactory getThreadFactory()
46 // 如果此执行程序已关闭，则返回 true。
47 boolean isShutdown()
48 // 如果关闭后所有任务都已完成，则返回 true。
49 boolean isTerminated()
50 // 如果此执行程序处于在 shutdown 或 shutdownNow 之后正在终止但尚未完全终止的过程中，则返回 true。
51 boolean isTerminating()
52 // 启动所有核心线程，使其处于等待工作的空闲状态。
53 int prestartAllCoreThreads()
54 // 启动核心线程，使其处于等待工作的空闲状态。
55 boolean prestartCoreThread()
56 // 尝试从工作队列移除所有已取消的 Future 任务。
57 void purge()
58 // 从执行程序的内部队列中移除此任务（如果存在），从而如果尚未开始，则其不再运行。
59 boolean remove(Runnable task)
60 // 设置核心线程数。
61 void setCorePoolSize(int corePoolSize)
62 // 设置线程在终止前可以保持空闲的时间限制。
63 void setKeepAliveTime(long time, TimeUnit unit)
64 // 设置允许的最大线程数。
65 void setMaximumPoolSize(int maximumPoolSize)
66 // 设置用于未执行任务的新处理程序。
67 void setRejectedExecutionHandler(RejectedExecutionHandler handler)
68 // 设置用于创建新线程的线程工厂。
69 void setThreadFactory(ThreadFactory threadFactory)
70 // 按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。
71 void shutdown()
72 // 尝试停止所有的活动执行任务、暂停等待任务的处理，并返回等待执行的任务列表。
73 List<Runnable> shutdownNow()
74 // 当 Executor 已经终止时调用的方法。
75 protected void terminated()
 

 5. ScheduledExecutorService

ScheduledExecutorService是一个接口，它继承于于ExecutorService。它相当于提供了"延时"和"周期执行"功能的ExecutorService。
ScheduledExecutorService提供了相应的函数接口，可以安排任务在给定的延迟后执行，也可以让任务周期的执行。

ScheduledExecutorService函数列表

 
1 // 创建并执行在给定延迟后启用的 ScheduledFuture。
2 <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)
3 // 创建并执行在给定延迟后启用的一次性操作。
4 ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
5 // 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，接着在 initialDelay + 2 * period 后执行，依此类推。
6 ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
7 // 创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。
8 ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
 

 

6. ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor继承于ThreadPoolExecutor，并且实现了ScheduledExecutorService接口。它相当于提供了"延时"和"周期执行"功能的ScheduledExecutorService。
ScheduledThreadPoolExecutor类似于Timer，但是在高并发程序中，ScheduledThreadPoolExecutor的性能要优于Timer。

ScheduledThreadPoolExecutor函数列表

 
 1 // 使用给定核心池大小创建一个新 ScheduledThreadPoolExecutor。
 2 ScheduledThreadPoolExecutor(int corePoolSize)
 3 // 使用给定初始参数创建一个新 ScheduledThreadPoolExecutor。
 4 ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler)
 5 // 使用给定的初始参数创建一个新 ScheduledThreadPoolExecutor。
 6 ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory)
 7 // 使用给定初始参数创建一个新 ScheduledThreadPoolExecutor。
 8 ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler)
 9 
10 // 修改或替换用于执行 callable 的任务。
11 protected <V> RunnableScheduledFuture<V> decorateTask(Callable<V> callable, RunnableScheduledFuture<V> task)
12 // 修改或替换用于执行 runnable 的任务。
13 protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task)
14 // 使用所要求的零延迟执行命令。
15 void execute(Runnable command)
16 // 获取有关在此执行程序已 shutdown 的情况下、是否继续执行现有定期任务的策略。
17 boolean getContinueExistingPeriodicTasksAfterShutdownPolicy()
18 // 获取有关在此执行程序已 shutdown 的情况下是否继续执行现有延迟任务的策略。
19 boolean getExecuteExistingDelayedTasksAfterShutdownPolicy()
20 // 返回此执行程序使用的任务队列。
21 BlockingQueue<Runnable> getQueue()
22 // 从执行程序的内部队列中移除此任务（如果存在），从而如果尚未开始，则其不再运行。
23 boolean remove(Runnable task)
24 // 创建并执行在给定延迟后启用的 ScheduledFuture。
25 <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)
26 // 创建并执行在给定延迟后启用的一次性操作。
27 ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
28 // 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，接着在 initialDelay + 2 * period 后执行，依此类推。
29 ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
30 // 创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。
31 ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
32 // 设置有关在此执行程序已 shutdown 的情况下是否继续执行现有定期任务的策略。
33 void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value)
34 // 设置有关在此执行程序已 shutdown 的情况下是否继续执行现有延迟任务的策略。
35 void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value)
36 // 在以前已提交任务的执行中发起一个有序的关闭，但是不接受新任务。
37 void shutdown()
38 // 尝试停止所有正在执行的任务、暂停等待任务的处理，并返回等待执行的任务列表。
39 List<Runnable> shutdownNow()
40 // 提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。
41 <T> Future<T> submit(Callable<T> task)
42 // 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
43 Future<?> submit(Runnable task)
44 // 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future。
45 <T> Future<T> submit(Runnable task, T result)

Executors

Executors是个静态工厂类。它通过静态工厂方法返回ExecutorService、ScheduledExecutorService、ThreadFactory 和 Callable 等类的对象。

Executors函数列表

 
 1 // 返回 Callable 对象，调用它时可运行给定特权的操作并返回其结果。
 2 static Callable<Object> callable(PrivilegedAction<?> action)
 3 // 返回 Callable 对象，调用它时可运行给定特权的异常操作并返回其结果。
 4 static Callable<Object> callable(PrivilegedExceptionAction<?> action)
 5 // 返回 Callable 对象，调用它时可运行给定的任务并返回 null。
 6 static Callable<Object> callable(Runnable task)
 7 // 返回 Callable 对象，调用它时可运行给定的任务并返回给定的结果。
 8 static <T> Callable<T> callable(Runnable task, T result)
 9 // 返回用于创建新线程的默认线程工厂。
10 static ThreadFactory defaultThreadFactory()
11 // 创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。
12 static ExecutorService newCachedThreadPool()
13 // 创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们，并在需要时使用提供的 ThreadFactory 创建新线程。
14 static ExecutorService newCachedThreadPool(ThreadFactory threadFactory)
15 // 创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。
16 static ExecutorService newFixedThreadPool(int nThreads)
17 // 创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程，在需要时使用提供的 ThreadFactory 创建新线程。
18 static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)
19 // 创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。
20 static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
21 // 创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行。
22 static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)
23 // 创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。
24 static ExecutorService newSingleThreadExecutor()
25 // 创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程，并在需要时使用提供的 ThreadFactory 创建新线程。
26 static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)
27 // 创建一个单线程执行程序，它可安排在给定延迟后运行命令或者定期地执行。
28 static ScheduledExecutorService newSingleThreadScheduledExecutor()
29 // 创建一个单线程执行程序，它可安排在给定延迟后运行命令或者定期地执行。
30 static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory)
31 // 返回 Callable 对象，调用它时可在当前的访问控制上下文中执行给定的 callable 对象。
32 static <T> Callable<T> privilegedCallable(Callable<T> callable)
33 // 返回 Callable 对象，调用它时可在当前的访问控制上下文中，使用当前上下文类加载器作为上下文类加载器来执行给定的 callable 对象。
34 static <T> Callable<T> privilegedCallableUsingCurrentClassLoader(Callable<T> callable)
35 // 返回用于创建新线程的线程工厂，这些新线程与当前线程具有相同的权限。
36 static ThreadFactory privilegedThreadFactory()
37 // 返回一个将所有已定义的 ExecutorService 方法委托给指定执行程序的对象，但是使用强制转换可能无法访问其他方法。
38 static ExecutorService unconfigurableExecutorService(ExecutorService executor)
39 // 返回一个将所有已定义的 ExecutorService 方法委托给指定执行程序的对象，但是使用强制转换可能无法访问其他方法。
40 static ScheduledExecutorService unconfigurableScheduledExecutorService(ScheduledExecutorService executor)

 

ThreadPoolExecutor简介

ThreadPoolExecutor是线程池类。对于线程池，可以通俗的将它理解为"存放一定数量线程的一个线程集合。线程池允许若个线程同时运行，允许同时运行的线程数量就是线程池的容量；当添加的到线程池中的线程超过它的容量时，会有一部分线程阻塞等待。线程池会通过相应的调度策略和拒绝策略，对添加到线程池中的线程进行管理。"

 

ThreadPoolExecutor数据结构

ThreadPoolExecutor的数据结构如下图所示：

 

 

各个数据在ThreadPoolExecutor.java中的定义如下：

 
// 阻塞队列。
private final BlockingQueue<Runnable> workQueue;
// 互斥锁
private final ReentrantLock mainLock = new ReentrantLock();
// 线程集合。一个Worker对应一个线程。
private final HashSet<Worker> workers = new HashSet<Worker>();
// “终止条件”，与“mainLock”绑定。
private final Condition termination = mainLock.newCondition();
// 线程池中线程数量曾经达到过的最大值。
private int largestPoolSize;
// 已完成任务数量
private long completedTaskCount;
// ThreadFactory对象，用于创建线程。
private volatile ThreadFactory threadFactory;
// 拒绝策略的处理句柄。
private volatile RejectedExecutionHandler handler;
// 保持线程存活时间。
private volatile long keepAliveTime;

private volatile boolean allowCoreThreadTimeOut;
// 核心池大小
private volatile int corePoolSize;
// 最大池大小
private volatile int maximumPoolSize;

1. workers
    workers是HashSet<Work>类型，即它是一个Worker集合。而一个Worker对应一个线程，也就是说线程池通过workers包含了"一个线程集合"。当Worker对应的线程池启动时，它会执行线程池中的任务；当执行完一个任务后，它会从线程池的阻塞队列中取出一个阻塞的任务来继续运行。
    wokers的作用是，线程池通过它实现了"允许多个线程同时运行"。

2. workQueue
    workQueue是BlockingQueue类型，即它是一个阻塞队列。当线程池中的线程数超过它的容量的时候，线程会进入阻塞队列进行阻塞等待。
    通过workQueue，线程池实现了阻塞功能。

3. mainLock
    mainLock是互斥锁，通过mainLock实现了对线程池的互斥访问。

4. corePoolSize和maximumPoolSize
    corePoolSize是"核心池大小"，maximumPoolSize是"最大池大小"。它们的作用是调整"线程池中实际运行的线程的数量"。
    例如，当新任务提交给线程池时(通过execute方法)。
          -- 如果此时，线程池中运行的线程数量< corePoolSize，则创建新线程来处理请求。
          -- 如果此时，线程池中运行的线程数量> corePoolSize，但是却< maximumPoolSize；则仅当阻塞队列满时才创建新线程。
          如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心池大小和最大池大小的值是在创建线程池设置的；但是，也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

5. poolSize
    poolSize是当前线程池的实际大小，即线程池中任务的数量。

6. allowCoreThreadTimeOut和keepAliveTime
    allowCoreThreadTimeOut表示是否允许"线程在空闲状态时，仍然能够存活"；而keepAliveTime是当线程池处于空闲状态的时候，超过keepAliveTime时间之后，空闲的线程会被终止。

7. threadFactory
    threadFactory是ThreadFactory对象。它是一个线程工厂类，"线程池通过ThreadFactory创建线程"。

8. handler
    handler是RejectedExecutionHandler类型。它是"线程池拒绝策略"的句柄，也就是说"当某任务添加到线程池中，而线程池拒绝该任务时，线程池会通过handler进行相应的处理"。

 

综上所说，线程池通过workers来管理"线程集合"，每个线程在启动后，会执行线程池中的任务；当一个任务执行完后，它会从线程池的阻塞队列中取出任务来继续运行。阻塞队列是管理线程池任务的队列，当添加到线程池中的任务超过线程池的容量时，该任务就会进入阻塞队列进行等待。

线程池源码分析

(一) 创建“线程池”

下面以newFixedThreadPool()介绍线程池的创建过程。

1. newFixedThreadPool()

newFixedThreadPool()在Executors.java中定义，源码如下：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

说明：newFixedThreadPool(int nThreads)的作用是创建一个线程池，线程池的容量是nThreads。
         newFixedThreadPool()在调用ThreadPoolExecutor()时，会传递一个LinkedBlockingQueue()对象，而LinkedBlockingQueue是单向链表实现的阻塞队列。在线程池中，就是通过该阻塞队列来实现"当线程池中任务数量超过允许的任务数量时，部分任务会阻塞等待"。

 

2. ThreadPoolExecutor()

ThreadPoolExecutor()在ThreadPoolExecutor.java中定义，源码如下：

 
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
 

说明：该函数实际上是调用ThreadPoolExecutor的另外一个构造函数。该函数的源码如下：

 
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    // 核心池大小
    this.corePoolSize = corePoolSize;
    // 最大池大小
    this.maximumPoolSize = maximumPoolSize;
    // 线程池的等待队列
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    // 线程工厂对象
    this.threadFactory = threadFactory;
    // 拒绝策略的句柄
    this.handler = handler;
}
 

说明：在ThreadPoolExecutor()的构造函数中，进行的是初始化工作。
corePoolSize, maximumPoolSize, unit, keepAliveTime和workQueue这些变量的值是已知的，它们都是通过newFixedThreadPool()传递而来。下面看看threadFactory和handler对象。

 

2.1 ThreadFactory

线程池中的ThreadFactory是一个线程工厂，线程池创建线程都是通过线程工厂对象(threadFactory)来完成的。
上面所说的threadFactory对象，是通过 Executors.defaultThreadFactory()返回的。Executors.java中的defaultThreadFactory()源码如下：

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

defaultThreadFactory()返回DefaultThreadFactory对象。Executors.java中的DefaultThreadFactory()源码如下：

 

 
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() +
                     "-thread-";
    }

    // 提供创建线程的API。
    public Thread newThread(Runnable r) {
        // 线程对应的任务是Runnable对象r
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        // 设为“非守护线程”
        if (t.isDaemon())
            t.setDaemon(false);
        // 将优先级设为“Thread.NORM_PRIORITY”
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}
 

 

说明：ThreadFactory的作用就是提供创建线程的功能的线程工厂。
         它是通过newThread()提供创建线程功能的，下面简单说说newThread()。newThread()创建的线程对应的任务是Runnable对象，它创建的线程都是“非守护线程”而且“线程优先级都是Thread.NORM_PRIORITY”。

 

2.2 RejectedExecutionHandler

handler是ThreadPoolExecutor中拒绝策略的处理句柄。所谓拒绝策略，是指将任务添加到线程池中时，线程池拒绝该任务所采取的相应策略。
线程池默认会采用的是defaultHandler策略，即AbortPolicy策略。在AbortPolicy策略中，线程池拒绝任务时会抛出异常！
defaultHandler的定义如下：

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

AbortPolicy的源码如下：

 
public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }

    // 抛出异常
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}
 

 

(二) 添加任务到“线程池”

1. execute()

execute()定义在ThreadPoolExecutor.java中，源码如下：

 
public void execute(Runnable command) {
    // 如果任务为null，则抛出异常。
    if (command == null)
        throw new NullPointerException();
    // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
    int c = ctl.get();
    // 当线程池中的任务数量 < "核心池大小"时，即线程池中少于corePoolSize个任务。
    // 则通过addWorker(command, true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 当线程池中的任务数量 >= "核心池大小"时，
    // 而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 否则，如果"线程池中任务数量"为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null。
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 通过addWorker(command, false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
    // 如果addWorker(command, false)执行失败，则通过reject()执行相应的拒绝策略的内容。
    else if (!addWorker(command, false))
        reject(command);
}
 

说明：execute()的作用是将任务添加到线程池中执行。它会分为3种情况进行处理：
        情况1 -- 如果"线程池中任务数量" < "核心池大小"时，即线程池中少于corePoolSize个任务；此时就新建一个线程，并将该任务添加到线程中进行执行。
        情况2 -- 如果"线程池中任务数量" >= "核心池大小"，并且"线程池是允许状态"；此时，则将任务添加到阻塞队列中阻塞等待。在该情况下，会再次确认"线程池的状态"，如果"第2次读到的线程池状态"和"第1次读到的线程池状态"不同，则从阻塞队列中删除该任务。
        情况3 -- 非以上两种情况。在这种情况下，尝试新建一个线程，并将该任务添加到线程中进行执行。如果执行失败，则通过reject()拒绝该任务。

 

2. addWorker()

addWorker()的源码如下：

 
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 更新"线程池状态和计数"标记，即更新ctl。
    for (;;) {
        // 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
        int c = ctl.get();
        // 获取线程池状态。
        int rs = runStateOf(c);

        // 有效性检查
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            // 获取线程池中任务的数量。
            int wc = workerCountOf(c);
            // 如果"线程池中任务的数量"超过限制，则返回false。
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 通过CAS函数将c的值+1。操作失败的话，则退出循环。
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            // 检查"线程池状态"，如果与之前的状态不同，则从retry重新开始。
            if (runStateOf(c) != rs)
                continue retry;
            // else CAS failed due to workerCount change; retry inner loop
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    // 添加任务到线程池，并启动任务所在的线程。
    try {
        final ReentrantLock mainLock = this.mainLock;
        // 新建Worker，并且指定firstTask为Worker的第一个任务。
        w = new Worker(firstTask);
        // 获取Worker对应的线程。
        final Thread t = w.thread;
        if (t != null) {
            // 获取锁
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);

                // 再次确认"线程池状态"
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 将Worker对象(w)添加到"线程池的Worker集合(workers)"中
                    workers.add(w);
                    // 更新largestPoolSize
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                // 释放锁
                mainLock.unlock();
            }
            // 如果"成功将任务添加到线程池"中，则启动任务所在的线程。 
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回任务是否启动。
    return workerStarted;
}
 

说明：
    addWorker(Runnable firstTask, boolean core) 的作用是将任务(firstTask)添加到线程池中，并启动该任务。
    core为true的话，则以corePoolSize为界限，若"线程池中已有任务数量>=corePoolSize"，则返回false；core为false的话，则以maximumPoolSize为界限，若"线程池中已有任务数量>=maximumPoolSize"，则返回false。
    addWorker()会先通过for循环不断尝试更新ctl状态，ctl记录了"线程池中任务数量和线程池状态"。
    更新成功之后，再通过try模块来将任务添加到线程池中，并启动任务所在的线程。

    从addWorker()中，我们能清晰的发现：线程池在添加任务时，会创建任务对应的Worker对象；而一个Workder对象包含一个Thread对象。(01) 通过将Worker对象添加到"线程的workers集合"中，从而实现将任务添加到线程池中。 (02) 通过启动Worker对应的Thread线程，则执行该任务。

 

3. submit()

补充说明一点，submit()实际上也是通过调用execute()实现的，源码如下：

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

 

(三) 关闭“线程池”

shutdown()的源码如下：

 
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    // 获取锁
    mainLock.lock();
    try {
        // 检查终止线程池的“线程”是否有权限。
        checkShutdownAccess();
        // 设置线程池的状态为关闭状态。
        advanceRunState(SHUTDOWN);
        // 中断线程池中空闲的线程。
        interruptIdleWorkers();
        // 钩子函数，在ThreadPoolExecutor中没有任何动作。
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        // 释放锁
        mainLock.unlock();
    }
    // 尝试终止线程池
    tryTerminate();
}
 

说明：shutdown()的作用是关闭线程池。

 

我们介绍过，线程有5种状态：新建状态，就绪状态，运行状态，阻塞状态，死亡状态。线程池也有5种状态；然而，线程池不同于线程，线程池的5种状态是：Running, SHUTDOWN, STOP, TIDYING, TERMINATED。

 

线程池状态定义代码如下：

复制代码
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;

private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
private static int ctlOf(int rs, int wc) { return rs | wc; }
复制代码

说明：
ctl是一个AtomicInteger类型的原子对象。ctl记录了"线程池中的任务数量"和"线程池状态"2个信息。
ctl共包括32位。其中，高3位表示"线程池状态"，低29位表示"线程池中的任务数量"。

RUNNING    -- 对应的高3位值是111。
SHUTDOWN   -- 对应的高3位值是000。
STOP       -- 对应的高3位值是001。
TIDYING    -- 对应的高3位值是010。
TERMINATED -- 对应的高3位值是011。

 

线程池各个状态之间的切换如下图所示：

 

 

1. RUNNING

(01) 状态说明：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。
(02) 状态切换：线程池的初始化状态是RUNNING。换句话说，线程池被一旦被创建，就处于RUNNING状态！
道理很简单，在ctl的初始化代码中(如下)，就将它初始化为RUNNING状态，并且"任务数量"初始化为0。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

 

2. SHUTDOWN

(01) 状态说明：线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。
(02) 状态切换：调用线程池的shutdown()接口时，线程池由RUNNING -> SHUTDOWN。

 

3. STOP

(01) 状态说明：线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。
(02) 状态切换：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。

 

4. TIDYING
(01) 状态说明：当所有的任务已终止，ctl记录的"任务数量"为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。
(02) 状态切换：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。
当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。

 

5. TERMINATED
(01) 状态说明：线程池彻底终止，就变成TERMINATED状态。
(02) 状态切换：线程池处在TIDYING状态时，执行完terminated()之后，就会由 TIDYING -> TERMINATED。

 

拒绝策略介绍

线程池的拒绝策略，是指当任务添加到线程池中被拒绝，而采取的处理措施。
当任务添加到线程池中之所以被拒绝，可能是由于：第一，线程池异常关闭。第二，任务数量超过线程池的最大限制。

线程池共包括4种拒绝策略，它们分别是：AbortPolicy, CallerRunsPolicy, DiscardOldestPolicy和DiscardPolicy。

AbortPolicy         -- 当任务添加到线程池中被拒绝时，它将抛出 RejectedExecutionException 异常。
CallerRunsPolicy    -- 当任务添加到线程池中被拒绝时，会在线程池当前正在运行的Thread线程池中处理被拒绝的任务。
DiscardOldestPolicy -- 当任务添加到线程池中被拒绝时，线程池会放弃等待队列中最旧的未处理任务，然后将被拒绝的任务添加到等待队列中。
DiscardPolicy       -- 当任务添加到线程池中被拒绝时，线程池将丢弃被拒绝的任务。

线程池默认的处理策略是AbortPolicy！

 

Callable 和 Future 简介

　　Callable 和 Future 是比较有趣的一对组合。当我们需要获取线程的执行结果时，就需要用到它们。Callable用于产生结果，Future用于获取结果。

1. Callable

Callable 是一个接口，它只包含一个call()方法。Callable是一个返回结果并且可能抛出异常的任务。

为了便于理解，我们可以将Callable比作一个Runnable接口，而Callable的call()方法则类似于Runnable的run()方法。

Callable的源码如下：

public interface Callable<V> {
    V call() throws Exception;
}

说明：从中我们可以看出Callable支持泛型。

 

2. Future

Future 是一个接口。它用于表示异步计算的结果。提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。

Future的源码如下：

 
public interface Future<V> {
    // 试图取消对此任务的执行。
    boolean  cancel(boolean mayInterruptIfRunning)

    // 如果在任务正常完成前将其取消，则返回 true。
    boolean  isCancelled()

    // 如果任务已完成，则返回 true。
    boolean  isDone()

    // 如有必要，等待计算完成，然后获取其结果。
    V  get() throws InterruptedException, ExecutionException;

    // 如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）。
    V  get(long timeout, TimeUnit unit)
          throws InterruptedException, ExecutionException, TimeoutException;
}
 

说明： Future用于表示异步计算的结果。它的实现类是FutureTask，在讲解FutureTask之前，我们先看看Callable, Future, FutureTask它们之间的关系图，如下：

说明：
(01) RunnableFuture是一个接口，它继承了Runnable和Future这两个接口。RunnableFuture的源码如下：

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

(02) FutureTask实现了RunnableFuture接口。所以，我们也说它实现了Future接口。

 

 

源码分析(基于JDK1.7.0_40)

1. submit()

submit()在java/util/concurrent/AbstractExecutorService.java中实现，它的源码如下：

 
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    // 创建一个RunnableFuture对象
    RunnableFuture<T> ftask = newTaskFor(task);
    // 执行“任务ftask”
    execute(ftask);
    // 返回“ftask”
    return ftask;
}
 

说明：submit()通过newTaskFor(task)创建了RunnableFuture对象ftask。它的源码如下：

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}

 

2. FutureTask的构造函数

FutureTask的构造函数如下：

 
public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    // callable是一个Callable对象
    this.callable = callable;
    // state记录FutureTask的状态
    this.state = NEW;       // ensure visibility of callable
}
 

 

3. FutureTask的run()方法

我们继续回到submit()的源码中。
在newTaskFor()新建一个ftask对象之后，会通过execute(ftask)执行该任务。此时ftask被当作一个Runnable对象进行执行，最终会调用到它的run()方法；ftask的run()方法在java/util/concurrent/FutureTask.java中实现，源码如下：

 
public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        // 将callable对象赋值给c。
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                // 执行Callable的call()方法，并保存结果到result中。
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            // 如果运行成功，则将result保存
            if (ran)
                set(result);
        }
    } finally {
        runner = null;
        // 设置“state状态标记”
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}
 

说明：run()中会执行Callable对象的call()方法，并且最终将结果保存到result中，并通过set(result)将result保存。
　　    之后调用FutureTask的get()方法，返回的就是通过set(result)保存的值。

 

rel： http://www.cnblogs.com/skywang12345/p/3509903.html

分享到：
| Java多线程系列-JUC集合
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    Java 多线程与并发(17-26)-JUC线程池- FutureTask详解.pdf

    ### Java多线程与并发(17-26)-JUC线程池-FutureTask详解 #### 一、概述 本文将围绕Java多线程与并发中的重要概念——`FutureTask`进行深入探讨。`FutureTask`是Java并发库中的一个关键组件，它实现了`...

    java并发编程：juc线程池

    Java并发编程中的JUC线程池是Java程序员必须掌握的关键技术之一，它允许开发者高效地管理并发执行的任务，充分利用多核处理器的性能。线程池的出现解决了在并发环境中线程创建、销毁带来的开销，提高了系统资源的...

    05-Java多线程并发编程JUC.pdf

    以上知识点涵盖了Java多线程并发编程和JUC的基础与高级特性，从线程的状态、创建、启动、通信、同步机制，到线程池的使用和线程中断处理，再到线程核心特性的保障，以及锁机制的深入理解和应用。掌握这些知识点对于...

    JAVA线程高级-线程按序交替执行

    在Java编程中，多线程是并发编程的重要组成部分，它允许程序同时执行多个任务，从而提高了系统的效率和响应性。然而，在某些场景下，我们可能需要控制线程的执行顺序，确保它们按照特定的顺序交替运行，这在并发编程...

    Java-JUC-多线程 进阶

    Java-JUC-多线程进阶resources是 Java 并发编程的高级课程，涵盖了 Java 中的并发编程概念、线程安全、锁机制、集合类、线程池、函数式接口、Stream流式计算等多个方面。 什么是JUC JUC（Java Utilities for ...

    Java 多线程与并发(7-26)-JUC - 类汇总和学习指南.pdf

    "Java 多线程与并发(7-26)-JUC - 类汇总和学习指南" Java 多线程与并发是 Java 编程语言中的一部分，用于处理多线程和并发编程。Java 提供了一个名为 JUC（Java Utilities for Concurrency）的框架，用于帮助开发者...

    阿里专家级并发编程架构师课程 彻底解决JAVA并发编程疑难杂症 JAVA并发编程高级教程

    阿里专家级并发编程架构师级课程，完成课程的学习可以帮助同学们解决非常多的JAVA并发编程疑难杂症，极大的提高JAVA并发编程的效率。课程内容包括了JAVA手写线程池，UC线程池API详解，线程安全根因详解，锁与原子类...

    Java多线程（Synchronized+Volatile+JUC 并发工具原理+线程状态+CAS+线程池）

    Java 多线程（Synchronized+Volatile+JUC 并发工具原理+线程状态+CAS+线程池） Java 多线程是 Java 语言中的一种并发编程机制，允许程序同时执行多个线程，以提高程序的执行效率和响应速度。 Java 多线程机制提供了...

    Java 多线程与并发(16-26)-JUC集合- BlockingQueue详解.pdf

    理解`BlockingQueue`和`BlockingDeque`的关键在于它们的阻塞特性，这使得它们在多线程环境中的使用既高效又安全。面试时，可能会被问到如何使用它们实现生产者-消费者模型，或者如何在并发环境下解决数据同步问题，...

    A-JUC-JVM-Java并发知识..pdf

    #### JUC多线程及高并发 Java并发编程包（java.util.concurrent，简称JUC）封装了大量用于高并发编程的工具类和接口，其中涉及了线程池、阻塞队列、同步器、原子操作类等。在并发环境下，可以有效降低线程创建和管理...

    个人学习-JUC-笔记

    Java并发编程是Java开发中的重要领域，而JUC（Java Util Concurrency）是Java平台提供的一套高级并发工具包，它极大地简化了多线程和并发控制的复杂性。本笔记主要围绕尚硅谷周阳老师的JUC课程展开，旨在帮助个人...

    thread.zip--狂神多线程+juc

    《狂神多线程+juc》是一份针对Java并发编程深度学习的资源包，主要涵盖了Java多线程基础以及Java并发工具集（JUC）的高级应用。在Java编程中，多线程是提高程序效率、实现并发执行的重要手段，而JUC（Java ...

    heima-JUC-资料

    Java并发编程是Java Util Concurrency（JUC）库的核心内容，它为多线程环境提供了高效、安全且灵活的工具和API。JUC是Java 5及后续版本引入的一个重要特性，极大地提升了Java在多处理器和高并发环境下的性能表现。 ...

    JUC学习笔记（Java多线程）

    本笔记将深入探讨JUC的相关知识点，帮助开发者理解并掌握Java多线程的核心概念。 首先，我们来了解一下Java线程。线程是程序执行的基本单元，允许多个任务在同一时间运行，从而提高了系统资源的利用率。在Java中，...

    【面试资料】-（机构内训资料）Java多线程面试59题（含答案）_.zip

    Java多线程是Java编程中的一个关键领域，尤其在面试中常常被重点考察。这份"【面试资料】-（机构内训资料）Java多线程面试59题（含答案）_"的压缩文件包含了59个关于Java多线程的面试问题及解答，可以帮助我们深入...

    java多线程源码-basic-of-java:java基础：算法、多线程、锁、AQS、JUC、NIO、一些JDK源码、一些问题总结

    基础：算法、多线程、锁、AQS、JUC、NIO、一些 JDK 源码、一些问题总结 记录自己对于 java 的一些探索和思路： 多线程、锁、map、list 的一些实现 网络编程中的一些 tips，如 nio server netty的一些探索，如 netty ...

    JUC代码收集，java高并发多线程学习

    本资源"JUC代码收集，java高并发多线程学习"显然是一个专注于探讨和学习JUC库的资料包。 JUC库包含多个子包，如`concurrent`, `atomic`, `locks`等，每个子包都有其特定的功能和用途： 1. **concurrent**：这是JUC...

    java之JUC系列.pdf

    JUC（Java.util.concurrent）是Java并发编程库的一个重要组成部分，专门提供了在多线程环境下用于控制并发执行的工具类和接口。本篇内容将详细介绍JUC中的一些核心组件及其使用方法，包括CountDownLatch、Executors...

Magicbox
Global site tag (gtag.js) - Google Analytics