public interface Executor {
// 执行Runnable任务
void execute(Runnable command);
}ExecutorService：
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：
execute(RunnableFuture ftask);invoke*的调用逻辑，聚合为doInvokeAnypublic 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：
核心创建方式：
核心API：
1、execute(Runnable command):履行Ruannable类型的任务,
2、submit(task):可用来提交Callable或Runnable任务，并返回代表此任务的Future 对象?
3、shutdown():在完成已提交的任务后封闭办事，不再接管新任务,
4、shutdownNow():停止所有正在履行的任务并封闭办事。
5、isTerminated():测试是否所有任务都履行完毕了。
6、isShutdown():测试是否该ExecutorService已被关闭。?
核心参数：
运行逻辑：
构造参数：
/**
* 使用给定的参数创建ThreadPoolExecutor.
*
* @param corePoolSize 核心线程池中的最大线程数
* @param maximumPoolSize 总线程池中的最大线程数
* @param keepAliveTime 空闲线程的存活时间
* @param unit keepAliveTime的单位
* @param workQueue 任务队列, 保存已经提交但尚未被执行的线程
* @param threadFactory 线程工厂(用于指定如何创建一个线程)
* @param handler 拒绝策略 (当任务太多导致工作队列满时的处理策略)
*/ThreadPoolExecutor在逻辑上将自身管理的线程池划分为两部分：
向线程池提交一个任务时，将创建一个工作线程——称之为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种线程池状态：
内部通过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 ≤ 工作线程数 ＜ maximumPoolSize）新建一个工作线程立即执行任务，否则执行拒绝策略。Worker线程的生命周期
拒绝策略：
ThreadPoolExecutor在以下两种情况下会执行拒绝策略：
四种拒绝策略：
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);
}
}要满足任务的延迟/周期调度功能，它会对所有的Runnable任务都进行包装，包装成一个RunnableScheduledFuture任务。
构造：
ScheduledThreadPoolExecutor的构造器，内部其实都是调用了父类ThreadPoolExecutor的构造器，这里最需要注意的就是任务队列的选择——DelayedWorkQueue
调度：
核心调度方法是schedule、scheduleAtFixedRate、scheduleWithFixedDelay
1、将任务包装为ScheduledFutureTask
2、判断状态，将任务添加到DelayedWorkQueue
如果核心线程池未满，则新建的工作线程会被放到核心线程池中。如果核心线程池已经满了，ScheduledThreadPoolExecutor不会像ThreadPoolExecutor那样再去创建归属于非核心线程池的工作线程，而是直接返回。也就是说，在ScheduledThreadPoolExecutor中，一旦核心线程池满了，就不会再去创建工作线程。
如何合理的设置线程池大小？
分析任务的特性：
合理设置线程池大小公式：
最佳线程数目 = （（线程等待时间+线程CPU时间）/ 线程CPU时间 ）* CPU数目
转化为：
最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目
结论：线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程
并发情况下线程池配置：
《Java并发编程实战》
Nthreads = NCPU * UCPU * (1 + W/C)
其中：
NCPU是处理器的核的数目，可以通过Runtime.getRuntime().availableProcessors()得到
UCPU是期望的CPU利用率（该值应该介于0和1之间）
W/C是等待时间与计算时间的比率
CPU密集型，设置核心线程永久存在：因为执行任务优先判断核心线程数量，不足则创建，避免资源浪费
CPU密集型任务，队列选择阻塞队列，高并发场景大量CAS资源占用大量CPU资源，影响性能。
原文：https://www.cnblogs.com/zuier/p/11388872.html