Java多线程之线程池（ThreadPoolExecutor）实
2018-05-14 本文已影响540人
小怪聊职场
在上一篇文章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类，以此去深入了解线程池的实现原理。
