正文
简单明了的Java线程池
小程序：扫一扫查出行【扫一扫了解最新限行尾号】
复制小程序
线程池
线程池从功能上来看，就是一个任务管理器。在Java中，Executor接口是线程池的根接口，其中只包含一个方法：
Executor
void execute(Runnable command); // 执行任务
ExecutorService继承了Executor接口，提供了一些线程池的基础方法：
void shutdown(); // 关闭线程池（不接受新任务，但是旧任务会执行）
List<Runnable> shutdownNow(); // 关闭线程池，返回待执行任务
boolean isShutdown(); // 线程池是否会关闭
boolean isTerminated(); // 关闭之前所有任务是否被关闭。（必须先执行过shutdown）
....
再往下是两种线程池的实现：ThreadPoolExecutor和ForkJoinPool。ThreadPoolExecutor中维护了一个BlockingQueue阻塞队列保存所有的待执行任务，而ForkJoinPool中每一个线程都有自己的BlockingQueue用来存储任务。
ThreadPoolExecutor
在ThreadPoolExecutor的构造方法中，需要提供几个参数：corePoolSize、maximumPoolSize、keepAliveTime、BlockingQueue、RejectedExecutionHandler。其中corePoolSize表示当前线程池维护几个线程，maximumPoolSize表示允许的最大线程数。keepAliveTime表示如果当前线程数在corePoolSize和maximumPoolSize之间时，允许在多久时间内保持存活等待新任务。BlockingQueue是保存任务的阻塞队列，RejectedExecutionHandler是不同的拒绝策略。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
那么
ThreadPoolExecutor中是如何创建新线程呢？
在接到请求执行一个新任务时，首先会判断当前线程数是否大于corePoolSize，如果没有则创建新线程。否则将当前任务放到阻塞队列中，如果当前队列已满，则创建新的线程执行任务。在成功将当前任务放到队列中之后，我们还需要二次判断当前线程池中是否有线程已经销毁或者当前线程池停止运行。当线程数量大于maximumPoolSize时，执行拒绝策略。
这段代码在ThreadPoolExecutor的excute方法中体现，其中也有详细地对执行任务顺序的描述。
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();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
如何设置线程池的线程数量：如果是CPU密集型应用，则线程大小为N或者N+1；如果是IO密集型应用，则线程大小为2N或者2N+2
ForkJoinPool
ForkJoinPool和ThreadPoolExecutor都是继承自AbstractExecutorService抽象类，所以它和ThreadPoolExecutor的使用几乎没有多少区别，除了任务变成了ForkJoinTask以外。
ForkJoinPool和ThreadPoolExecutor最主要的区别就是ForkJoinPool中每一个线程都有属于自己的队列，当某个线程队列任务全部执行完了时，会通过"窃取工作"从别的线程队列中取出一个任务进行执行。
具体的策略就是每一个线程维护一个自己的队列，先进后出（FILO）将任务塞到队列的头部，执行任务时从队列头部取出任务执行。其他线程从队列尾部窃取任务执行。减少阻塞消耗，特别适用于计算型任务。
Callable和Future
在jvm内存模型中我们会发现，每个线程有自己的内存空间，而且线程的run方法返回空值。这个时候就会出现一个问题，如果想要在其他线程中拿到当前线程运行的结果是不可能的。所以就有了Callable和Future的存在，Callable可以让线程返回值，而Future可以拿到线程的返回值。
这里有一个误区，Callable的使用不是在线程的run方法中将返回值传递给call方法，再从future.get()中取值。实际上Callable和Future的配合使用，是利用了一个叫做FutureTask的类，这个类同时继承了Runnable和Future接口，初始化的构造函数中会接受一个Callable对象或者将Runnable对象封装到Callable对象中。在他的run方法中，会调用call方法。
public void run() {
if (state != NEW ||
!RUNNER.compareAndSet(this, null, Thread.currentThread()))
return;
try {
Callable<V> c = callable; // 获取传入的callable对象
if (c != null && state == NEW) {
V result; // call返回的结果
boolean ran;
try {
result = c.call(); // 执行call方法中的任务
ran = true; // 成功执行
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result); // 返回结果值
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
实际上这个set(result)会将返回的结果塞到private Object outcome对象中，这个就是我们最终通过future.get()获取到的值。
RejectedExecutionHandler
上述线程池中我们提到拒绝策略，就是在线程池满时拒绝添加新线程，执行拒绝策略，就是通过这个RejectedExecutionHandler进行处理。在ThreadPoolExecutor线程池中，定义了4种不同的拒绝策略，他们都继承了RejectedExecutionHandler。
- AbortPolicy
这个是默认的拒绝策略，他将丢弃当前的任务，并抛出异常。
/**
* A handler for rejected tasks that throws a
* {@link RejectedExecutionException}.
*
* This is the default handler for {@link ThreadPoolExecutor} and
* {@link ScheduledThreadPoolExecutor}.
*/
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());
}
}
他对当前的任务r没有做任何处理。这是默认的拒绝策略，没有固定的使用场景，但是有一点需要注意，Executors中提供的几种线程池的队列都是无界的，所以不会触发拒绝策略。
- DiscardPolicy
丢弃当前任务，不抛出异常。
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler { public DiscardPolicy() { } /**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
他既没有执行任务，也没有抛出异常。这种拒绝策略一般是你的任务无关紧要时使用，因为他不会返回异常。
- DiscardOldestPolicy
将当前队列头（即将被执行的任务）丢弃，执行当前任务。
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler { public DiscardOldestPolicy() { } /**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
在这个拒绝策略中，他将当前队列头丢弃，调用e.execute(r)重新尝试执行当前任务，当然他也会悄无声息的丢弃任务。这种一般是新任务的优先度更高时使用，比如说新消息来了，那么旧消息就无关紧要了。
- CallerRunsPolicy
在当前线程下运行该任务。
/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler { public CallerRunsPolicy() { } /**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
在当前线程执行该任务，相当于做了一个阻塞，当前线程暂停接受新任务并且当前线程不处于空闲状态，但是如果该任务执行时间过长可能会导致其他线程等待。这种一般是在不允许失败的、并发量较小的情况下使用。
工具类Executors
Executors是线程池工具类和工厂类，最主要的是提供了几种常见的线程池的创建。jdk8以前一共是4种，而在1.8之后添加了forkjoin连接池。
newSingleThreadExecutor：这是单例的线程池，只包含一条线程，如果该线程意外中断，会创建新线程。newFixedThreadPool：固定数量线程池，由于是固定大小的线程池，所以maximumPoolSize是没有意义的，只会创建corePoolSize个线程。newCachedThreadPool：缓存线程池。corePoolSize大小为0，maximumPoolSize为Integer.MAX_VALUE，60秒内无任务则销毁线程。ScheduledThreadPoolExecutor：定时线程池。可以设置延时时间。
在jdk8之后，Executors添加了一种forkjoinpool线程池：
newWorkStealingPool: 工作窃取线程池，通过ForkJoinPool实现。每个线程有独立的队列，完成之后从别的线程窃取工作。
当然除了除了上述之外，也可以通过ThreadPoolExecutor或者ForkJoinPool的构造函数来自定义线程池。
Semaphore
Semaphore叫做信号量，我们可以通过设置个数来限制同时进入资源的线程数。比如一个停车场有十个停车位，我们设置停车位为10，那么同时只能有10辆车进入停车场。
Semaphore的初始化过程中，我们可以提供两个参数，permits和fair分别表示信号量的个数和是否公平，如果fair为true的话会使用公平锁的机制初始化线程队列同步器（AQS）。公平锁和非公平锁的区别在于公平锁线程等待的时间越长越优先，所以公平锁的吞吐量比非公平锁小，非公平锁可能会造成某个线程等待时间过长。
在Semaphore两种锁的具体实现就是在公平锁的tryAcquireShared方法中多了一句：if (hasQueuedPredecessors()) return -1; 如果当前线程不是排在最前面的话，就返回-1；
Semaphore使用acquire方法来获取信号，通过release方法在结束后释放信号。他的具体实现就是通过AQS中的state状态来保存promits数量，被拿走就减去指定的数量。
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors()) // 判断当前线程是否在队列最前面
return -1;
int available = getState(); // 获取AQS中保存的promits数量
int remaining = available - acquires; // 总promits减去当前要取的信号量
if (remaining < 0 ||
compareAndSetState(available, remaining)) // CAS(compare and swap)乐观锁的方式更换state值
return remaining; // 返回剩下的信号量个数
}
}
可能看到这里会对AQS的概念比较模糊，如果想要进一步了解AQS的话，可以直接跳到AQS的部分，因为接下来的东西很多都跟AQS有关。
CountDownLatch
CountDownLatch一般叫做计数器，他的作用是挂起线程等待其他线程运行到计算器清0之后再继续运行。一般用于流程控制，等待前置线程执行。
他也是通过AQS来进行实现的，构造函数中要求提供一个count值赋值给state，调用countDown之后计数器减1，直到state值为0时，才能获取锁，继续执行下面的任务。
public static void main(String[] args) throws InterruptedException, ExecutionException {
CountDownLatch countDownLatch = new CountDownLatch(2);Runnable taskMain = () -> {
try {
countDownLatch.await(); // 挂起，等待AQS的state值为0时被唤醒，解锁继续执行
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("继续执行任务");
};Runnable task1 = () -> {
countDownLatch.countDown(); // 将AQS的state值减1 (计数器减1)
System.out.println("前置任务1完成");
};Runnable task2 = () -> {
countDownLatch.countDown(); // 将AQS的state值减1 (计数器减1)
System.out.println("前置任务2完成");
};new Thread(taskMain).start();
new Thread(task1).start();
new Thread(task2).start();}
用法也很简单，在前置线程中使用countDown计数减1，在后续线程中使用await等待锁释放。
锁AQS
AbstractQueuedSynchronizer是concurrent包下非常重要的一个接口，ReentranLock、Semaphore、CountDownLatch等工具类的底层都是通过AQS来实现。
那么AQS底层是怎么来进行实现的呢？不同于synchronized直接作用于jvm底层，AQS定义了一个volatile属性的变量state来表示当前资源是否被锁。那么如何在多线程下对一个变量进行修改？这里借用了乐观锁的概念，采用CAS的方式来修改变量。CAS指的是compare and swap，它假设所有线程对资源的访问是没有冲突的，如果有冲突，则通过比较交换的方式来解决。如果要修改，则会传两个值，一个是state的预期值，一个是修改之后的值，如果预期值跟state值一致，那么就允许更改，否则则判断是有别的线程已经对state修改过了，表示锁定状态。这个时候当前线程会被挂起，放到CLH队列中，等待其他线程释放state唤醒。
