1.线程池的流程
image.png
image.png
首先要掌握一个最基本的线程池的几个概念：
任务队列：这个就是 list<job>，job extends runnable，是用来存放提交的任务的
工作者队列，就是 list<worker>, Worker implements Runnable，即线程池里面的工作线程，用来处理任务
关键点是：当工作者取任务队列中的job时，要加锁，防止重复取，或者其他，取出对象后，既可以放开锁，执行取出的任务
（这里有一个全局锁的过程，后面会优化）
2.在最基本的线程池实现上，进行优化
优化的主要目的当然是为了加快实际的并行能力和处理一些极端情况的发生情况
2.1 coolPoolSize
当线程池大小等于coolPoolSize时，会将任务加入任务队列，知道加不进的，才会额外创建新线程
原因：是为了在执行execute()方法时，尽可能地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。
2.2 创建线程池的参数
看图中的，我们可以将每一部分都作为构造的参数
1）corePoolSize（线程池的基本大小）
2）runnableTaskQueue（任务队列），是否有界，是否有优先级等
3）maximumPoolSize（线程池最大数量）
4）ThreadFactory：用于设置创建线程的工厂：，可以通过线程工厂给每个创建出来的线程设
置更有意义的名字
5）RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状
态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法
处理新任务时抛出异常。在JDK 1.5中Java线程池框架提供了以下4种策略。
·AbortPolicy：直接抛出异常。
·CallerRunsPolicy：只用调用者所在线程来运行任务。
·DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
·DiscardPolicy：不处理，丢弃掉
6）keepAliveTime（线程活动保持时间）：线程池的工作线程空闲后，保持存活的时间。所以，
如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提高线程的利用率。
3.线程池的使用
可以使用两个方法向线程池提交任务，分别为execute()和submit()方法。前者不需要返回值
submit()方法用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个
future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方
法会阻塞当前线程直到任务完成，而使用get（long timeout，TimeUnit unit）方法则会阻塞当前线
程一段时间后立即返回，这时候有可能任务没有执行完
4.合理配置线程池
可以从以下几个角度来分析。
·任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
·任务的优先级：高、中和低。
·任务的执行时间：长、中和短。
·任务的依赖性：是否依赖其他系统资源，如数据库连接
CPU密集型任务应配置尽可能小的线程，如配置Ncpu+1个线程的线程池
由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*Ncpu。
混合型的任务：如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务
如：依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，等待的时间越长，则CPU空闲时间就越长，那么线程数应该设置得越大，这样才能更好地利用CPU。
5.excutor框架
本质上就是线程池技术的支持，提供了future接口，这里有2个要点，一个是result的返回，另外一个是定时的实现
可以参考futureTask的实现
https://yq.aliyun.com/articles/39088
https://www.jianshu.com/p/ceb8870ef2c5
从普通的线程池-----excutor框架----scheduled excutor框架
6.线程池的异常退出，线程到底去哪了？
我们知道线程的生命周期，有创建-就绪-运行-阻塞-死亡状态，但是实际生产中，由于大量线程池的使用，导致一些细节反而不明白了。
特别是我特别迷惑，当线程池中的线程异常退出时，到底发生了什么？
拿线程池的任务-工作线程模型进行分析线程泄露问题
- 线程泄漏
各种类型的线程池中一个严重的风险是线程泄漏，当从池中除去一个线程以执行一项任务，而在任务完成后该线程却没有返回池时，会发生这种情况。发生线程泄漏的一种情形出现在任务抛出一个 RuntimeException 或一个 Error 时。如果池类没有捕捉到它们，那么线程只会退出而线程池的大小将会永久减少一个。当这种情况发生的次数足够多时，线程池最终就为空，而且系统将停止，因为没有可用的线程来处理任务。
有些任务可能会永远等待某些资源或来自用户的输入，而这些资源又不能保证变得可用，用户可能也已经回家了，诸如此类的任务会永久停止，而这些停止的任务也会引起和线程泄漏同样的问题。如果某个线程被这样一个任务永久地消耗着，那么它实际上就被从池除去了。对于这样的任务，应该要么只给予它们自己的线程，要么只让它们等待有限的时间。
自己写了测试的线程池，在worker的run方法中，catch住job的exception，可以完成线程的重复利用
private volatile boolean running = true;
public void run() {
// TODO Auto-generated method stub
while (running) {
Job job = null;
synchronized (jobs) {
while (jobs.isEmpty()) {
try {
jobs.wait();
} catch (InterruptedException ex) {
// TODO: handle exception
// 感知到外部的中断操作，返回
Thread.currentThread().interrupt();
return;
}
}
job = jobs.removeFirst();
}
if (job != null) {
try {
job.run();
} catch (Exception e) {
// TODO: handle exception
System.out.println("error :"+e);
}
}
}
}
public void shutdown() {
running = false;
}
}
再来看最常使用的Spring的 excutor是如何处理的，查看ThreadPoolTaskExecutor
@Override
public void execute(Runnable task) {
Executor executor = getThreadPoolExecutor();
try {
executor.execute(task);
}
catch (RejectedExecutionException ex) {
throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
}
}
@Override
public void execute(Runnable task, long startTimeout) {
execute(task);
}
而其中的excuteor的execute方法，分为3中情况，核心池不满，核心池满了，超过最大值，看其中
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);
}
看worker的关键代码
...//省略从工作队列，同步取得任务的代码
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
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)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
//转换了好久，才真正执行
if (workerAdded) {
t.start();
workerStarted = true;
}
}
//最终用finally来保证线程不会减少泄露
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
网友评论