1.1 tomcat线程池和juc线程池流程
jdk 线程池策略：
- 当线程池中线程数量小于 corePoolSize，每来一个任务，就会创建一个线程执行这个任务
- 当前线程池线程数量大于等于 corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列中；若是添加成功，则该任务会等待线程将其取出去执行；若添加失败(一般来说任务缓存队列已满)，则会尝试创建新的线程执行
- 当前线程池线程数量等于 maximumPoolSize，则会采取任务拒绝策略进行处理
tomcat 线程池策略：
- 当前线程数小于 corePoolSize，则去创建工作线程
- 当前线程数大于 corePoolSize，但小于 maximumPoolSize，则去创建工作线程
- 当前线程数大于 maximumPoolSize，则将任务放入到阻塞队列中，当阻塞队列满了之后，则调用拒绝策略丢弃任务
1.2 tomcat线程池和juc线程池的区别
tomcat 线程池是在 juc线程池的基础上，修改少量代码来实现，tomcat 线程池执行流程和 juc的线程池执行流程有着很大的区别，那么tomcat为什么要这样设计？
使用线程池的任务有两种：
-
IO 密集型任务(如调用接口、查询数据库)
-
CPU 密集型任务
-
JDK 的线程池 ThreadPoolExecutor 主要目的解决的便是CPU密集型任务的并发处理
-
Tomcat 若使用原生的JDK线程池，一旦接收的请求数量大于线程池的核心线程数，这些请求就会被放到队列中，等待核心线程处理，这样会降低请求的总体处理速度，所以 Tomcat并没有使用JDK原生线程池的策略 (例如：10个核心线程，20个最大线程，15个请求过来的时候，有5个请求放入队列进行等待处理，这也是为什么 tomcat 要使 work 线程优先 queue的原因)
-
JDK线程池：当线程数达到 corePoolSize 后，任务首先被放到 queue，发挥CPU多核的并行优势，减少多个线程导致的上下文切换，适合的场景是：CPU密集型任务
-
Tomcat线程池：当大量请求达到时，接收的请求数量大于核心线程池的 corePoolSize 时，会继续创建 worker 线程去处理请求，而后续请求量变少时，只会销毁 maximumPoolSize 线程数，适合的场景是：IO密集型
1.3 IO密集型和CPU密集型任务核心参数的设置
-
场景：JDK线程池执行IO密集型的任务，解决方案：可以提高corePoolSize的大小，引入问题：系统中线程数过多
-
线程数过多问题的解决方案：指定 ThreadPoolExecutor 的 allowCoreThreadTimeout=true，那么核心线程若处于闲置状态的话，超过一定的时间(KeepAliveTime)，就会销毁掉
-
引入问题：当核心线程数被销毁时，而有大量请求到达时系统重新创建 worker 线程，会使得前期接口响应时间变长(这也是为什么系统上线需要预热)
-
所以JDK线程池无法完美的去处理IO密集型的任务，这也就是为什么Tomcat需要重写JDK线程池的原因
Tomcat线程池执行流程
-
Tomcat 重写 JDK 线程池(即改动的是少量的源码)实现的功能的增强
-
思想：主要流程还是JDK线程池流程，即先开启 corePoolSize 线程，然后在 queue，最后在开启 maximumPoolSize 线程，Tomcat 重点改造的是 queue 的 offer()，即在向 queue 放入任务时，若发现未达到最大线程数，那么 offer() 返回 false，即放入队列失败，此时，便继续开启 maximumPoolSize 线程
自定义队列
Tomcat 主要是通过实现自定义队列来完成逻辑的改造。
/** * 实现Tomcat特有逻辑的自定义队列 */
public class TaskQueue extends LinkedBlockingQueue<Runnable> {
private static final long serialVersionUID = 1L;
private transient volatile ThreadPoolExecutor parent = null;
private static final int DEFAULT_FORCED_REMAINING_CAPACITY = -1;
/** * 强制遗留的容量 */
private int forcedRemainingCapacity = -1;
/** * 队列的构建方法 */
public TaskQueue() {
}
public TaskQueue(int capacity) {
super(capacity);
}
public TaskQueue(Collection<? extends Runnable> c) {
super(c);
}
/** * 设置核心变量 */
public void setParent(ThreadPoolExecutor parent) {
this.parent = parent;
}
/** * put：向阻塞队列填充元素，当阻塞队列满了之后，put时会被阻塞。 * offer：向阻塞队列填充元素，当阻塞队列满了之后，offer会返回false。 * * @param o 当任务被拒绝后，继续强制的放入到线程池中 * @return 向阻塞队列塞任务，当阻塞队列满了之后，offer会返回false。 */
public boolean force(Runnable o) {
if (parent == null || parent.isShutdown()) {
throw new RejectedExecutionException("taskQueue.notRunning");
}
return super.offer(o);
}
/** * 带有阻塞时间的塞任务 */
@Deprecated
public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
if (parent == null || parent.isShutdown()) {
throw new RejectedExecutionException("taskQueue.notRunning");
}
return super.offer(o, timeout, unit); //forces the item onto the queue, to be used if the task is rejected
}
/** * 当线程真正不够用时，优先是开启线程（直至最大线程），其次才是向队列填充任务。 * * @param runnable 任务 * @return false 表示向队列中添加任务失败， */
@Override
public boolean offer(Runnable runnable) {
if (parent == null) {
return super.offer(runnable);
}
//若是达到最大线程数，进队列。
if (parent.getPoolSize() == parent.getMaximumPoolSize()) {
return super.offer(runnable);
}
//当前活跃线程为10个，但是只有8个任务在执行，于是，直接进队列。
if (parent.getSubmittedCount() < (parent.getPoolSize())) {
return super.offer(runnable);
}
//当前线程数小于最大线程数，那么直接返回false，去创建最大线程
if (parent.getPoolSize() < parent.getMaximumPoolSize()) {
return false;
}
//否则的话，将任务放入到队列中
return super.offer(runnable);
}
/** * 获取任务 */
@Override
public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
Runnable runnable = super.poll(timeout, unit);
//取任务超时，会停止当前线程，来避免内存泄露
if (runnable == null && parent != null) {
parent.stopCurrentThreadIfNeeded();
}
return runnable;
}
/** * 阻塞式的获取任务，可能返回null。 */
@Override
public Runnable take() throws InterruptedException {
//当前线程应当被终止的情况下：
if (parent != null && parent.currentThreadShouldBeStopped()) {
long keepAliveTime = parent.getKeepAliveTime(TimeUnit.MILLISECONDS);
return poll(keepAliveTime, TimeUnit.MILLISECONDS);
}
return super.take();
}
/** * 返回队列的剩余容量 */
@Override
public int remainingCapacity() {
if (forcedRemainingCapacity > DEFAULT_FORCED_REMAINING_CAPACITY) {
return forcedRemainingCapacity;
}
return super.remainingCapacity();
}
/** * 强制设置剩余容量 */
public void setForcedRemainingCapacity(int forcedRemainingCapacity) {
this.forcedRemainingCapacity = forcedRemainingCapacity;
}
/** * 重置剩余容量 */
void resetForcedRemainingCapacity() {
this.forcedRemainingCapacity = DEFAULT_FORCED_REMAINING_CAPACITY;
}
}
2.2 自定义线程池ThreadPoolExecutor
Tomcat 线程池 ThreadPoolExecutor 是继承的 AbstractExecutorService 类，但是很多代码依旧使用的是 JDK 的 ThreadPoolExecutor，只是稍微改造了一部分
public void execute(Runnable command, long timeout, TimeUnit unit) {
/** * 提交任务的数量+1 */
submittedCount.incrementAndGet();
try {
/** * 线程池内部方法，真正执行的方法。就是JDK线程池原生的方法。 * * 因为重写了阻塞队列，才完成Tomcat特有逻辑的实现。 * * 1. 重写队列方法；达到核心线程数，然后向阻塞队列中放置，阻塞队列直接返回false * 2. 返回false，则开启maximum pool size； * 3. maximum pool size到达极限时，会抛出RejectedExecutionException方法。 * */
executeInternal(command);
/** * 任务被拒绝 */
} catch (RejectedExecutionException rx) {
/** * 在将被拒绝的任务放入到队列中。 */
if (getQueue() instanceof TaskQueue) {
//如果Executor接近最大线程数，应该将任务添加到队列中，而不是拒绝。
final TaskQueue queue = (TaskQueue) getQueue();
try {
//强制的将任务放入到阻塞队列中
if (!queue.force(command, timeout, unit)) {
//放入失败，则继续抛出异常
submittedCount.decrementAndGet();
throw new RejectedExecutionException("threadPoolExecutor.queueFull");
}
} catch (InterruptedException x) {
//被中断也抛出异常
submittedCount.decrementAndGet();
throw new RejectedExecutionException(x);
}
} else {
//不是这种队列，那么当任务满了之后，直接抛出去。
submittedCount.decrementAndGet();
throw rx;
}
}
}
/** * JDK线程池的任务执行的逻辑 */
private void executeInternal(Runnable command) {
if (command == null) {
throw new NullPointerException();
}
int c = ctl.get();
//未达到corePoolSize数量，则去开启线程
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true)) {
return;
}
c = ctl.get();
}
//开启到corePoolSize数量的工作线程，则将任务放入队列。
//但是Tomcat重写了阻塞队列。
//当放入workQueue.offer(command)返回false，则继续开启线程数
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);
}
}
注意事项
tomcat 的优化
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) {
decrementWorkerCount();
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
//获取剩余的最小线程数，若配置了允许最小线程数关闭的参数，min=0
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
//当配置了允许最小线程关闭参数，且队列不为空的情况下，允许保留的最小线程数为1
/** * 这是关闭空闲核心线程数时的判断。 */
if (min == 0 && !workQueue.isEmpty()) {
min = 1;
}
// https://bz.apache.org/bugzilla/show_bug.cgi?id=65454
// If the work queue is not empty, it is likely that a task was
// added to the work queue between this thread timing out and
// the worker count being decremented a few lines above this
// comment. In this case, create a replacement worker so that
// the task isn't held in the queue waiting for one of the other
// workers to finish.
/** * 若当前工作数量>允许的最小线程数，那么关闭改线程。 * Tomcat对此处进行了优化。 */
if (workerCountOf(c) >= min && workQueue.isEmpty()) {
return; // replacement not needed
}
}
/** * 没有终止线程，又重新开启线程 */
addWorker(null, false);
}
}
文章评论