简介
但凡有点开发经验的同学都知道,频繁的创建和销毁线程是会给系统带来比较大的性能开销的。所以线程池就营运而生了。那么使用线程池有什么好处呢?
- 降低资源消耗
可以重复利用已创建的线程降低线程创建和销毁造成的消耗。 - 提高响应速度
当任务到达时,任务可以不需要等到线程创建就能立即执行。 - 提高线程的可管理性
线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一分配、调优和监控。
线程池的运行策略原理
线程池所涉及到的接口和类并不是很多,其继承体系也相对简单。相关继承关系如图:
继承关系图
最顶层的接口Executor仅声明了一个方法execute。ExecutorService 接口在其父接口基础上,声明了包含但不限于shutdown、shutdownNow、submit、invokeAll等方法。ScheduledExecutorService接口,则是声明了一些和定时任务相关的方法:schedule、scheduleAtFixedRate等。线程池的核心实现是在ThreadPoolExecutor类中,我们使用Executors调用newFixedThreadPool、newSingleThreadExecutor和newCachedThreadPool等方法创建线程池均是ThreadPoolExecutor类型。
我们看看的构造函数:
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {}
核心参数分析:
2. 线程创建规则策略
对于线程池的创建,线程池所能创建的线程数量受限于corePoolSize和maximumPoolSize两个参数值。线程的创建时机则和corePoolSize以及workQueue 两个参数有关。
下面列举一下线程创建的几个规则(线程池中无空闲线程),如下:
简化一下上面的规则:
《Android开发艺术探索》一书中建议:
a. 核心线程数等于CPU核心数+1;
b. 线程池的最大线程数等于CPU的核心数的2倍+1;
c. 核心线程无超时机制,分核心线程的闲置时间为4秒;
d. 任务队列的容量为128.
- 线程资源回收策略
考虑到系统资源是有限的,对于线程池超出 corePoolSize 数量的空闲线程应进行回收操作。进行此操作存在一个问题,即回收时机。目前的实现方式是当线程空闲时间超过 keepAliveTime 后,进行回收。除了核心线程数之外的线程可以进行回收,核心线程内的空闲线程也可以进行回收。回收的前提是allowCoreThreadTimeOut属性被设置为 true,通过public void allowCoreThreadTimeOut(boolean) 方法可以设置属性值。 - 排队策略
如上面线程创建规则所说的,当线程数量大于等于corePoolSize,workQueue未满时,则缓存新任务。这里要考虑使用什么类型的容器缓存新任务,通过 JDK 文档介绍,我们可知道有3种类型的容器可供使用,分别是同步队列,有界队列和无界队列。对于有优先级的任务,这里还可以增加优先级队列。以上所介绍的4种类型的队列,对应的实现类如下:
5. 拒绝策略
如上线程创建规则策略中所说,当线程数量大于等于 maximumPoolSize,且 workQueue 已满,或者是当前线程池被关闭了则使用拒绝策略处理新任务。Java 线程池提供了4种拒绝策略实现类, 如下:
以上4个拒绝策略中,AbortPolicy 是线程池实现类所使用的默认策略。我们也可以通过方法:
public void setRejectedExecutionHandler(RejectedExecutionHandler handler)
- 修改线程池的拒绝策略。
线程池中的线程是如何创建的
在线程池的实现上,线程的创建是通过线程工厂接口ThreadFactory的实现类来完成的。默认情况下,线程池使用Executors.defaultThreadFactory()方法返回的线程工厂实现类。当然,我们也可以通过方法:
public void setThreadFactory(ThreadFactory threadFactory)
进行动态修改线程的创建。具体细节可以参考具体细节可以参考Executors.defaultThreadFactory()的实现。
- 线程池的线程是如何实现复用的
在线程池中,线程的复用是线程池的关键所在。这就要求线程在执行完一个任务后,不能立即退出。对应到具体实现上,工作线程在执行完一个任务后,会再次到任务队列获取新的任务。如果任务队列中没有任务,且 keepAliveTime 也未被设置,工作线程则会被一致阻塞下去。通过这种方式即可实现线程复用。
说完原理,再来看看线程的创建和复用的相关代码(基于 JDK 1.8),如下:
+----ThreadPoolExecutor.Worker.javaWorker(Runnable firstTask) {setState(-1);this.firstTask = firstTask;// 调用线程工厂创建线程this.thread = getThreadFactory().newThread(this);
}// Worker 实现了 Runnable 接口
public void run() {runWorker(this);
}+----ThreadPoolExecutor.java
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock();boolean completedAbruptly = true;try {// 循环从任务队列中获取新任务while (task != null || (task = getTask()) != null) {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 interruptif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();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 {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {// 线程退出后,进行后续处理processWorkerExit(w, completedAbruptly);}
}
- 提交任务
有两种提交任务的方式execute(Runnable command)和submit(Runnable task)。而他们的区别是什么呢?
AbstractExecutorService.javapublic Future<?> submit(Runnable task) {if (task == null) throw new NullPointerException();// 创建任务RunnableFuture<Void> ftask = newTaskFor(task, null);// 提交任务execute(ftask);return ftask;
}
最直观的是execute(Runnable command)没有返回值,而submit(Runnable task)返回一个Future。从源码上看submit(Runnable task)最终也是执行了execute(Runnable command)。但是具体是啥区别?
submit在执行过程中与execute不一样,不会抛出异常而是把异常保存在成员变量中,在FutureTask.get阻塞获取的时候再把异常抛出来。通过Future可以很轻易地获得任务的执行情况,比如是否执行完成、是否被取消、是否异常等等。
+---- ThreadPoolExecutor.java
public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();// 如果工作线程数量 < 核心线程数,则创建新线程if (workerCountOf(c) < corePoolSize) {// 添加工作者对象if (addWorker(command, true))return;c = ctl.get();}// 缓存任务,如果队列已满,则 offer 方法返回 false。否则,offer 返回 trueif (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 添加工作者对象,并在 addWorker 方法中检测线程数是否小于最大线程数else if (!addWorker(command, false))// 线程数 >= 最大线程数,使用拒绝策略处理任务reject(command);
}private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);// 检测工作线程数与核心线程数或最大线程数的关系if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (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;mainLock.lock();try {int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();// 将 worker 对象添加到 workers 集合中workers.add(w);int s = workers.size();// 更新 largestPoolSize 属性if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 开始执行任务t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}
代码略多,不过结合线程的创建策略和拒绝策略加上注释,理解主逻辑应该不难。
- 关闭线程池
我们可以通过shutdown和shutdownNow两个方法关闭线程池。两个方法的区别在于,shutdown 会将线程池的状态设置为SHUTDOWN,同时该方法还会中断空闲线程。shutdownNow 则会将线程池状态设置为STOP,并尝试中断所有的线程。中断线程使用的是Thread.interrupt方法,未响应中断方法的任务是无法被中断的。最后,shutdownNow 方法会将未执行的任务全部返回。
调用 shutdown 和 shutdownNow 方法关闭线程池后,就不能再向线程池提交新任务了。对于处于关闭状态的线程池,会使用拒绝策略处理新提交的任务。
Android中的几种线程池
一般情况下,我们并不直接使用 ThreadPoolExecutor 类创建线程池,而是通过 Executors 工具类去构建线程池。通过 Executors 工具类,我们可以构造5中不同的线程池。下面通过一个表格简单介绍一下几种线程池,如下:
阿里巴巴Android开发手册对线程池使用的建议
- 【推荐】ThreadPoolExecutor 设置线程存活时间(setKeepAliveTime),确保空闲时
线程能被释放。 - 【强制】线程池不允许使用Executors 去创建,而是通过ThreadPoolExecutor 的方式,这样的处理方式让写的同学更加明确线程池的运行规则,规避资源耗尽的风险。
说明:
Executors 返回的线程池对象的弊端如下:
- FixedThreadPool 和SingleThreadPool :允许的请求队列长度为
Integer.MAX_VALUE,可能会堆积大量的请求,从而导致OOM; - CachedThreadPool 和ScheduledThreadPool :允许的创建线程数量为
Integer.MAX_VALUE,可能会创建大量的线程,从而导致OOM。
正例:int NUMBER_OF_CORES = Runtime.getRuntime().availableProcessors();
int KEEP_ALIVE_TIME = 1;
TimeUnit KEEP_ALIVE_TIME_UNIT = TimeUnit.SECONDS;
BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<Runnable>();
ExecutorService executorService = new ThreadPoolExecutor(NUMBER_OF_CORES,
NUMBER_OF_CORES*2, KEEP_ALIVE_TIME, KEEP_ALIVE_TIME_UNIT,
taskQueue, new BackgroundThreadFactory(), new DefaultRejectedExecutionHandler());反例:
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();