为什么使用线程池
创建线程可以继承Thread类或者实现Runnable接口，根据线程的生命周期，这两种方式创建的线程在运行结束后会被虚拟机销毁，进行垃圾回收，如果线程数量过多，频繁的创建和销毁线程会浪费资源，降低效率，线程池的引入就很好解决了这一问题，线程执行结束后，不立即销毁，而是让线程复用，大大提高了效率。
在实际生产中，阿里开发手册中明确指出，线程资源必须通过线程池提供，不允许在应用中显式的创建线程。如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者“过度切换”的问题。
Executor接口
Executor是线程池的顶层接口，JDK1.5开始引入了，位于java.util.concurrent 包。
public interface Executor {
void execute(Runnable command);
}
查看Executor的类图关系
Executor根接口，它将任务的提交与任务的执行分离开来
ExecutorService子接口，线程池的主要接口，增加了返回Future 对象
ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务
ScheduledExecutorService继承ExecutorService接口，定义延迟或定期执行的方法
ScheduledThreadPoolExecutor继承ThreadPoolExecutor，在给定的延迟之后运行任务或定期执行任务
ThreadPoolExecutor
ThreadPoolExecutor是线程池的核心实现类，这里分析一下其使用方法和源码。
构造方法
如何利用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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
参数含义：
- corePoolSize：核心线程池的大小
- maximumPoolSize：最大线程池的大小
- keepAliveTime：当线程池中线程数大于corePoolSize，并且没有可执行任务时大于corePoolSize那部分线程的存活时间
- unit：keepAliveTime的时间单位
- workQueue：用来暂时保存任务的工作队列
- threadFactory：传入一个线程工厂
- handler：当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时（达到了最大线程池大小且工作队列已满），execute()方法将要调用的Handler。
Executors类
Executors是Executor框架的工具类，提供了几种线程池创建方法，但阿里开发手册中是不允许使用Executors创建线程池，简单分析。
- SingleThreadExecutors
使用Executors.newSingleThreadExecutor()创建，查看其创建过程
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
- corePoolSize和maximumPoolSize都是1；
- LinkedBlockingQueue是一个最大值为Integer.MAX_VALUE的无界队列；
- 当线程正在执行任务，新任务会被加入到LinkedBlockingQueue队列中，任务加入队列的速度远大于核心线程处理的能力时，无界队列会一直增大到最大值，可能导致OOM。
- FixedThreadPool
使用Executors.newFixedThreadPool() 创建，查看其创建过程
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
- corePoolSize和maximumPoolSize值均为nTreads；
- 存活时间为0L，超过核心线程数的空闲线程会被立即销毁；
- 队列依然为LinkedBlockingQueue，当线程数达到corePoolSize时，新任务会一直在无界队列中等待
- 线程池中的线程数不会超过corePoolSize，新建任务也会一直被加入到队列等待，不会执行拒绝策略。
- ThreadPoolExecutor中的7个参数，maximumPoolSize，keepAliveTime，RejectedExecutionHandler为无效参数。
- CachedThreadPool
使用Executors.newCachedThreadPool()创建，查看其创建过程
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
- corePoolSize为0，maximumPoolSize为Integer.MAX_VALUE，即maximumPool是无界的；
- keepAliveTime为60L，空闲线程等待新任务的最长时间为60秒，超过60秒后将会被终止；
- SynchronousQueue为线程池的工作队列，没有容量，maximumPool是无界的，如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，会不断创建新线程，最终导致创建过多线程而耗尽CPU和内存资源。
- 总结
因此在创建线程池的时候要根据业务需求，应用场景合理的规划线程池的参数，使用ThreadPoolExecutor来创建线程池。
线程池状态
查看ThreadPoolExecutor源码，线程池有五种状态
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;
任务单元Worker
ThreadPoolExecutor中核心任务单元是由一个Worker内部类来实现，Worker类中定义了两个重要方法runWorker方法和addWorker方法。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// 省略代码...
}
addWorker和runWorker
- addWorker用来实例化任务单元Worker对象
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 外层自旋
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
// (rs > SHUTDOWN) ||
// (rs == SHUTDOWN && firstTask != null) ||
// (rs == SHUTDOWN && workQueue.isEmpty())
// 1. 线程池状态大于SHUTDOWN时，直接返回false
// 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
// 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内层自旋
for (;;) {
int wc = workerCountOf(c);
// worker数量超过容量，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 使用CAS的方式增加worker数量。
// 若增加成功，则直接跳出外层循环进入到第二部分
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 线程池状态发生变化，对外层循环进行自旋
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 {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
// worker的添加必须是串行的，因此需要加锁
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 这儿需要重新检查线程池状态
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// worker已经调用过了start()方法，则不再创建worker
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// worker创建并添加到workers成功
workers.add(w);
// 更新`largestPoolSize`变量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 启动worker线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
- runWorker是核心线程执行逻辑
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 调用unlock()是为了让外部可以中断
w.unlock(); // allow interrupts
// 这个变量用于判断是否进入过自旋（while循环）
boolean completedAbruptly = true;
try {
// 这儿是自旋
// 1. 如果firstTask不为null，则执行firstTask；
// 2. 如果firstTask为null，则调用getTask()从队列获取任务。
// 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
while (task != null || (task = getTask()) != null) {
// 这儿对worker进行加锁，是为了达到下面的目的
// 1. 降低锁范围，提升性能
// 2. 保证每个worker执行的任务是串行的
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 如果线程池正在停止，则对当前线程进行中断操作
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
// 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
// 这两个方法在当前类里面为空实现。
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
// 帮助gc
task = null;
// 已完成任务数加一
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 自旋操作被退出，说明线程池正在结束
processWorkerExit(w, completedAbruptly);
}
}
submit和execute
ThreadPoolExecutor执行任务有submit和execute两种方法，这两种方法区别在于
- submit方法有返回值，便于异常处理
- execute方法没有返回值
下面来简单介绍一下submit和execute的用法
-
submit方法有三种传入参数的形式
<T> Future<T> submit(Callable<T> callable); <T> Future<T> submit(Runnable var1, T result); Future<?> submit(Runnable runnable);在ExecutorService接口中定义submit方法，抽象类AbstractExecutorService实现了ExecutorService中的submit方法，
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; } /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); execute(ftask); return ftask; } /** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }当submit方法传入Runnable对象调用Future对象的get方法返回值为null，传入Callable对象时返回get自定义的值，在返回结果之前，主线程会阻塞等待结果返回再执行。
class RunnableDemo implements Runnable{ @Override public void run() { System.out.println("RunableDemo is execute"); } } class CallableDemo implements Callable<String>{ @Override public String call() throws Exception { return "Call is Done"; } } public class Test { public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newSingleThreadExecutor(); Future<?> call = executorService.submit(new CallableDemo()); System.out.println("Callable'S Result："+call.get()); Future<?> run = executorService.submit(new RunnableDemo()); System.out.println("Runnable'S Result："+run.get()); System.out.println("Current Thread："+Thread.currentThread().getName()); } }输出结果
Callable'S Result：Call is Done RunableDemo is execute Runnable'S Result：null Current Thread：main -
execute方法只有一种，传入实现Runnable接口的对象。
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); // worker数量比核心线程数小，直接创建worker执行任务 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // worker数量超过核心线程数，任务直接进入队列 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。 // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。 if (! isRunning(recheck) && remove(command)) reject(command); // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0 else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。 // 这儿有3点需要注意： // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态 // 2. addWorker第2个参数表示是否创建核心线程 // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作 else if (!addWorker(command, false)) reject(command); }使用submit方法可以对task执行的结果成功，失败，或者执行过程中抛出的异常及时处理，暂停处理其他task，使用execute不能及时处理程序在运行中出现的异常情况。
class CallableDemo implements Callable<String>{ @Override public String call() throws Exception { return "Call is Done"; } } public class Test { public static void main(String[] args) { ExecutorService executorService = Executors.newSingleThreadExecutor(); Future<?> call = executorService.submit(new CallableDemo()); try { call.get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } }
我们可以根据具体业务场景考虑可能出现的异常，由实现Callable的接口throws，然后由ThreadPoolExecutor调用者来处理，提高多线程场景下的容错率。
ScheduledThreadPoolExecutor
构造方法
ScheduledThreadPoolExecutor继承自ThreadPoolExecutor。它主要用来在给定的延迟之后运行任务，或者定期执行任务。ScheduledThreadPoolExecutor的功能与Timer类似，但ScheduledThreadPoolExecutor功能更强大、更灵活。Timer对应的是单个后台线程，而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数。
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
核心内部类
ScheduledThreadPoolExecutor有两个内部类DelayedWorkQueue和ScheduledFutureTask。
查看源码DelayedWorkQueue实现了BlockingQueue接口，也是一个阻塞队列，ScheduledFutureTask继承了FutureTask类，表示该类用于返回异步任务的结果。
static class DelayedWorkQueue
extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {}
private class ScheduledFutureTask<V>
extends FutureTask<V> implements RunnableScheduledFuture<V> {}
调度方法
可延时执行异步任务和可周期执行异步任务方法
/**这里传入的是实现Runnable接口的任务*/
public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);
/**这里传入的是实现Callable接口的任务*/
public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit);
/**在initialDelay之后以上一个任务开始的时间计时，period时间过去后，检测上一个任务是否执行完毕，如果上一个任务执行完毕，则当前任务立即执行，如果上一个任务没有执行完毕，则需要等上一个任务执行完毕后立即执行*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
/**是以上一个任务结束时开始计时，period时间过去后，立即执行*/
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
总结
Executor框架主要由三部分组成，任务，任务的执行者，执行结果，ThreadPoolExecutor和ScheduledThreadPoolExecutor的设计思想也是将这三个关键要素进行了解耦，将任务的提交和执行分离。
-
任务
在
ThreadPoolExecutor和ScheduledThreadPoolExecutor中任务是指实现了Runnable接口和Callable接口的类，ThreadPoolExecutor中将任务转换成FutureTask类，ScheduledThreadPoolExecutor中任务被转换成ScheduledFutureTask类，该类继承FutureTask，并重写了run方法，实现了延时执行任务和周期性执行任务。 -
任务的执行者
包括任务执行机制的核心接口
Executor，以及继承自Executor的ExecutorService接口和两个关键类（实现了ExecutorService接口的ThreadPoolExecutor和ScheduledThreadPoolExecutor类）。任务的执行机制，交由Worker类，进一步封装了Thread向线程池提交任务，ThreadPoolExecutor的execute方法和submit方法，以及ScheduledThreadPoolExecutor的schedule方法都是先将任务移到阻塞队列中，然后通过addWorker方法新建Worker对象，并通过runWorker方法启动线程，不断的从阻塞对列中获取异步任务交给Worker执行，直至阻塞队列中任务执行完为止。 -
执行结果
包括接口
Future和实现Future接口的FutureTask类，获取任务执行结果，在ThreadPoolExecutor中提交任务后实际上为FutureTask类，在ScheduledThreadPoolExecutor中则是ScheduledFutureTask类。
参考资料
Java并发编程的艺术
并发编程网：http://ifeve.com/
京公网安备 11010502036488号