图灵学院Java架构师2345期-VIP-百度云-并发编程（线程池）
【微信642620018，获取全套课程】
1、Executor
public interface Executor {
// 执行Runnable任务
void execute(Runnable command);
}
ExecutorService：
- 提供了多种提交任务的方式，支持Callable
- 定义了线程池相关操作的接口
public interface ExecutorService extends Executor {
// 启动有序关闭，其中先前提交的任务将被执行，但不会接受任何新任务。
void shutdown();
/**
* 立即关闭执行器, 主要有以下特点:
* 1. 尝试停止所有正在执行的任务, 无法保证能够停止成功, 但会尽力尝试(例如, 通过 Thread.interrupt中断任务, 但是不响应中断的任务可能无法终止);
* 2. 暂停处理已经提交但未执行的任务;
*
* @return 返回已经提交但未执行的任务列表
*/
List<Runnable> shutdownNow();
// 如果此执行程序已关闭，则返回true。
boolean isShutdown();
// 仅当执行器已关闭且所有任务都已经执行完成, 才返回true.
boolean isTerminated();
// 阻止所有任务在关闭请求之后完成执行，或者发生超时，或者当前线程被中断，以先发生者为准
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
// 接收Callable实现，返回Future对象
<T> Future<T> submit(Callable<T> task);
// 提交Runnable任务以执行并返回表示该任务的Future。
<T> Future<T> submit(Runnable task, T result);
// 提交Runnable任务以执行并返回表示该任务的Future
// 注意: Future的get方法在成功完成时将会返回null.
Future<?> submit(Runnable task);
// 执行给定集合中的所有任务, 当所有任务都执行完成后, 返回保持任务状态和结果的 Future 列表.
// 注意: 该方法为同步方法. 返回列表中的所有元素的Future.isDone() 为 true.
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
// 执行给定集合中的所有任务, 当所有任务都执行完成后或超时期满时（无论哪个首先发生）
// 返回保持任务状态和结果的 Future 列表.
<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;
}
AbstractExecutorService：
- 聚合各种提交方式的入口，将各种任务包装为RunnableFuture，统一调用
execute(RunnableFuture ftask); - 实现各种
invoke*的调用逻辑，聚合为doInvokeAny
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;
}
ScheduledExecutorService：
交给执行器的某些任务能够定时执行或周期性地执行
public interface ScheduledExecutorService extends ExecutorService {
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
}
Executors：
- 提供线程池的工厂函数
- 操作线程池创建
核心创建方式：
- newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
- newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
- newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行
- newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
核心API：
1、execute(Runnable command):履行Ruannable类型的任务,
2、submit(task):可用来提交Callable或Runnable任务，并返回代表此任务的Future 对象
3、shutdown():在完成已提交的任务后封闭办事，不再接管新任务,
4、shutdownNow():停止所有正在履行的任务并封闭办事。
5、isTerminated():测试是否所有任务都履行完毕了。
6、isShutdown():测试是否该ExecutorService已被关闭。
核心参数：
- corePoolSize :池中所保存的线程数，包括空闲线程
- maximumPoolSize:池中允许的最大线程数
- keepAliveTime: 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间
- unit:keepAliveTime 参数的时间单位
- workQueue :执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务
- threadFactory:执行程序创建新线程时使用的工厂
- handler :由于超出线程范围和队列容量而使执行
运行逻辑：
- 如果当前池大小，poolSize 小于 corePoolSize ，则创建新线程执行任务
- 如果当前池大小，poolSize 大于 corePoolSize ，且等待队列未满，则进入等待队列
- 如果当前池大小，poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程，执行任务
- 如果当前池大小，poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策
- 线程池里的每个线程执行完任务后不会立刻退出，而是会去检查下等待队列里是否还有线程任务需要执行， 如果在keepAliveTime 里等不到新的任务了，那么线程就会退出
2、执行原理
构造参数：
/**
* 使用给定的参数创建ThreadPoolExecutor.
*
* @param corePoolSize 核心线程池中的最大线程数
* @param maximumPoolSize 总线程池中的最大线程数
* @param keepAliveTime 空闲线程的存活时间
* @param unit keepAliveTime的单位
* @param workQueue 任务队列, 保存已经提交但尚未被执行的线程
* @param threadFactory 线程工厂(用于指定如何创建一个线程)
* @param handler 拒绝策略 (当任务太多导致工作队列满时的处理策略)
*/
ThreadPoolExecutor在逻辑上将自身管理的线程池划分为两部分：
- 核心线程池（大小对应为corePoolSize）、
- 非核心线程池（大小对应为maximumPoolSize-corePoolSize）。
向线程池提交一个任务时，将创建一个工作线程——称之为Worker；Worker在逻辑上从属于下图中的【核心线程池】或【非核心线程池】，具体属于哪一种，要根据corePoolSize、maximumPoolSize、Worker总数进行判断
【核心线程池】【非核心线程池】都是逻辑上的概念，ThreadPoolExecutor在任务调度过程中会根据corePoolSize和maximumPoolSize的大小，判断应该如何调度任务
工作原理：
ThreadPoolExecutor内部定义了一个AtomicInteger变量——ctl，通过按位划分的方式，在一个变量中记录线程池状态和工作线程数——低29位保存线程数，高3位保存线程池状态：
/**
* 保存线程池状态和工作线程数:
* 低29位: 工作线程数
* 高3位 : 线程池状态
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
// 最大线程数: 2^29-1
private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 00011111 11111111 11111111 11111111
// 线程池状态
// 11100000 00000000 00000000 00000000
private static final int RUNNING = -1 << COUNT_BITS;
// 00000000 00000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 00100000 00000000 00000000 00000000
private static final int STOP = 1 << COUNT_BITS;
// 01000000 00000000 00000000 00000000
private static final int TIDYING = 2 << COUNT_BITS;
// 01100000 00000000 00000000 00000000
private static final int TERMINATED = 3 << COUNT_BITS;
可以看到，ThreadPoolExecutor一共定义了5种线程池状态：
- RUNNING : 接受新任务, 且处理已经进入阻塞队列的任务
- SHUTDOWN : 不接受新任务, 但处理已经进入阻塞队列的任务
- STOP : 不接受新任务, 且不处理已经进入阻塞队列的任务, 同时中断正在运行的任务
- TIDYING : 所有任务都已终止, 工作线程数为0, 线程转化为TIDYING状态并准备调用terminated方法
- TERMINATED : terminated方法已经执行完成
内部通过HashSet维护Worker
/**
* 工作线程集合.
*/
private final HashSet<Worker> workers = new HashSet<Worker>();
Worker：
/**
* Worker表示线程池中的一个工作线程, 可以与任务相关联.
* 由于实现了AQS框架, 其同步状态值的定义如下:
* -1: 初始状态
* 0: 无锁状态
* 1: 加锁状态
*/
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
/**
* 与该Worker关联的线程.
*/
final Thread thread;
/**
* Initial task to run. Possibly null.
*/
Runnable firstTask;
/**
* Per-thread task counter
*/
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // 初始的同步状态值
this.firstTask = firstTask;
// 每个worker对象会调用线程工厂，
this.thread = getThreadFactory().newThread(this);
}
/**
* 执行任务
*/
public void run() {
runWorker(this);
}
/**
* 是否加锁
*/
protected boolean isHeldExclusively() {
return getState() != 0;
}
/**
* 尝试获取锁
*/
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
/**
* 尝试释放锁
*/
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() {
acquire(1);
}
public boolean tryLock() {
return tryAcquire(1);
}
public void unlock() {
release(1);
}
public boolean isLocked() {
return isHeldExclusively();
}
/**
* 中断线程(仅任务非初始状态)
*/
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
线程工厂：
以下为默认的线程工厂
public static ThreadFactory defaultThreadFactory() {
return new DefaultThreadFactory();
}
/**
* 默认的线程工厂.
*/
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
execute的整个执行流程关键是下面两点：
- 如果工作线程数小于核心线程池上限（CorePoolSize），则直接新建一个工作线程并执行任务；
- 如果工作线程数大于等于CorePoolSize，则尝试将任务加入到队列等待以后执行。如果加入队列失败了（比如队列已满的情况），则在总线程池未满的情况下（
CorePoolSize ≤ 工作线程数 ＜ maximumPoolSize）新建一个工作线程立即执行任务，否则执行拒绝策略。
Worker线程的生命周期
拒绝策略：
ThreadPoolExecutor在以下两种情况下会执行拒绝策略：
- 当核心线程池满了以后，如果任务队列也满了，首先判断非核心线程池有没满，没有满就创建一个工作线程（归属非核心线程池）， 否则就会执行拒绝策略；
- 提交任务时，ThreadPoolExecutor已经关闭了。
四种拒绝策略：
- AbortPolicy（默认）：AbortPolicy策略其实就是抛出一个RejectedExecutionException异常
- DiscardPolicy：DiscardPolicy策略其实就是无为而治，什么都不做，等任务自己被回收
- DiscardOldestPolicy：DiscardOldestPolicy策略是丢弃任务队列中的最近一个任务，并执行当前任务
- CallerRunsPolicy：CallerRunsPolicy策略相当于以自身线程来执行任务，这样可以减缓新任务提交的速度
3、源码解析：
构造：
ExecutorService executorService = Executors.newFixedThreadPool(100);
创建ThreadPoolExecutor实例：
// 创建ThreadPoolExecutor
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
// 指定默认的线程创建工厂和拒绝策略
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
/**
* 使用给定的参数创建ThreadPoolExecutor.
*
* @param corePoolSize 核心线程池中的最大线程数
* @param maximumPoolSize 总线程池中的最大线程数
* @param keepAliveTime 空闲线程的存活时间
* @param unit keepAliveTime的单位
* @param workQueue 任务队列, 保存已经提交但尚未被执行的线程
* @param threadFactory 线程工厂(用于指定如何创建一个线程)
* @param 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;
}
执行任务：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// CASE1: 工作线程数 < 核心线程池上限
if (workerCountOf(c) < corePoolSize) {
// 添加工作线程并执行
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果工作队列未满，再次检查运行状态，并插入到任务
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 再次check判断运行状态如果是非运行状态就移除出去&reject掉
if (!isRunning(recheck) && remove(command))
reject(command);
// 发现可能运行线程数是0那么增加一个null的worker
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 直接增加worker如果不成功直接reject
else if (!addWorker(command, false))
reject(command);
}
addWorker()
/**
* 添加工作线程并执行任务
*
* @param firstTask 如果指定了该参数, 表示将立即创建一个新工作线程执行该firstTask任务;
否则复用已有的工作线程，从工作队列中获取任务并执行
* @param core 执行任务的工作线程归属于哪个线程池: true-核心线程池 false-非核心线程池
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
// 获取线程池状态
int rs = runStateOf(c);
/**
* 这个if主要是判断哪些情况下, 线程池不再接受新任务执行
* 1. 线程池状态为 STOP 或 TIDYING 或 TERMINATED: 线程池状态为上述任一一种
* 2. 线程池状态 ≥ SHUTDOWN 且 firstTask != null:
因为当线程池状态 ≥ SHUTDOWN时, 不再接受新任务的提交，所以直接返回
* 3. 线程池状态 ≥ SHUTDOWN 且 队列为空
队列中已经没有任务了, 所以也就不需要执行任何任务了，可以直接返回
*/
if (rs >= SHUTDOWN &&
!(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 检查容量是否超出
// 1. 超出最大容量；2、core为true表示核心线程数量，为false表示非核心线程数量
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 增加worker数量
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 线程池状态发生变化, 重新自旋判断
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
// 创建worker(AQS)
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();
// 添加到worker队列中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 开始执行任务
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
// 添加worker失败处理
addWorkerFailed(w);
}
return workerStarted;
}
启动Worker后，会调用run()
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 当task==null时会通过getTask从队列取任务
while (task != null || (task = getTask()) != null) {
w.lock();
/**
* 下面这个if判断的作用如下:
* 1.保证当线程池状态为STOP/TIDYING/TERMINATED时，
当前执行任务的线程wt是中断状态(因为线程池处于上述任一状态时，均不能再执行新任务)
* 2.保证当线程池状态为RUNNING/SHUTDOWN时，当前执行任务的线程wt不是中断状态
*/
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
// 执行前置处理, 1.8版本为空实现
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 执行任务
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
// 执行后置处理, 1.8版本为空实现
afterExecute(task, thrown);
}
} finally {
task = null;
// 更新该worker完成状态
w.completedTasks++;
w.unlock();
}
}
// 执行到此处, 说明该工作线程自身既没有携带任务, 也没从任务队列中获取到任务
// 正常退出时，为false, 如果抛出异常退出，则为true
completedAbruptly = false;
} finally {
// 处理工作线程的退出工作
processWorkerExit(w, completedAbruptly);
}
}
processWorkerExit(Worker w, boolean completedAbruptly)
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 工作线程因异常情况而退出
if (completedAbruptly)
// 工作线程数减1(如果工作线程执行时没有出现异常, 在getTask()方法中已经对线程数减1了)
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// completedTaskCount记录线程池完成的总任务数
completedTaskCount += w.completedTasks;
// 从工作线程集合中移除(该工作线程会自动被GC回收)
workers.remove(w);
} finally {
mainLock.unlock();
}
// 根据线程池状态, 判断是否需要终止线程池
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
// 工作线程为正常退出
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 新建一个工作线程
addWorker(null, false);
}
}
4、ScheduledThreadPoolExecutor
要满足任务的延迟/周期调度功能，它会对所有的Runnable任务都进行包装，包装成一个RunnableScheduledFuture任务。
- ScheduledThreadPoolExecutor中的任务队列是一种特殊的延时队列
- DelayedWorkQueue：它和DelayQueue类似，只不过要求所有入队元素必须实现RunnableScheduledFuture接口。
执行原理
构造：
ScheduledThreadPoolExecutor的构造器，内部其实都是调用了父类ThreadPoolExecutor的构造器，这里最需要注意的就是任务队列的选择——DelayedWorkQueue
调度：
核心调度方法是schedule、scheduleAtFixedRate、scheduleWithFixedDelay
1、将任务包装为ScheduledFutureTask
2、判断状态，将任务添加到DelayedWorkQueue
- 首先，任务被提交到线程池后，会判断线程池的状态，如果不是RUNNING状态会执行拒绝策略。
- 然后，将任务添加到阻塞队列中。（注意，由于DelayedWorkQueue是无界队列，所以一定会add成功）
- 然后，会创建一个工作线程，加入到核心线程池或者非核心线程池
如果核心线程池未满，则新建的工作线程会被放到核心线程池中。如果核心线程池已经满了，ScheduledThreadPoolExecutor不会像ThreadPoolExecutor那样再去创建归属于非核心线程池的工作线程，而是直接返回。也就是说，在ScheduledThreadPoolExecutor中，一旦核心线程池满了，就不会再去创建工作线程。
生产实践
如何合理的设置线程池大小？
分析任务的特性：
- 任务的性质：CPU密集型任务、IO密集型任务、混合型任务。
- 任务的优先级：高、中、低。
- 任务的执行时间：长、中、短。
- 任务的依赖性：是否依赖其他系统资源，如数据库连接等。
- CPU密集型任务应配置尽可能小的线程，如配置CPU个数+1的线程数
- IO密集型任务应配置尽可能多的线程，因为IO操作不占用CPU，不要让CPU闲下来，应加大线程数量，如配置两倍CPU个数+1
- 而对于混合型的任务，如果可以拆分，拆分成IO密集型和CPU密集型分别处理，前提是两者运行的时间是差不多的，如果处理时间相差很大，则没必要拆分
- 若任务对其他系统资源有依赖，如某个任务依赖数据库的连接返回的结果，这时候等待的时间越长，则CPU空闲的时间越长，那么线程数量应设置得越大，才能更好的利用CPU。
合理设置线程池大小公式：
最佳线程数目 = （（线程等待时间+线程CPU时间）/ 线程CPU时间 ）* CPU数目
转化为：
最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目
**结论：**线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程
并发情况下线程池配置：
- 并发高，执行时间短的任务配置尽可能少的线程：CPU核数 + 1
- 并发高、业务执行时间长的任务，对于系统的压力很大，应尽可能通过架构的优化，而不是线程池的配置解决。例如转换为异步削峰解藕
- 并发不高，业务时间长：
- 如果任务长时间消耗在IO操作上，应加大线程池数量，不让CPU闲下来，尽量执行更多的任务
- 如果任务长时间消耗在计算上，应减少CPU线程的切换，设置和CPU核数一致的线程数量。
《Java并发编程实战》
Nthreads = NCPU * UCPU * (1 + W/C)
其中：
NCPU是处理器的核的数目，可以通过Runtime.getRuntime().availableProcessors()得到
UCPU是期望的CPU利用率（该值应该介于0和1之间）
W/C是等待时间与计算时间的比率
CPU密集型，设置核心线程永久存在：因为执行任务优先判断核心线程数量，不足则创建，避免资源浪费
CPU密集型任务，队列选择阻塞队列，高并发场景大量CAS资源占用大量CPU资源，影响性能。