1.ThreadPool 基本原理

先借用一个图来看一下

1. ExecutorService是线程池最基本的接口。里面定义了包括提交任务，关闭线程池的方法
2. ScheduledExecutorService是一个扩展接口，它是在基础线程池的功能上又添加了任务调度的功能，可以用来去定时执行任务
3. ThreadPoolExecutor和ScheduledThreadPoolExecutore分别是两个实现类。我们先介绍一下ThreadPoolExecutor

1.1 线程池状态

ThreadPoolExecutor使用int的高3位来表示线程池的状态，低29位来表示线程数量

状态名	高3位	接受新任务	处理阻塞队列任务	说明
RUNNING	111	Y	Y
SHUTDOWN	000	N	Y	不会接受新任务，但会处理阻塞队列剩余任务
STOP	001	N	N	会中断正在执行的任务，并抛弃阻塞队列任务
TIDYING	010	-	-	任务全执行完毕，活动线程为0即将进入终结
TERMINATED	011	-	-	终结状态

从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING（因为32位中的第一位代表的是符号，0代表正数，1代表负数） 这些信息存储在一个原子变量ctl中，目的是将线程池状态和线程个数合二为一，这样就可以用一次cas原子操作进行赋值。

    // c 为旧值， ctlOf 返回结果为新值
    ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
    // rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
    private static int ctlOf(int rs, int wc) { return rs | wc; }

2 构造方法

public ThreadPoolExecutor(int corePoolSize, //核心线程数量
                          int maximumPoolSize, //最大线程数量
                          long keepAliveTime, //生存时间--针对救急线程
                          TimeUnit unit,//时间单位--针对救急线程
                          BlockingQueue<Runnable> workQueue, //阻塞队列
                          ThreadFactory threadFactory, //线程工程--可以为线程创建时起个好名字
                          RejectedExecutionHandler handler //拒绝策略)

这里面可能就是我们常说的线程池的七个参数了。这里所有的救急线程就是最大线程数量减去核心线程数量（当然有的人也不叫救急线程，但本质差不多）。

3.线程池工作流程

网上线程池工作流程有很多，我们借用一下。主要对流程进行描述

1. 线程池中刚开始还没有线程，当一个任务提交以后，线程池会创建一个线程来执行任务
2. 当线程数达到核心线程数的时候如果再有任务提交进来，新加入的任务就会进入到workQueue队列排队，直到有空间的核心线程
3. 如果队列选择了有届队列，当任务数量超过了队列大小就是队列满了以后。会创建救急线程来执行任务。
4. 如果救急线程也满了。还有任务提交的话，这个时候就会执行拒绝策略 keepAliveTime的意义就是当救急线程没有任务执行的时候，需要结束为了节省资源，这个时间就由keepAliveTime和unit来确定。（但核心线程不会主动结束哦！！！是需要我们调用方法主动结束）

4.线程池的拒绝策略

jdk实现了4种拒绝策略。现在还有一些框架也实现了一些更好的拒绝策略。先来看一下jdk自带的

1. AbortPolicy让调用者抛出RejectedExecutionException异常，这是默认策略
2. CallerRunsPolicy让调用者运行任务
3. DisCardPolicy放弃本次任务
4. DiscardOldestPolicy放弃本次任务

还有一些框架也实现了一些拒绝策略，简单介绍几个

1. Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
2. Netty 的实现，是创建一个新线程来执行任务

5.JDK自带的四种线程池

5.1 newFixedThreadPool 固定大小的线程池

源码如下：

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

• 核心线程数等于最大线程数（没有救急线程），因此也不需要超时时间
• 阻塞队列是无界的，可以放任意数量的任务
• 适用于任务量已知，相对耗时的任务

5.2 newCachedThreadPool 带缓冲的线程池。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

• 核心线程数是0，最大线程数是Integer.MAX_VALUE,救急线程的空闲生存时间是60s，意味着全部是救急线程，且救急线程可以无限创建。
• 队列采用了SynchronousQueue实现特点是，它没有容量，没有线程来取是放不进去的。（如果有线程往里面放任务的话，如果没有线程来取的话，就会一直处于阻塞状态，直到有其他线程来取这个任务，才继续执行）
• 整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况

5.3 newSingleThreadExecutor 单线程线程池

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

希望多个任务排队执行。线程数固定为1，任务数多于1时，会放入无界队列排队，任务执行完毕，这唯一的线程也不会被释放。 区别：

• 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一 个线程，保证池的正常工作
• Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改
  • FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
• Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改
  • 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

5.4 newScheduledTheadPool 可以定时执行的线程池。

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

5.4.1 延迟执行

通过schedule()方法可以看到一共有三个参数，1.要执行的任务2.延迟的时间3.时间单位

public static void main(String[] args) throws Exception, InterruptedException {
    ScheduledExecutorService service = Executors.newScheduledThreadPool(2);
    service.schedule(() -> {
        log.debug("start111");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug("end111");
    }, 1, TimeUnit.SECONDS);
    service.schedule(() -> {
        log.debug("start222");
        log.debug("end222");
    }, 1, TimeUnit.SECONDS);

}

这里我们提交了俩个任务，都是延时1秒执行，可以看到俩个任务都是在同一时刻执行的。

5.4.2 定时执行任务

通过scheduleAtFixedRate()方法我们可以定时执行任务(initialDelay)，并且规定每隔多久执行一次(period)

public static void main(String[] args) throws Exception, InterruptedException {
    ScheduledExecutorService service = Executors.newScheduledThreadPool(2);
    log.debug("启动");
    service.scheduleAtFixedRate(() -> {
        log.debug("start111");
    }, 1,1, TimeUnit.SECONDS);

}

可以看到在主线程启动1秒后开始执行任务，之后每隔一秒执行一次。

如果任务执行时间超过间隔时间

public static void main(String[] args) throws Exception, InterruptedException {
    ScheduledExecutorService service = Executors.newScheduledThreadPool(2);
    log.debug("启动");
    service.scheduleAtFixedRate(() -> {
        log.debug("start111");
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }, 1,1, TimeUnit.SECONDS);

}

从结果可以看到，它是会执行完任务以后再去进行下一次任务，但是间隔时间却被任务本身时间覆盖了。这个的好处会保证任务不会重叠。

5.4.3 scheduleWithFixedDelay()

scheduleWithFixedDelay()这个方法就是任务执行的间隔时间是从你上一次任务执行结束开始算的。和scheduleAtFixedDelay()是有所不同的。

6.提交任务

6.1 void execute(Runnable command);---执行任务

ExecutorService executor = Executors.newFixedThreadPool(2);
executor.execute(()->{
    System.out.println(1);
});
executor.execute(()->{
    System.out.println(2);
});

当我们创建好线程池以后可以通过execute()方法向线程池里面提交任务。

6.2  Future submit(Callable task);--提交任务 task，用返回值 Future 获得任务执行结果。

ExecutorService executor = Executors.newFixedThreadPool(2);
Future<String> s1 = executor.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
        return "hello--s1";
    }
});
Future<String> s2 = executor.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
        return "hello--s2";
    }
});
System.out.println(s1.get());
System.out.println(s2.get());

sumbit()方法可以像线程池中提交带有返回值的任务，并且还可以抛出异常，这个execute做不到的。同时返回值可以通过Future的get方法获取。

6.3 invokeAll

// 提交 tasks 中所有任务
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException;

// 提交 tasks 中所有任务，带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)throws InterruptedException;

invokeAll方法就是我们可以以集合的形式向线程池里面提交任务。它还可以指定超时时间，如果执行时间超过了我们的超时时间，就不再执行，直接返回结果。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    List<Future<String>> futures = executor.invokeAll(Arrays.asList(
            () -> {
                log.debug("start");
                Thread.sleep(2000);
                return "s1";
            },
            () -> {
                log.debug("start");
                Thread.sleep(2000);
                return "s2";
            },
            () -> {
                log.debug("start");
                Thread.sleep(2000);
                return "s3";
            }
    ));
    futures.forEach(f->{
        try {
            log.debug(f.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    });
}

这里我们只展示一下不带参数的。可以看到一个小细节就是thread1和thread2先各自执行了一个任务，然后thread2又去执行了第二个任务。是因为我们制定了核心线程数是2，当俩个核心线程都有任务以后，第三个任务会进入到阻塞队列，等到执行。所以线程执行时间为2+2=4秒。

6.4 invokeAny

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    String str = executor.invokeAny(Arrays.asList(
            () -> {
                log.debug("start");
                Thread.sleep(1000);
                log.debug("end");
                return "s1";
            },
            () -> {
                log.debug("start");
                Thread.sleep(500);
                log.debug("end");
                return "s2";
            },
            () -> {
                log.debug("start");
                Thread.sleep(2000);
                log.debug("end");
                return "s3";
            }
    ));
    log.debug(str);
}

可以看到s2方法最先执行完，然后返回，其他的俩个任务不在执行。

7.关闭线程池

7.1 shutdown

/*
线程池状态变为 SHUTDOWN
- 不会接收新任务
- 但已提交任务会执行完
- 此方法不会阻塞调用线程的执行
*/
void shutdown();

7.2 shutdownNow

/*
线程池状态变为 STOP
- 不会接收新任务
- 会将队列中的任务返回
- 并用 interrupt 的方式中断正在执行的任务
*/
List<Runnable> shutdownNow();

7.3 常用方法

// 不在 RUNNING 状态的线程池，此方法就返回 true
boolean isShutdown();
// 线程池状态是否是 TERMINATED
boolean isTerminated();
// 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

8. 处理线程池异常

8.1 在任务中抛出

public static void main(String[] args) throws Exception, InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(1);
    pool.execute(()->{
        try {
            log.debug("start");
            int i=1/0;
        }catch (Exception e){
            e.printStackTrace();
        }
    });
}

我们可以在我们认为会出现异常的地方通过try catch块捕获。这种方式可以抛出异常。

8.2通过Future得到异常信息

public static void main(String[] args) throws Exception, InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(1);
    Future<Boolean> start = pool.submit(() -> {
        try {
            log.debug("start");
            int i = 1 / 0;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    });
    log.debug(start.get()+"");
}

这种方式如果线程正常执行的话就会正常得到返回结果，如果出现异常，返回的就是异常信息。