在上一篇文章 Java 中实现多线程的 3 种方法介绍和比较中, 我们讲解了 Java 中实现多线程的 3 种方法. 使用多线程, 就必须要考虑使用线程池, 今天我们来聊聊线程池的那些事.
注: 源码都是基于 JDK1.8
一, 为什么要使用线程池?
如果并发的线程数量很多, 并且每个线程都是执行一个时间很短的任务就结束了, 这样频繁创建线程就会大大降低系统的效率, 因为频繁创建线程和销毁线程需要时间.
那么有没有一种办法使得线程可以复用, 就是执行完一个任务, 并不被销毁, 而是可以继续执行其他的任务?
在 Java 中可以通过线程池来达到这样的效果. 今天我们就来详细讲解一下 Java 的线程池, 首先我们从最核心的 ThreadPoolExecutor 类中的方法讲起, 然后再讲述它的实现原理, 接着给出了它的使用示例, 最后讨论了一下如何合理配置线程池的大小.
二, Java 中的 ThreadPoolExecutor 类
java.uitl.concurrent.ThreadPoolExecutor
类是线程池中最核心的一个类, 因此如果要透彻地了解 Java 中的线程池, 必须先了解这个类. 下面我们来看一下 ThreadPoolExecutor 类的具体实现源码.
在 ThreadPoolExecutor 类中提供了四个构造方法:
- public class ThreadPoolExecutor extends AbstractExecutorService {
- ...
- public ThreadPoolExecutor(int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
- Executors.defaultThreadFactory(), defaultHandler);
- }
- public ThreadPoolExecutor(int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
- threadFactory, defaultHandler);
- }
- public ThreadPoolExecutor(int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
- Executors.defaultThreadFactory(), handler);
- }
- 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;
- }
从上面的代码可以得知, ThreadPoolExecutor 继承了 AbstractExecutorService 类, 并提供了四个构造器, 事实上, 通过观察每个构造器的源码具体实现, 发现前面三个构造器都是调用的第四个构造器进行的初始化工作.
下面解释下一下构造器中各个参数的含义:
corePoolSize: 核心池的大小.
maximumPoolSize: 线程池最大线程数, 它表示在线程池中最多能创建多少个线程, 注意与 corePoolSize 区分, 后面会讲到.
keepAliveTime: 表示线程没有任务执行时最多保持多久时间会终止.
unit: 参数 keepAliveTime 的时间单位, 有 7 种取值, 在 TimeUnit 类中有 7 种静态属性:
- /**
- * Time unit representing one thousandth of a microsecond
- */
- NANOSECONDS {
- public long toNanos(long d) { return d; }
- public long toMicros(long d) { return d/(C1/C0); }
- public long toMillis(long d) { return d/(C2/C0); }
- public long toSeconds(long d) { return d/(C3/C0); }
- public long toMinutes(long d) { return d/(C4/C0); }
- public long toHours(long d) { return d/(C5/C0); }
- public long toDays(long d) { return d/(C6/C0); }
- public long convert(long d, TimeUnit u) { return u.toNanos(d); }
- int excessNanos(long d, long m) { return (int)(d - (m*C2)); }
- },
- /**
- * Time unit representing one thousandth of a millisecond
- */
- MICROSECONDS {
- public long toNanos(long d) { return x(d, C1/C0, MAX/(C1/C0)); }
- public long toMicros(long d) { return d; }
- public long toMillis(long d) { return d/(C2/C1); }
- public long toSeconds(long d) { return d/(C3/C1); }
- public long toMinutes(long d) { return d/(C4/C1); }
- public long toHours(long d) { return d/(C5/C1); }
- public long toDays(long d) { return d/(C6/C1); }
- public long convert(long d, TimeUnit u) { return u.toMicros(d); }
- int excessNanos(long d, long m) { return (int)((d*C1) - (m*C2)); }
- },
- /**
- * Time unit representing one thousandth of a second
- */
- MILLISECONDS {
- public long toNanos(long d) { return x(d, C2/C0, MAX/(C2/C0)); }
- public long toMicros(long d) { return x(d, C2/C1, MAX/(C2/C1)); }
- public long toMillis(long d) { return d; }
- public long toSeconds(long d) { return d/(C3/C2); }
- public long toMinutes(long d) { return d/(C4/C2); }
- public long toHours(long d) { return d/(C5/C2); }
- public long toDays(long d) { return d/(C6/C2); }
- public long convert(long d, TimeUnit u) { return u.toMillis(d); }
- int excessNanos(long d, long m) { return 0; }
- },
- /**
- * Time unit representing one second
- */
- SECONDS {
- public long toNanos(long d) { return x(d, C3/C0, MAX/(C3/C0)); }
- public long toMicros(long d) { return x(d, C3/C1, MAX/(C3/C1)); }
- public long toMillis(long d) { return x(d, C3/C2, MAX/(C3/C2)); }
- public long toSeconds(long d) { return d; }
- public long toMinutes(long d) { return d/(C4/C3); }
- public long toHours(long d) { return d/(C5/C3); }
- public long toDays(long d) { return d/(C6/C3); }
- public long convert(long d, TimeUnit u) { return u.toSeconds(d); }
- int excessNanos(long d, long m) { return 0; }
- },
- /**
- * Time unit representing sixty seconds
- */
- MINUTES {
- public long toNanos(long d) { return x(d, C4/C0, MAX/(C4/C0)); }
- public long toMicros(long d) { return x(d, C4/C1, MAX/(C4/C1)); }
- public long toMillis(long d) { return x(d, C4/C2, MAX/(C4/C2)); }
- public long toSeconds(long d) { return x(d, C4/C3, MAX/(C4/C3)); }
- public long toMinutes(long d) { return d; }
- public long toHours(long d) { return d/(C5/C4); }
- public long toDays(long d) { return d/(C6/C4); }
- public long convert(long d, TimeUnit u) { return u.toMinutes(d); }
- int excessNanos(long d, long m) { return 0; }
- },
- /**
- * Time unit representing sixty minutes
- */
- HOURS {
- public long toNanos(long d) { return x(d, C5/C0, MAX/(C5/C0)); }
- public long toMicros(long d) { return x(d, C5/C1, MAX/(C5/C1)); }
- public long toMillis(long d) { return x(d, C5/C2, MAX/(C5/C2)); }
- public long toSeconds(long d) { return x(d, C5/C3, MAX/(C5/C3)); }
- public long toMinutes(long d) { return x(d, C5/C4, MAX/(C5/C4)); }
- public long toHours(long d) { return d; }
- public long toDays(long d) { return d/(C6/C5); }
- public long convert(long d, TimeUnit u) { return u.toHours(d); }
- int excessNanos(long d, long m) { return 0; }
- },
- /**
- * Time unit representing twenty four hours
- */
- DAYS {
- public long toNanos(long d) { return x(d, C6/C0, MAX/(C6/C0)); }
- public long toMicros(long d) { return x(d, C6/C1, MAX/(C6/C1)); }
- public long toMillis(long d) { return x(d, C6/C2, MAX/(C6/C2)); }
- public long toSeconds(long d) { return x(d, C6/C3, MAX/(C6/C3)); }
- public long toMinutes(long d) { return x(d, C6/C4, MAX/(C6/C4)); }
- public long toHours(long d) { return x(d, C6/C5, MAX/(C6/C5)); }
- public long toDays(long d) { return d; }
- public long convert(long d, TimeUnit u) { return u.toDays(d); }
- int excessNanos(long d, long m) { return 0; }
- };
workQueue: 一个阻塞队列, 用来存储等待执行的任务.
threadFactory: 线程工厂, 主要用来创建线程.
handler: 表示当拒绝处理任务时的策略.
从源码可以得知 ThreadPoolExecutor 继承了 AbstractExecutorService, 我们看下 AbstractExecutorService 的实现:
- public abstract class AbstractExecutorService implements ExecutorService {
- protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
- return new FutureTask<T>(runnable, value);
- }
- protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
- return new FutureTask<T>(callable);
- }
- public Future<?> submit(Runnable task) {
- if (task == null) throw new NullPointerException();
- RunnableFuture<Void> ftask = newTaskFor(task, null);
- execute(ftask);
- return ftask;
- }
- 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;
- }
- public <T> Future<T> submit(Callable<T> task) {
- if (task == null) throw new NullPointerException();
- RunnableFuture<T> ftask = newTaskFor(task);
- execute(ftask);
- return ftask;
- }
- private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
- ...
- }
- public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
- throws InterruptedException, ExecutionException {
- try {
- return doInvokeAny(tasks, false, 0);
- } catch (TimeoutException cannotHappen) {
- assert false;
- return null;
- }
- }
- public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
- return doInvokeAny(tasks, true, unit.toNanos(timeout));
- }
- public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
- throws InterruptedException {
- ...
- }
- public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
- throws InterruptedException {
- ...
- }
AbstractExecutorService 是一个抽象类, 它实现了 ExecutorService 接口, 我们看下 ExecutorService 接口的实现:
- 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;
}
而 ExecutorService 又是继承了 Executor 接口, 我们看一下 Executor 接口的实现:
- public interface Executor {
- void execute(Runnable command);
- }
到这里, 大家应该明白了 ThreadPoolExecutor,AbstractExecutorService,ExecutorService 和 Executor 几个之间的关系了.
1,Executor 是一个顶层接口, 在它里面只声明了一个方法 execute(Runnable), 返回值为 void, 参数为 Runnable 类型, 从字面意思可以理解, 就是用来执行传进去的任务的.
2, 然后 ExecutorService 接口继承了 Executor 接口, 并声明了一些方法: submit,invokeAll,invokeAny 以及 shutDown 等;
3, 抽象类 AbstractExecutorService 实现了 ExecutorService 接口, 基本实现了 ExecutorService 中声明的所有方法;
4, 然后 ThreadPoolExecutor 继承了类 AbstractExecutorService.
在 ThreadPoolExecutor 类中有几个非常重要的方法:
- 1,
- public void execute(Runnable command)
- 2,
- public void shutdown()
- 3,
- public List<Runnable> shutdownNow()
- 4, submit
- public Future<?> submit(Runnable task)
- public <T> Future<T> submit(Runnable task, T result)
- public <T> Future<T> submit(Callable<T> task)
execute() 方法实际上是 Executor 中声明的方法, 在 ThreadPoolExecutor 进行了具体的实现, 这个方法是 ThreadPoolExecutor 的核心方法, 通过这个方法可以向线程池提交一个任务, 交由线程池去执行.
shutdown() 和 shutdownNow() 是用来关闭线程池的.
submit() 方法是在 ExecutorService 中声明的方法, 在 AbstractExecutorService 就已经有了具体的实现, 在 ThreadPoolExecutor 中并没有对其进行重写, 这个方法也是用来向线程池提交任务的, 但是它和 execute() 方法不同, 它能够返回任务执行的结果, 去看 submit() 方法的实现, 会发现它实际上还是调用的 execute() 方法, 只不过它利用了 Future 来获取任务执行结果.
本文只对 ThreadPoolExecutor 类做一个宏观的介绍, 下一篇文章将会深入剖析 ThreadPoolExecutor 类, 以此去深入了解线程池的实现原理.
来源: http://www.jianshu.com/p/96fc74e43088