1. 线程池介绍

线程池（Thread pool）是一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。

线程池的优点：

• 降低资源消耗

  通过重复利用自己已创建的线程降低线程创建和销毁造成的消耗。

• 提高响应速度

  当任务到达时，任务可以不需要等到线程创建就能立即执行。

• 提高线程的可管理性

  线程时稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池就可以进行统一的分配、调度和监控。

2. 标准库中的线程池使用

2.1 ThreadPoolExecutor

2.1.1 构造方法和参数说明

Java 标准库中内置了线程池的实现类 ThreadPoolExecutor，不过这个类的使用比较复杂，构造方法有如下四个：

这些构造方法中都有很多参数，汇总如下：

2.1.2 线程工厂

线程工厂 threadFactory 是用来执行线程池创建新线程的方式的，由同一个 threadFactory 创建的线程，属于同一个 ThreadGroup，创建的线程优先级都为 Thread.NORM_PRIORITY，以及是非守护进程状态。threadFactory创建的线程是采用 new Thread() 方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）。

线程工厂指定创建线程的方式，需要实现 ThreadFactory 接口，并重写 newThread(Runnable r) 方法。线程工厂这个参数可以不用指定，Executors 框架已经为我们实现了一个默认的线程工厂：

private 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;
    }
}
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

2.1.3 拒绝策略

拒绝策略是当线程和队列都满了的时候，即线程池饱和的时候，要执行的策略。拒绝策略需要实现 RejectedExecutionHandler 接口，并重写 rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法。拒绝策略这个参数可以不用指定，Executors 框架已经实现了如下4种拒绝策略：

2.1.4 关闭线程池

2.1.5 代码示例

public class Demo17 {
    // 创建线程池
    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            4, // 核心线程数
            10, // 最大线程数
            300,  // 线程闲置超时时长
            TimeUnit.SECONDS, // 时间单位
            new LinkedBlockingQueue<>()); // 任务队列

    public static void main(String[] args) {
        // 向线程池提交任务
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                // 线程执行的具体任务
            }
        });
        
        // 关闭线程
        threadPool.shutdownNow();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

2.2 Executors

2.2.1 四种功能线程池

Java 标准库中也提供了一个封装版本的线程池 Executors，这个类对 ThreadPoolExecutor 的复杂的参数进行了不同的封装，最终提供了四个静态方法，通过以下四种方法就能够创建出具体策略的线程池：

以上四种创建线程的方法的线程工厂的默认类为 DefaultThreadFactory，默认的拒绝策略为 ThreadPoolExecutor.AbortPolicy。

2.2.2 代码示例

public class Demo18 {

    public static void main(String[] args) {
        // 创建一个线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(10);

        // 通过 submit 将任务添加到线程池中
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("具体任务");
            }
        });
        
        // 关闭线程
        threadPool.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2.2.3 Executors 4 个功能线程的弊端

• newFixedThreadPool 和 newSingleThreadExecutor：主要问题是堆积的请求处理队列均采用 LinkedBlockingQueue，可能会耗费非常大的内存，甚至 OOM。
• newCachedThreadPool 和 newScheduledThreadPool：主要问题是线程数最大是 Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至 OOM。

OOM（Out Of Memory）表示内存用完了。当 JVM 因为没有足够的内存来为对象分配空间并且垃圾回收器也已经没有空间可回收时，就会抛出这个 java.lang.OutOfMemoryError。

3. 实现线程池

实现线程池，需要完成以下几个工作：

1. 通过阻塞队列来组织要完成的任务
2. 实现一个将任务注册到线程池的任务队列中的方法
3. 创建一个工作线程类，能够获取到任务队列，并能够反复的从任务队列中获取任务来执行
4. 通过数据结构将工作线程组织起来，并通过线程池的构造方法来确定工作线程的个数

class MyThreadPool {
    // 描述任务，直接使用 Runnable
    // 组织任务，使用一个阻塞队列来保存若干任务
    private BlockingQueue<Runnable> runnables = new LinkedBlockingQueue<>();

    // 描述一个工作线程
    static class Worker extends Thread {
        private BlockingQueue<Runnable> runnables = null;

        public Worker(BlockingQueue<Runnable> runnables){
            this.runnables = runnables;
        }

        @Override
        public void run(){
            // 一个线程要做的工作就是反复的从队列中获取任务，执行任务
            while (true){
                try {
                    // 如果任务队列为空，就会产生阻塞，直到有元素加入
                    Runnable take = runnables.take();
                    take.run();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // 组织若干个工作线程
    private List<Worker> workerList = new ArrayList<>();

    // 通过构造方法，指定线程池的个数
    public MyThreadPool(int n){
        for(int i=0; i<n; i++){
            Worker worker = new Worker(runnables);
            // 开启每个线程
            worker.start();
            workerList.add(worker);
        }
    }

    // 将任务注册到线程池中
    public void submit(Runnable runnable){
        try {
            runnables.put(runnable);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50