自定义线程池1.2

自定义线程池 1.2

1. 简介

上次我们实现了 1.1 版本，将线程池中的线程数量交给使用者决定，并且将线程的创建延迟到任务提交的时候，在本文中我们将对这个版本进行如下的优化：

• 在新建线程时交给线程一个任务。
• 让线程在某种情况下停止。

2. 优化一

在 1.1 版本的实现中，我们使用阻塞队列作为媒介，将 提交任务 和 执行任务 变成 异步 的，和 消息队列 类似，将消息的 生产 和 消费 变成 异步 的，俗称 解耦。

好处是 降低了系统的耦合性，增强了系统的扩展性。对于消息队列而言，扩展性在于支持 同一消费组中的多个消费者同时订阅同一个主题，进行 动态地水平扩展，从而增加整个消费组消费消息的性能；对于线程池而言，扩展性在于支持 多个线程同时“订阅”同一个阻塞队列，在运行过程中增加线程池处理任务的能力。

但随之而来的缺点就是 存在一定的延迟，这是异步一定会产生的问题。对于同步而言，提交任务和执行任务是直接的方法调用：submit -> execute；而对于异步而言，提交任务和执行任务通过阻塞队列解耦，流程会变成这样：submit -> taskQueue, taskQueue -> execute，从 submit 到 execute 会有一定的延迟。

注：当阻塞队列为空时，从 offer() 到 take() 是有一定延迟的。因为在阻塞队列的实现中，例如 ArrayBlockingQueue，offer() 和 take() 会共用一个条件变量 notEmpty：

• 当 take() 时，如果发现队列中没有任务，则会调用 notEmpty.await() 阻塞当前线程。
• 当 offer() 时，会调用 notEmpty.signal() 来唤醒所有使用 notEmpty.await() 阻塞的线程。

新建一个线程后，这个线程就调用 take() 阻塞起来，之后执行到 offer() 往队列中增加一个任务后，才唤醒了这个线程，延迟就存在于 阻塞到唤醒 的过程。

对于线程池来说，为了在运行时动态增加线程的数量，异步是必需的，但是当提交新任务时，希望尽量能减少提交到执行的延迟，应该怎么做呢？

---

揭晓答案：给线程一个初始任务 initialTask，线程一旦启动就去执行这个初始任务，然后再进入无限循环，从阻塞队列中领取任务。

此外，我们需要注意到一个细节：如果我们将初始任务作为线程的一个成员字段 initialTask，在构造方法中传入，在 run() 方法中执行，那么在 run() 方法中，当初始任务执行完之后，这个初始任务就再也不会用到了，如果 一直保留对其的引用，则会造成 内存泄漏，所以我们应该 在执行完初始任务后将 initialTask 置为空。

于是我们得到了如下的任务类：

/*** 线程池中的线程执行的任务，永不结束*/
private final class Worker implements Runnable {/*** 线程执行的初始任务*/private Runnable initialTask;public Worker(Runnable initialTask) {this.initialTask = initialTask;}@Overridepublic void run() {initialTask.run();initialTask = null; // help GCwhile (true) {Runnable t = null;try {// 取出任务，这步操作是阻塞的，如果线程没有取到任务，则会一直阻塞在这里t = taskQueue.take();} catch (InterruptedException e) {throw new RuntimeException(e);}t.run();}}
}

3. 优化二

上面提到过，在提交任务时，调用 offer() 会唤醒 所有 阻塞在 take() 中的线程。如果有很多线程都在阻塞，当提交任务时，这些线程都会被唤醒，去争抢同一个任务，但只有一个线程能够抢到，而其他线程被唤醒后再次在 take() 中阻塞。假如这些线程的数量十分庞大，那么将会消耗很多性能。

这种情况也算做 线程饥饿，不过线程饥饿真正的定义是 在循环中长时间抢不到锁，线程一直处于 运行 状态，而线程池中的线程大部分时间处于在 take() 中的阻塞状态，只有提交任务才会唤醒它们。

所以想到 线程长时间领不到任务就停止运行，于是得到了如下的任务类：

/*** 线程池中的线程执行的任务，在指定时间内无法获取到任务时结束*/
private final class Worker implements Runnable {/*** 线程执行的初始任务*/private Runnable initialTask;public Worker(Runnable initialTask) {this.initialTask = initialTask;}/*** 对在 {@link #threadPool} 中的线程的引用，用于在线程退出时将线程从 {@link #threadPool} 中移除*/private Thread threadInPool;public void setThreadInPool(Thread threadInPool) {this.threadInPool = threadInPool;}@Overridepublic void run() {initialTask.run();initialTask = null; // help GCwhile (true) {Runnable t = null;try {// 取出任务，这步操作是阻塞的，如果线程没有取到任务，则会在这里阻塞 3st = taskQueue.poll(3, TimeUnit.SECONDS);} catch (InterruptedException e) {throw new RuntimeException(e);}// 检查是否获取到任务了，如果没有则退出循环，停止运行if (t == null) {break;}t.run();}if (threadInPool == null) {throw new RuntimeException("没有设置线程");}// 从线程池中移除当前线程synchronized (threadPoolMonitor) {threadPool.remove(threadInPool);}}
}

4. 实现 1.2 版本

public class ThreadPool1_2 {/*** 线程池中线程的数量*/private final int poolSize;/*** 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高*/private final Set<Thread> threadPool = new HashSet<>();/*** 线程池的管程* <p>* 用于保证 <strong>获取线程池大小</strong>、<strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性*/private final Object threadPoolMonitor = new Object();/*** 任务队列，大小为 10*/private final BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(10);/*** 线程池中的线程执行的任务，在指定时间内无法获取到任务时结束*/private final class Worker implements Runnable {/*** 线程执行的初始任务*/private Runnable initialTask;public Worker(Runnable initialTask) {this.initialTask = initialTask;}/*** 对在 {@link #threadPool} 中的线程的引用，用于在线程退出时将线程从 {@link #threadPool} 中移除*/private Thread threadInPool;public void setThreadInPool(Thread threadInPool) {this.threadInPool = threadInPool;}@Overridepublic void run() {initialTask.run();initialTask = null; // help GCwhile (true) {Runnable t = null;try {// 取出任务，这步操作是阻塞的，如果线程没有取到任务，则会在这里阻塞 3st = taskQueue.poll(3, TimeUnit.SECONDS);} catch (InterruptedException e) {throw new RuntimeException(e);}// 检查是否获取到任务了，如果没有则退出循环，停止运行if (t == null) {break;}t.run();}if (threadInPool == null) {throw new RuntimeException("没有设置线程");}// 从线程池中移除当前线程synchronized (threadPoolMonitor) {threadPool.remove(threadInPool);}}}/*** 构造一个线程池** @param poolSize 线程池中的线程数量*/public ThreadPool1_2(int poolSize) {this.poolSize = poolSize;}/*** 提交任务** @param task 待执行的任务*/public void submit(Runnable task) {synchronized (threadPoolMonitor) {// 如果线程数量小于最大线程数量，则新建一个线程执行任务，然后直接返回if (threadPool.size() < poolSize) {Worker taskRunInThread = new Worker(task);Thread thread = new Thread(taskRunInThread);taskRunInThread.setThreadInPool(thread);threadPool.add(thread);thread.start();return;}}// 否则将任务放到任务队列中，等待空闲线程执行taskQueue.offer(task);}/*** 获取当前线程池中的线程数量** @return 当前线程池中的线程数量*/public int getCurrPoolSize() {synchronized (threadPoolMonitor) {return threadPool.size();}}
}

注意：本实现中将 threadPool 的类型从 ArrayList 变成了 HashSet，因为 Set.remove() 的性能更高。

5. 测试程序

public class ThreadPool1_2Test {/*** 测试线程池 1.2 版本的基本功能*/@Testpublic void test() throws InterruptedException {final int taskSize = 10;CountDownLatch latch = new CountDownLatch(taskSize);ThreadPool1_2 threadPool = new ThreadPool1_2(5);LogUtil.infoWithTimeAndThreadName("提交任务前");for (int i = 0; i < taskSize; i++) {int finalI = i;threadPool.submit(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}LogUtil.infoWithTimeAndThreadName("正在执行任务" + finalI);latch.countDown();});}LogUtil.infoWithTimeAndThreadName("提交任务后");// 等待测试结束latch.await();LogUtil.infoWithTimeAndThreadName("任务执行完毕");}/*** 测试线程池 1.2 版本的 线程 3s 领不到任务就停止 的功能*/@Testpublic void testBreak() throws InterruptedException {final int taskSize = 10;CountDownLatch latch = new CountDownLatch(taskSize);ThreadPool1_2 threadPool = new ThreadPool1_2(5);LogUtil.infoWithTimeAndThreadName("提交任务前");for (int i = 0; i < taskSize; i++) {threadPool.submit(() -> latch.countDown());}LogUtil.infoWithTimeAndThreadName("提交任务后");// 等待测试结束latch.await();LogUtil.infoWithTimeAndThreadName("任务执行完毕");// 等待 3.1s，所有线程都停止，多出来的 100ms 是为了避免延迟导致线程停止较晚Thread.sleep(3100);LogUtil.infoWithTimeAndThreadName("当前线程数量为：" + threadPool.getCurrPoolSize());}
}

6. 测试结果

6.1 test

00:19:31 [    main] 提交任务前
00:19:31 [    main] 提交任务后
00:19:32 [Thread-3] 正在执行任务3
00:19:32 [Thread-1] 正在执行任务1
00:19:32 [Thread-2] 正在执行任务2
00:19:32 [Thread-0] 正在执行任务0
00:19:32 [Thread-4] 正在执行任务4
00:19:33 [Thread-0] 正在执行任务8
00:19:33 [Thread-1] 正在执行任务6
00:19:33 [Thread-3] 正在执行任务5
00:19:33 [Thread-2] 正在执行任务7
00:19:33 [Thread-4] 正在执行任务9
00:19:33 [    main] 任务执行完毕

注：对于 test() 方法测试的结果解析，可以参考 1.0 版本。

6.2 testBreak

00:21:06 [    main] 提交任务前
00:21:06 [    main] 提交任务后
00:21:06 [    main] 任务执行完毕
00:21:09 [    main] 当前线程数量为：0

对于这个方法的测试结果，我们需要看到最后输出的 “当前线程数量为：0”，因为只要线程在 3s 内没有领到任务，就会停止，从而到达线程池中线程数量为 0 的效果。

7. 思考

• 在本实现中，threadPoolMonitor 的使用变得更频繁了，你能想出每次使用 threadPoolMonitor 加锁的原因是什么吗？
• 之前讲到“池化”思想是 提前创建/延迟删除，在 1.2 版本的 testBreak 测试中，我们发现最终线程池中没有一个线程，这与“池化”的思想有悖。你能调整一下设计，使新的设计 在一定程度上 满足“池化”思想吗？（这里如果有很大疑惑，说明我之前讲的池化思想太简略了，下次实现新版本会补充）

8. 总结

本次我们实现了 1.2 版本，优化一基于 异步必定有延迟 的思想，优化二基于 减少线程无效唤醒 的思想。

9. 补充

当我在实现更高版本时，我发现当前版本还存在另一种设计，如下所示：

public class ThreadPool1_2 {/*** 线程池中线程的数量*/private final int poolSize;/*** 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高*/private final Set<Worker> threadPool = new HashSet<>();/*** 线程池的管程* <p>* 用于保证 <strong>获取线程池大小</strong>、<strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性*/private final Object threadPoolMonitor = new Object();/*** 任务队列，大小为 10*/private final BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(10);/*** 线程池中的线程执行的任务，在指定时间内无法获取到任务时结束*/private final class Worker implements Runnable {/*** 线程执行的初始任务*/private Runnable initialTask;/*** 对 <strong>真正运行的线程</strong> 的引用，用于调用其 {@link Thread#start()} 方法启动线程*/private final Thread actuallyRunningThread;public Worker(Runnable initialTask) {this.initialTask = initialTask;this.actuallyRunningThread = new Thread(this);threadPool.add(this);}@Overridepublic void run() {initialTask.run();initialTask = null; // help GCwhile (true) {Runnable t = null;try {// 取出任务，这步操作是阻塞的，如果线程没有取到任务，则会在这里阻塞 3st = taskQueue.poll(3, TimeUnit.SECONDS);} catch (InterruptedException e) {throw new RuntimeException(e);}// 检查是否获取到任务了，如果没有则退出循环，停止运行if (t == null) {break;}t.run();}// 从线程池中移除当前线程synchronized (threadPoolMonitor) {threadPool.remove(this);}}/*** 启动内部保存的线程*/public void start() {actuallyRunningThread.start();}}/*** 构造一个线程池** @param poolSize 线程池中的线程数量*/public ThreadPool1_2(int poolSize) {this.poolSize = poolSize;}/*** 提交任务** @param task 待执行的任务*/public void submit(Runnable task) {synchronized (threadPoolMonitor) {// 如果线程数量小于最大线程数量，则新建一个线程执行任务，然后直接返回if (threadPool.size() < poolSize) {Worker worker = new Worker(task);worker.start();return;}}// 否则将任务放到任务队列中，等待空闲线程执行taskQueue.offer(task);}/*** 获取当前线程池中的线程数量** @return 当前线程池中的线程数量*/public int getCurrPoolSize() {synchronized (threadPoolMonitor) {return threadPool.size();}}
}

注：这两种实现的核心区别在于放到 threadPool 中的具体是什么对象，第一种实现将 Thread 对象放到了 threadPool 中，第二种实现将 Worker 对象放到了 threadPool 中，从而也导致了移除逻辑的不同。第二种实现不需要在移除时判断 actuallyRunningThread 是否为空，因为它移除的是 this。