Java高并发 -- 线程池
主要是学习慕课网实战视频《Java并发编程入门与高并发面试》的笔记
在使用线程池后，创建线程变成了从线程池里获得空闲线程，关闭线程变成了将线程归坏给线程池。
JDK有一套Executor框架，大概包括Executor、ExecutorService、AbstractExeccutorService、ThreadPoolExecutor、Executors等成员，位于java.util.concurrent包下。它们之间的关系如下：
Executor是顶层的接口，ExecutorService接口继承了它，AbstrctExecutorService继承了ExecutorService，ThreadPoolExecutor继承了AbstrctExecutorService。如果用<——表示继承，<--表示实现接口，它们的关系可表示如下：
Executor(接口) <—— ExecutorService(接口) <-- AbstrctExecutorService(抽象类) <—— ThreadPoolExecutor(类)
Executors是单独的一个类，可以看成是“线程池工厂”，它有很多静态方法，比如：
- newFixedThreadPool(int nThread)
- newSingleThreadExecutor()
- newCachedThreadPool()
- newSingleThreadScheduledExecutor()
- newScheduledThreadPool(int corePoolSize)
newFixedThreadPool该方法返回一个固定线程数的线程池。当有新任务提交时，如果线程池中有空闲线程就立即执行，否则会进入任务队列中，等到有空闲线程了才能执行。
newSingleThreadExecutor，该方法返回只有一个线程的线程池，处理策略和上面一样。实际上就是上面的参数指定为1而已。
newCachedThreadPool该方法返回一个可根据实际情况调整线程数的线程池，任务提交后，如果有空闲线程可以复用，则优先复用。若线程池中的线程全部在工作，而此时有新任务，则会创建新的线程来处理任务，所有线程执行完后会将线程归还给线程池。
newScheduledThreadPool返回一个ScheduledExecutorService对象，可以有计划地执行任务，比如在某个延时之后开始执行，或者周期性地执行某个任务。可以指定线程数量。
newSingleThreadScheduledExecutor实现了和上面一样的功能，不过线程池的大小为1。
ScheduledExecutorService有三个方法可以有计划地执行任务。如：
-
schedule(Runnable command, long delay, TimeUnit unit);该方法可以在给定的延时后，执行一个任务； -
scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);该方法以任务开始执行的时间为initialDelay，加上周期period，就是下一个任务开始执行的时间，以此类推，因此这个方法任务调度的频率是一定的； -
scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);该方法表示每执行完一个任务，延迟delay的时间后，开始执行下一个任务，initialDelay还是表示任务开始的初始时延，上一个任务结束的时间点与下一个任务开始的时间点之差是固定的，固定为delay。
即使单个任务的执行时间超过调度周期，scheduleAtFixedRate也不会让多个任务堆叠，比如任务执行需要8s，而调度周期是2s，调度第二个任务时，第一个还没执行完，因此为了避免任务堆叠，此时调度周期会变成8s；而采用scheduleWithFixedDelay，两个任务之间的实际间隔会变成10s，8s的执行+2s的delay。
Executors是线程池的工厂类，通过调用它的静态方法如
Executors.newCachedThreadPool();
Executors.newFixedThreadPool(n);
可返回一个线程池。这些静态方法统一返回一个ThreadPoolExecutor，只是参数不同而已。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {}
包括以上几个参数，其中：
- corePoolSize：指定了线程池中线程的数量；
- maximumPoolSize：线程池中的最大线程数量；
- keepAliveTime：当线程池中线程数量超过corePoolSize时，多余的空闲线程的存活时间；
- unit：上一个参数keepAliveTime的单位
- 任务队列，被提交但还未被执行额任务
- threadFactory：线程工厂，用于创建线程，一般用默认工厂即可。
- handler：拒绝策略。当任务太多来不及处理的时候，采用什么方法拒绝任务。
最重要的是任务队列和拒绝策略。
任务队列主要有ArrayBlockingQueue有界队列、LinkedBlockingQueue无界队列、SynchronousQueue直接提交队列。
使用ArrayBlockingQueue，当线程池中实际线程数小于核心线程数时，直接创建线程执行任务；当大于核心线程数而小于最大线程数时，提交到任务队列中；因为这个队列是有界的，当队列满时，在不大于最大线程的前提下，创建线程执行任务；若大于最大线程数，执行拒绝策略。
使用LinkedBlockingQueue时，当线程池中实际线程数小于核心线程数时，直接创建线程执行任务；当大于核心线程数而小于最大线程数时，提交到任务队列中；因为这个队列是有无界的，所以之后提交的任务都会进入任务队列中。newFixedThreadPool就采用了无界队列，同时指定核心线程和最大线程数一样。
使用SynchronousQueue时，该队列没有容量，对提交任务的不做保存，直接增加新线程来执行任务。newCachedThreadPool使用的是直接提交队列，核心线程数是0，最大线程数是整型的最大值，keepAliveTime是60s，因此当新任务提交时，若没有空闲线程都是新增线程来执行任务，不过由于核心线程数是0，当60s就会回收空闲线程。
当实际线程数超过maxPoolSize时，该采取什么样的策略？
- AbortPolicy：丢弃任务并抛出异常；
- CallerRunPolicy：该任务被线程池拒绝，由调用execute方法的线程执行该任务；
- DiscardOldestPolicy：丢弃最老的一个，也就是马上要执行的一个任务；
- DiscardPolicy：默默丢弃被拒绝的任务，体现在代码中就是什么也不做。
下面看看CallerRunPolicy怎么拒绝的
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
DiscardOldestPolicy是这样做的
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll(); // 最老的一个请求在队列头部
e.execute(r);
}
}
实现一个简单的线程池
实现一个类似于Executors.newFixedThreadPool(n)的固定大小线程池，当小于corePoolSize时候，优先创建线程去执行该任务；当超过该值时，将任务提交到任务队列中，然后各个线程从任务队列中取任务来执行。
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class MyThreadPool {
private int workerCount;
private int corePoolSize;
private BlockingQueue<Runnable> workQueue;
private Set<Worker> workers;
private volatile boolean RUNNING = true;
public MyThreadPool(int corePoolSize) {
this.corePoolSize = corePoolSize;
workQueue = new LinkedBlockingQueue<>();
workers = new HashSet<>();
}
public void execute(Runnable r) {
if (workerCount < corePoolSize) {
addWorker(r);
} else {
try {
workQueue.put(r);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void addWorker(Runnable r) {
workerCount++;
Worker worker = new Worker(r);
Thread t = worker.thread;
workers.add(worker);
t.start();
}
class Worker implements Runnable {
Runnable task;
Thread thread;
public Worker(Runnable task) {
this.task = task;
this.thread = new Thread(this);
}
@Override
public void run() {
while (RUNNING) {
Runnable task = this.task;
// 执行当前的任务，所以把这个任务置空，以免造成死循环
this.task = null;
if (task != null || (task = getTask()) != null) {
task.run();
}
}
}
}
private Runnable getTask() {
Runnable r = null;
try {
r = workQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
return r;
}
public static void main(String[] args) {
MyThreadPool threadPool = new MyThreadPool(5);
Runnable r = new Writer();
for (int i = 0; i < 10; i++) {
threadPool.execute(r);
}
}
}
class Writer implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " ");
}
}
Worker实现了Runnale，是真正执行任务的类。当线程池中工作线程小于核心线程时候，调用addWorker直接start线程执行它的第一个任务。否则，将任务放入任务队列中，等线程来执行它们。Worker中的run方法是一个死循环，执行第一个任务（addWorker时调用start方法执行的那个任务），或者通过getTask方法不断从任务队列中取得任务来执行。正是getTask方法实现了线程的复用，即一个线程虽然只能调用一次start方法，但是后续的任务可以在Worker的run方法里直接调用任务的run方法得以执行。简单来说就是在Worker的run里调用任务的run方法。
任务全部执行完毕后，线程池需要被关闭，否则程序一直死循环。上述代码中并没有实现shutdown()方法。
