Java多线程系列--“JUC线程池”

2019年3月10日 253次阅读来源: JUC

这一系列文章是我在网上看到的一个写的很不错的系列，在此特整合作为学习的收藏和备忘资料，也希望可以帮到需要的人。转自：
http://www.cnblogs.com/skywang12345/p/3509903.html
Java多线程系列–“JUC线程池”01之线程池架构

概要

前面分别介绍了”Java多线程基础“、”JUC原子类“和”JUC锁“。本章介绍JUC的最后一部分的内容——线程池。内容包括：
线程池架构图
 线程池示例

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3509903.html

线程池架构图

线程池的架构图如下：

1. Executor

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

void execute(Runnable command)

2. ExecutorService

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

ExecutorService的函数列表

《Java多线程系列--“JUC线程池”》
View Code

3. AbstractExecutorService

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

AbstractExecutorService函数列表
由于它的函数列表和ExecutorService一样，这里就不再重复列举了。

4. ThreadPoolExecutor

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

ThreadPoolExecutor函数列表

《Java多线程系列--“JUC线程池”》
View Code

5. ScheduledExecutorService

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

ScheduledExecutorService函数列表

《Java多线程系列--“JUC线程池”》
View Code

6. ScheduledThreadPoolExecutor

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

ScheduledThreadPoolExecutor函数列表

《Java多线程系列--“JUC线程池”》
View Code

7. Executors

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

Executors函数列表

《Java多线程系列--“JUC线程池”》
View Code

线程池示例

下面通过示例来对线程池的使用做简单演示。

 1 import java.util.concurrent.Executors;
 2 import java.util.concurrent.ExecutorService;
 3 
 4 public class ThreadPoolDemo1 {
 5 
 6     public static void main(String[] args) {
 7         // 创建一个可重用固定线程数的线程池
 8         ExecutorService pool = Executors.newFixedThreadPool(2);
 9         // 创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口
10         Thread ta = new MyThread();
11         Thread tb = new MyThread();
12         Thread tc = new MyThread();
13         Thread td = new MyThread();
14         Thread te = new MyThread();
15         // 将线程放入池中进行执行
16         pool.execute(ta);
17         pool.execute(tb);
18         pool.execute(tc);
19         pool.execute(td);
20         pool.execute(te);
21         // 关闭线程池
22         pool.shutdown();
23     }
24 }
25 
26 class MyThread extends Thread {
27 
28     @Override
29     public void run() {
30         System.out.println(Thread.currentThread().getName()+ " is running.");
31     }
32 }

运行结果：

pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.

结果说明：
主线程中创建了线程池pool，线程池的容量是2。即，线程池中最多能同时运行2个线程。
紧接着，将ta,tb,tc,td,te这3个线程添加到线程池中运行。
最后，通过shutdown()关闭线程池。

Java多线程系列–“JUC线程池”02之线程池原理(一)

概要

在上一章”Java多线程系列–“JUC线程池”01之线程池架构“中，我们了解了线程池的架构。线程池的实现类是ThreadPoolExecutor类。本章，我们通过分析ThreadPoolExecutor类，来了解线程池的原理。内容包括：
ThreadPoolExecutor简介
 ThreadPoolExecutor数据结构
 线程池调度

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3509941.html

ThreadPoolExecutor简介

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

ThreadPoolExecutor数据结构

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

《Java多线程系列--“JUC线程池”》

各个数据在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来管理”线程集合”，每个线程在启动后，会执行线程池中的任务；当一个任务执行完后，它会从线程池的阻塞队列中取出任务来继续运行。阻塞队列是管理线程池任务的队列，当添加到线程池中的任务超过线程池的容量时，该任务就会进入阻塞队列进行等待。

线程池调度

我们通过下面的图看看下面线程池中任务的调度策略，加深对线程池的理解。

图-01:

《Java多线程系列--“JUC线程池”》

图-02:

《Java多线程系列--“JUC线程池”》

说明：
在”图-01″中，线程池中有N个任务。”任务1″, “任务2”, “任务3″这3个任务在执行，而”任务3″到”任务N”在阻塞队列中等待。正在执行的任务，在workers集合中，workers集合包含3个Worker，每一个Worker对应一个Thread线程，Thread线程每次处理一个任务。
当workers集合中处理完某一个任务之后，会从阻塞队列中取出一个任务来继续执行，如图-02所示。图-02表示”任务1″处理完毕之后，线程池将”任务4″从阻塞队列中取出，放到workers中进行处理。

Java多线程系列–“JUC线程池”03之线程池原理(二)

概要

在前面一章”Java多线程系列–“JUC线程池”02之线程池原理(一)“中介绍了线程池的数据结构，本章会通过分析线程池的源码，对线程池进行说明。内容包括：
线程池示例
 参考代码(基于JDK1.7.0_40)
线程池源码分析
 (一) 创建“线程池”
(二) 添加任务到“线程池”
(三) 关闭“线程池”

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3509954.html

线程池示例

在分析线程池之前，先看一个简单的线程池示例。

 1 import java.util.concurrent.Executors;
 2 import java.util.concurrent.ExecutorService;
 3 
 4 public class ThreadPoolDemo1 {
 5 
 6     public static void main(String[] args) {
 7         // 创建一个可重用固定线程数的线程池
 8         ExecutorService pool = Executors.newFixedThreadPool(2);
 9         // 创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口
10         Thread ta = new MyThread();
11         Thread tb = new MyThread();
12         Thread tc = new MyThread();
13         Thread td = new MyThread();
14         Thread te = new MyThread();
15         // 将线程放入池中进行执行
16         pool.execute(ta);
17         pool.execute(tb);
18         pool.execute(tc);
19         pool.execute(td);
20         pool.execute(te);
21         // 关闭线程池
22         pool.shutdown();
23     }
24 }
25 
26 class MyThread extends Thread {
27 
28     @Override
29     public void run() {
30         System.out.println(Thread.currentThread().getName()+ " is running.");
31     }
32 }

运行结果：

pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.

示例中，包括了线程池的创建，将任务添加到线程池中，关闭线程池这3个主要的步骤。稍后，我们会从这3个方面来分析ThreadPoolExecutor。

参考代码(基于JDK1.7.0_40)

Executors完整源码

《Java多线程系列--“JUC线程池”》
View Code

ThreadPoolExecutor完整源码

《Java多线程系列--“JUC线程池”》
View Code

线程池源码分析

(一) 创建“线程池”

下面以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是单向链表实现的阻塞队列。在线程池中，就是通过该阻塞队列来实现”当线程池中任务数量超过允许的任务数量时，部分任务会阻塞等待”。
关于LinkedBlockingQueue的实现细节，读者可以参考”Java多线程系列–“JUC集合”08之 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()的作用是关闭线程池。

Java多线程系列–“JUC线程池”04之线程池原理(三)

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3509960.html

本章介绍线程池的生命周期。在”Java多线程系列–“基础篇”01之基本概念“中，我们介绍过，线程有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。

Java多线程系列–“JUC线程池”05之线程池原理(四)

概要

本章介绍线程池的拒绝策略。内容包括：
拒绝策略介绍
 拒绝策略对比和示例

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3512947.html

拒绝策略介绍

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

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

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

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

拒绝策略对比和示例

下面通过示例，分别演示线程池的4种拒绝策略。
1. DiscardPolicy 示例
 2. DiscardOldestPolicy 示例
 3. AbortPolicy 示例
 4. CallerRunsPolicy 示例

1. DiscardPolicy 示例

 1 import java.lang.reflect.Field;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy;
 6 
 7 public class DiscardPolicyDemo {
 8 
 9     private static final int THREADS_SIZE = 1;
10     private static final int CAPACITY = 1;
11 
12     public static void main(String[] args) throws Exception {
13 
14         // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
15         ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
16                 new ArrayBlockingQueue<Runnable>(CAPACITY));
17         // 设置线程池的拒绝策略为"丢弃"
18         pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
19 
20         // 新建10个任务，并将它们添加到线程池中。
21         for (int i = 0; i < 10; i++) {
22             Runnable myrun = new MyRunnable("task-"+i);
23             pool.execute(myrun);
24         }
25         // 关闭线程池
26         pool.shutdown();
27     }
28 }
29 
30 class MyRunnable implements Runnable {
31     private String name;
32     public MyRunnable(String name) {
33         this.name = name;
34     }
35     @Override
36     public void run() {
37         try {
38             System.out.println(this.name + " is running.");
39             Thread.sleep(100);
40         } catch (Exception e) {
41             e.printStackTrace();
42         }
43     }
44 }

运行结果：

task-0 is running.
task-1 is running.

结果说明：线程池pool的”最大池大小”和”核心池大小”都为1(THREADS_SIZE)，这意味着”线程池能同时运行的任务数量最大只能是1″。
线程池pool的阻塞队列是ArrayBlockingQueue，ArrayBlockingQueue是一个有界的阻塞队列，ArrayBlockingQueue的容量为1。这也意味着线程池的阻塞队列只能有一个线程池阻塞等待。
根据””中分析的execute()代码可知：线程池中共运行了2个任务。第1个任务直接放到Worker中，通过线程去执行；第2个任务放到阻塞队列中等待。其他的任务都被丢弃了！

2. DiscardOldestPolicy 示例

 1 import java.lang.reflect.Field;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 import java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy;
 6 
 7 public class DiscardOldestPolicyDemo {
 8 
 9     private static final int THREADS_SIZE = 1;
10     private static final int CAPACITY = 1;
11 
12     public static void main(String[] args) throws Exception {
13 
14         // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
15         ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
16                 new ArrayBlockingQueue<Runnable>(CAPACITY));
17         // 设置线程池的拒绝策略为"DiscardOldestPolicy"
18         pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
19 
20         // 新建10个任务，并将它们添加到线程池中。
21         for (int i = 0; i < 10; i++) {
22             Runnable myrun = new MyRunnable("task-"+i);
23             pool.execute(myrun);
24         }
25         // 关闭线程池
26         pool.shutdown();
27     }
28 }
29 
30 class MyRunnable implements Runnable {
31     private String name;
32     public MyRunnable(String name) {
33         this.name = name;
34     }
35     @Override
36     public void run() {
37         try {
38             System.out.println(this.name + " is running.");
39             Thread.sleep(200);
40         } catch (Exception e) {
41             e.printStackTrace();
42         }
43     }
44 }

运行结果：

task-0 is running.
task-9 is running.

结果说明：将”线程池的拒绝策略”由DiscardPolicy修改为DiscardOldestPolicy之后，当有任务添加到线程池被拒绝时，线程池会丢弃阻塞队列中末尾的任务，然后将被拒绝的任务添加到末尾。

3. AbortPolicy 示例

 1 import java.lang.reflect.Field;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
 6 import java.util.concurrent.RejectedExecutionException;
 7 
 8 public class AbortPolicyDemo {
 9 
10     private static final int THREADS_SIZE = 1;
11     private static final int CAPACITY = 1;
12 
13     public static void main(String[] args) throws Exception {
14 
15         // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
16         ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
17                 new ArrayBlockingQueue<Runnable>(CAPACITY));
18         // 设置线程池的拒绝策略为"抛出异常"
19         pool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
20 
21         try {
22 
23             // 新建10个任务，并将它们添加到线程池中。
24             for (int i = 0; i < 10; i++) {
25                 Runnable myrun = new MyRunnable("task-"+i);
26                 pool.execute(myrun);
27             }
28         } catch (RejectedExecutionException e) {
29             e.printStackTrace();
30             // 关闭线程池
31             pool.shutdown();
32         }
33     }
34 }
35 
36 class MyRunnable implements Runnable {
37     private String name;
38     public MyRunnable(String name) {
39         this.name = name;
40     }
41     @Override
42     public void run() {
43         try {
44             System.out.println(this.name + " is running.");
45             Thread.sleep(200);
46         } catch (Exception e) {
47             e.printStackTrace();
48         }
49     }
50 }

(某一次)运行结果：

java.util.concurrent.RejectedExecutionException
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:1774)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:768)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:656)
    at AbortPolicyDemo.main(AbortPolicyDemo.java:27)
task-0 is running.
task-1 is running.

结果说明：将”线程池的拒绝策略”由DiscardPolicy修改为AbortPolicy之后，当有任务添加到线程池被拒绝时，会抛出RejectedExecutionException。

4. CallerRunsPolicy 示例

 1 import java.lang.reflect.Field;
 2 import java.util.concurrent.ArrayBlockingQueue;
 3 import java.util.concurrent.ThreadPoolExecutor;
 4 import java.util.concurrent.TimeUnit;
 5 import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
 6 
 7 public class CallerRunsPolicyDemo {
 8 
 9     private static final int THREADS_SIZE = 1;
10     private static final int CAPACITY = 1;
11 
12     public static void main(String[] args) throws Exception {
13 
14         // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
15         ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
16                 new ArrayBlockingQueue<Runnable>(CAPACITY));
17         // 设置线程池的拒绝策略为"CallerRunsPolicy"
18         pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
19 
20         // 新建10个任务，并将它们添加到线程池中。
21         for (int i = 0; i < 10; i++) {
22             Runnable myrun = new MyRunnable("task-"+i);
23             pool.execute(myrun);
24         }
25 
26         // 关闭线程池
27         pool.shutdown();
28     }
29 }
30 
31 class MyRunnable implements Runnable {
32     private String name;
33     public MyRunnable(String name) {
34         this.name = name;
35     }
36     @Override
37     public void run() {
38         try {
39             System.out.println(this.name + " is running.");
40             Thread.sleep(100);
41         } catch (Exception e) {
42             e.printStackTrace();
43         }
44     }
45 }

(某一次)运行结果：

task-2 is running.
task-3 is running.
task-4 is running.
task-5 is running.
task-6 is running.
task-7 is running.
task-8 is running.
task-9 is running.
task-0 is running.
task-1 is running.

结果说明：将”线程池的拒绝策略”由DiscardPolicy修改为CallerRunsPolicy之后，当有任务添加到线程池被拒绝时，线程池会将被拒绝的任务添加到”线程池正在运行的线程”中取运行。

Java多线程系列–“JUC线程池”06之 Callable和Future

概要

本章介绍线程池中的Callable和Future。
Callable 和 Future 简介
 示例和源码分析(基于JDK1.7.0_40)

转载请注明出处：http://www.cnblogs.com/skywang12345/p/3544116.html

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)

我们先通过一个示例看看Callable和Future的基本用法，然后再分析示例的实现原理。

 1 import java.util.concurrent.Callable;
 2 import java.util.concurrent.Future;
 3 import java.util.concurrent.Executors;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.ExecutionException;
 6 
 7 class MyCallable implements Callable {
 8 
 9     @Override 
10     public Integer call() throws Exception {
11         int sum    = 0;
12         // 执行任务
13         for (int i=0; i<100; i++)
14             sum += i;
15         //return sum; 
16         return Integer.valueOf(sum);
17     } 
18 }
19 
20 public class CallableTest1 {
21 
22     public static void main(String[] args) 
23         throws ExecutionException, InterruptedException{
24         //创建一个线程池
25         ExecutorService pool = Executors.newSingleThreadExecutor();
26         //创建有返回值的任务
27         Callable c1 = new MyCallable();
28         //执行任务并获取Future对象 
29         Future f1 = pool.submit(c1);
30         // 输出结果
31         System.out.println(f1.get()); 
32         //关闭线程池 
33         pool.shutdown(); 
34     }
35 }

运行结果：

结果说明：
　　在主线程main中，通过newSingleThreadExecutor()新建一个线程池。接着创建Callable对象c1，然后再通过pool.submit(c1)将c1提交到线程池中进行处理，并且将返回的结果保存到Future对象f1中。然后，我们通过f1.get()获取Callable中保存的结果；最后通过pool.shutdown()关闭线程池。

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)保存的值。

    原文作者：JUC
    原文地址: https://blog.csdn.net/Together_CZ/article/details/68960522
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。