线程池认识

1. 线程池的分类

从以下代码可知，线程池分为5种：

1. FixedThreadPool
2. SingleThreadExecutor
3. CachedThreadPool
4. SingleThreadScheduleExecutor
5. ScheduledThreadPool

其中前3个属于ThreadPoolExecutor，后2个属于ScheduledThreadPoolExecutor；

public static ExecutorService newFixedThreadPool(int var0) {
        return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());
    }
public static ExecutorService newSingleThreadExecutor() {
    return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()));
}
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue());
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new Executors.DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newScheduledThreadPool(int var0) {
    return new ScheduledThreadPoolExecutor(var0);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2. 线程池的创建

不管是ThreadPoolExecutor还是SchedduledThreadPoolExecutor，最终都是需要经过ThreadPoolExecutor

/**
corePoolSize：该线程池中核心线程数最大值
maximumPoolSize：该线程池中线程总数的最大值
keepAliveTime：该线程池中非核心线程闲置超时时长
unit：上面时间的单位
workQueue：保存等待执行任务的阻塞队列
threadFactory：可以通过线程工厂给创建出来的线程设置更加有意义的名字
handler：饱和策略。默认为AbortPolicy：表示无法处理新任务时抛出异常
*/
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;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3. 关闭线程池

有两个方法可以关闭线程池，shutdown和shutdownNow方法

原理：遍历线程池中的工作线程，然后逐个调用线程的interrupt方法，但是根据前面说的中断线程的方法中，如果线程使用sleep()或wait()方法进入了就绪状态，那么可以使用interrupt()方法是线程离开run()方法，同时结束线程。所以遇到轮询线程，有可能会无法中止。

shutdown是将线程池的状态设置为SHUTDOWN状态，然后遍历任务表，挨个调用线程的interrupt方法

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}
1
2
3
4
5
6
7
8
9
10
11
12
13

shutdownNow是将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表

public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

线程池在项目中是需要持续工作的全局场景，可以结合具体的场景而定，一般不建议手动关闭线程池

4. 阻塞队列

BlockingQueue中的几个具体API介绍：

offer(E e)：将给定的元素设置到队列中，有返回值true/false；如果限定了长度的队列中设置值，推荐使用

add(E e)：将给定的元素设置到队列中，成功返回true，否则抛异常

put(E e)：将给定的元素设置到队列中，若没有多余空间，则阻塞，直到有多余空间

take()：取，如果没有值，则阻塞

poll()：取，如果没有值，则抛异常

poll(long timeOut, TimeUnit unit)：取，给定时间没取出来，抛异常

一般来说，workQueue有以下4种队列类型：

SynchronousQueue：同步队列，不保留任务直接提交给线程处理，线程如果不够就会报错，所以会将maximumPoolSize指定成Integer.MAX_VALUE，即无限大，去规避这个风险

LinkedBlockingQueue：链表阻塞队列，当前线程数<核心线程数，则新建核心线程处理任务；当前线程数=核心线程数，则进入队列等待。这个队列没有最大值限制，即超过核心线程数的任务都将被添加到队列中，这也就导致maximumPoolSize设定失效，因为总线程数永远不会超过corePoolSize

ArrayBlockingQueue：数组阻塞队列，由于是数组可以限定大小，接收到任务的时候，如果没有达到corePoolSize的值，则新建线程(核心线程)执行任务，如果达到了，则入队等候，如果队列已满，则新建线程(非核心线程)执行任务，又如果总线程数到了maximumPoolSize，并且队列也满了，则发生错误

DelayQueue：延迟队列，队列内元素必须实现Delayed接口，这就意味着你传进去的任务必须先实现Delayed接口。这个队列接收到任务时，首先先入队，只有达到了指定的延时时间，才会执行任务

实际上，Java中的Executors已经为我们提供了4种线程池：

newFixedThreadPool：创建一个定长线程池，因为采用LinkedBlockingQueue，所以可以控制最大并发数

newCachedThreadPool：利用SynchronousQueue特性创建，都是非核心线程

newScheduledThreadPool：创建一个定长任务线程池，支持定时及周期性任务执行，DelayedWorkQueue这边不需要其中的元素实现Delayed接口

// Executors.java
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
// ScheduledThreadPoolExecutor.java
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}
1
2
3
4
5
6
7
8
9
10

newSingleThreadExecutor：创建1个单线程的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序（FIFO，LIFO，优先级）执行，遵循LinkedBlockingQueue规则（无大小限制，保持等待）