概念
维护了两个集合，一个是线程的集合，一个是任务的集合。线程池就是首先创建一些线程，它们的集合称为线程池。使用线程池可以很好地提高性能，线程池在系统启动时即创建大量空闲的线程，程序将一个任务传给线程池，线程池就会启动一条线程来执行这个任务，执行结束以后，该线程并不会死亡，而是再次返回线程池中成为空闲状态，等待执行下一个任务。
Executor接口
顶层接口，里面只有一个execute方法，具体怎么实现，有实现的方法来定。
public interface Executor {
void execute(Runnable command);
}
ExecutorService
继承了Executor接口，并新增了一些方法。拓展了Executor接口。相当于有了一整个的生命周期的方法。
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<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;
}
Callable
类似于Runnable，不过它有返回值。一般配合线程池使用。
@FunctionalInterface//函数式接口
public interface Callable<V> {
V call() throws Exception;//它有返回值
}
只有线程池里的submit方法可以使用实现Callable的任务类。并且会把返回值封装到一个Future对象里。
<T> Future<T> submit(Callable<T> task);
Future
存储执行的将来才会产生的结果。
submit方法的返回值是一个Future。
所以这个任务提交以后，要求这个任务是有一个结果的。当然任务是异步执行的。
需要获取结果，调用get方法，这个方法是阻塞的，直到有返回结果，阻塞结束。
FutureTask
该类实现了RunnableFuture，而RunnableFuture继承了Runnable, Future接口。所以它是Future + Runnable
它既是一个Future，又是一个task。也就是说它自己本身就是一个Runnable，同时又有返回结果。
FutureTask futureTask = new FutureTask(() -> {
int a = 1 + 3;
return a;
});
new Thread(futureTask).start();
System.out.println(futureTask.get());
CompletableFuture
如何管理多个Future结果？可以使用该类来管理。汇总结果。
ThreadPoolExecutor
ThreadPoolExecutor继承了AbstractExecutorService，AbstractExecutorService实现了ExecutorService
阿里巴巴开发手册中规范，就是要使用它来自定义线程池。不允许使用Executors工具类来定义线程池。
它有好多构造方法。选择一个参数最多的解释：
int corePoolSize：核心线程数。默认情况下，没有任务进入为0。当任务数=corePoolSize时，其他任务会进入队列。
int maximumPoolSize：最大线程数。核心线程不够，扩展使用。队列达到上限，开始扩容线程数。
线程池内部有一个ctl原子类变量(poolSize)是记录线程池中当前有多少个线程的。
当新提交一个任务时：
（1）如果poolSize < corePoolSize，新增加一个线程处理新的任务。
（2）如果poolSize = corePoolSize，新任务会被放入阻塞队列等待。
（3）如果阻塞队列的容量达到上限，且这时poolSize < maximumPoolSize，新增线程来处理任务。
（4）如果阻塞队列满了，且poolSize =maximumPoolSize，那么线程池已经达到极限，
会根据饱和策略RejectedExecutionHandler拒绝新的任务。
long keepAliveTime：生存时间。闲置超出规定时间，将线程归还给操作系统。
TimeUnit unit：生存时间单位。
BlockingQueue<Runnable> workQueue：任务队列。
ThreadFactory threadFactory：线程工厂。
RejectedExecutionHandler handler：拒绝策略。
ThreadFactory是一个接口，可以通过实现它来定义自己的线程工厂。
阿里巴巴手册规范是要自己指定线程工厂，尤其是线程名称，方便出错时根据名称查找是哪里出了问题。
handler拒绝策略有四种取值，但仍然可以自定义，实际开发一般都自定义：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务，也就是哪个线程调用了execute方法，哪个线程去执行。
ThreadPoolExecutor源码理解
常用变量的解释
// 1. ctl，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 2. COUNT_BITS，`Integer.SIZE`为32位，所以`COUNT_BITS`为29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 3. CAPACITY，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//装线程的容器，所有的线程将被装进workers
private final HashSet<Worker> workers = new HashSet<Worker>();
// runState is stored in the high-order bits
// 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
// 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0
private static int workerCountOf(int c) { return c & CAPACITY; }
// 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
// 8. `runStateLessThan()`，线程池状态小于xx
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 9. `runStateAtLeast()`，线程池状态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
构造方法
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
// 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
提交执行task的过程
public void execute(Runnable command) {
//任务为空抛出异常
if (command == null)
throw new NullPointerException();
/*
* 分三步进行:
*
* 1、如果运行的线程少于corePoolSize，请尝试
* 以给定的命令作为第一个线程启动新线程
* 任务。对addWorker的调用自动检查运行状态和
* workerCount，从而防止会添加
* 当线程不应该时，返回false。
*
* 2、如果任务可以成功排队，那么我们仍然需要
* 仔细检查我们是否应该添加一个线程
* （因为现有的自上次检查后死亡）或
* 自进入此方法后，池已关闭。所以我们
* 重新检查状态，必要时回滚排队
* 已停止，如果没有，则启动新线程。
*
* 3、如果无法将任务排队，则尝试添加新的
* 线。如果失败了，我们知道我们被关闭或饱和
* 所以拒绝这个任务。
*/
int c = ctl.get();//拿到ctl
// 活跃线程数比核心线程数小，直接创建worker执行任务
if (workerCountOf(c) < corePoolSize) {
//addWorker相当于添加一个新的线程
if (addWorker(command, true))
return;
c = ctl.get();//再次获取ctl
}
// worker数量超过核心线程数，任务直接进入队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
// 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
if (! isRunning(recheck) && remove(command))
reject(command);
// 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
// 这儿有3点需要注意：
// 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
// 2. addWorker第2个参数表示是否创建核心线程
// 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
else if (!addWorker(command, false))
reject(command);
}
addworker源码
//以自旋的方式添加线程，一丝的性能都不放过
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 外层自旋
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);//获取线程池状态
// 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
// (rs > SHUTDOWN) ||
// (rs == SHUTDOWN && firstTask != null) ||
// (rs == SHUTDOWN && workQueue.isEmpty())
// 1. 线程池状态大于SHUTDOWN时，直接返回false
// 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
// 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内层自旋
for (;;) {
int wc = workerCountOf(c);
// worker数量超过容量，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 使用CAS的方式增加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 {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
// worker的添加必须是串行的，因此需要加锁
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)) {
// worker已经调用过了start()方法，则不再创建worker
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// worker创建并添加到workers成功
workers.add(w);
// 更新`largestPoolSize`变量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 启动worker线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
线程池worker任务单元
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// 省略代码...
}
核心线程执行逻辑-runworker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 调用unlock()是为了让外部可以中断
w.unlock(); // allow interrupts
// 这个变量用于判断是否进入过自旋（while循环）
boolean completedAbruptly = true;
try {
// 这儿是自旋
// 1. 如果firstTask不为null，则执行firstTask；
// 2. 如果firstTask为null，则调用getTask()从队列获取任务。
// 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
while (task != null || (task = getTask()) != null) {
// 这儿对worker进行加锁，是为了达到下面的目的
// 1. 降低锁范围，提升性能
// 2. 保证每个worker执行的任务是串行的
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 如果线程池正在停止，则对当前线程进行中断操作
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
// 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
// 这两个方法在当前类里面为空实现。
try {
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 {
afterExecute(task, thrown);
}
} finally {
// 帮助gc
task = null;
// 已完成任务数加一
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 自旋操作被退出，说明线程池正在结束
processWorkerExit(w, completedAbruptly);
}
}
Executors
线程池的工厂，生成线程池的工具类。
1、newSingleThreadExecutor()：该线程池里只有一个线程。可以保证任务是顺序执行的。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L,
TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}
2、newCachedThreadPool()：缓存线程池。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
3、newFixedThreadPool()：定长线程池。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
4、newScheduledThreadPool()：定时任务线程池。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
//super就是ThreadPoolExecutor
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
线程池大小设置建议：
threadcount = cpuCore * cpuRatio * (1 + W/C)
cpuCore：cpu核数。
cpuRatio：期望的cpu利用率(该值应该介于0和1之间，0%~100%)
W/C：wait，compatation，等待时间与计算时间的比率。
如计算型任务，wait时间基本为0，如果是IO型，wait时间就体现出来了。
具体还是根据实际情况进行压测。
WorkStealingPool
内部使用的是ForkJoinPool。每个线程都有自己单独的队列。当自己执行完以后，会去别的队列里去拿任务来执行。
newWorkStealingPool适合使用在很耗时的操作，但是newWorkStealingPool不是
ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，
但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），
所以适合使用在很耗时的任务中。
Executors.newWorkStealingPool();
public static ExecutorService newWorkStealingPool() {
return new ForkJoinPool
(Runtime.getRuntime().availableProcessors(),
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}
ForkJoinPool
ForkJoinPool采用工作窃取算法，将一个大任务根据阈值分割成很多个子任务，
最后根据场景是否要合并子任务运算结果;
用很少的线程可以执行很多的任务(子任务)ThreadPoolExecutor做不到先执行子任务。
根据是否需要合并子任务运算结果，任务需要继承抽象类RecursiveAction或RecursiveTask<V>，
后者为需要合并子任务结果，泛型为结果类型。前者无返回结果。
基本思想：
1.ForkJoinPool 的每个工作线程都维护着一个工作队列（WorkQueue），
这是一个双端队列（Deque），里面存放的对象是任务（ForkJoinTask）。
2.每个工作线程在运行中产生新的任务（通常是因为调用了 fork()）时，
会放入工作队列的队尾，并且工作线程在处理自己的工作队列时，
使用的是 LIFO(后进先出) 方式，也就是说每次从队尾取出任务来执行。
3.每个工作线程在处理自己的工作队列同时，会尝试窃取一个任务（或是来自于刚刚提交到 pool 的任务，
或是来自于其他工作线程的工作队列），窃取的任务位于其他线程的工作队列的队首，
也就是说工作线程在窃取其他工作线程的任务时，使用的是 FIFO 方式。
4.在遇到 join() 时，如果需要 join 的任务尚未完成，则会先处理其他任务，并等待其完成。
5.在既没有自己的任务，也没有可以窃取的任务时，进入休眠。
使用示例：
public class ForkJoinTaskExample extends RecursiveTask<Integer> {
public static final int threshold = 2;
private int start;
private int end;
public ForkJoinTaskExample(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
int sum = 0;
//如果任务足够小就计算任务
boolean canCompute = (end - start) <= threshold;
if (canCompute) {
for (int i = start; i <= end; i++) {
sum += i;
}
} else {
// 如果任务大于阈值，就分裂成两个子任务计算
int middle = (start + end) / 2;
ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);
// 执行子任务
leftTask.fork();
rightTask.fork();
// 等待任务执行结束合并其结果
int leftResult = leftTask.join();
int rightResult = rightTask.join();
// 合并子任务
sum = leftResult + rightResult;
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkjoinPool = new ForkJoinPool();
//生成一个计算任务，计算1+2+3+4
ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);
//执行一个任务
Future<Integer> result = forkjoinPool.submit(task);
try {
log.info("result:{}", result.get());
} catch (Exception e) {
log.error("exception", e);
}
}
}
重点注意：
1.ForkJoinPool 使用submit 或 invoke 提交的区别：invoke是同步执行，调用之后需要等待任务完成，
才能执行后面的代码；submit是异步执行，只有在Future调用get的时候会阻塞。
2.这里继承的是RecursiveTask，还可以继承RecursiveAction。
前者适用于有返回值的场景，而后者适合于没有返回值的场景
3.这一点是最容易忽略的地方，其实这里执行子任务调用fork方法并不是最佳的选择，
最佳的选择是invokeAll方法。
leftTask.fork();
rightTask.fork();
替换为
invokeAll(leftTask, rightTask);
原理：
对于Fork/Join模式，假如Pool里面线程数量是固定的，那么调用子任务的fork方法相当于A先分工给B，
然后A当监工不干活，B去完成A交代的任务。所以上面的模式相当于浪费了一个线程。
那么如果使用invokeAll相当于A分工给B后，A和B都去完成工作。这样可以更好的利用线程池，缩短执行的时间。
若有收获，就点个赞吧
0 人点赞
