6. ThreadGroup详解

6.1 ThreadGroup和Thread

新的线程都会被加入到main线程所在的group中, main线程的group名字同线程名。ThreadGroup同样存在父子关系. 如下:
Java高并发编程详解(一) - 图1

6.2 创建ThreadGroup

  1. package com.zoro.concurrent.chapter06;
  3. /**
  4.  * 创建ThreadGroup
  5.  *
  6.  * @author yx.jiang
  7.  * @date 2021/7/27 10:26
  8.  */
  9. public class ThreadGroupCreator {
  11.     public static void main(String[] args) {
  12.         ThreadGroup currentGroup = Thread.currentThread().getThreadGroup();
  13.         //currentGroup = java.lang.ThreadGroup[name=main,maxpri=10]
  14.         System.out.println("currentGroup = " + currentGroup);
  16.         ThreadGroup group1 = new ThreadGroup("Group1");
  18.         //输出true
  19.         System.out.println(group1.getParent() == currentGroup);
  21.         ThreadGroup group2 = new ThreadGroup(group1, "Group2");
  22.         //输出true
  23.         System.out.println(group2.getParent() == group1);
  24.     }
  25. }

6.3 复制Thread数组和ThreadGroup数组

6.3.1 复制Thread数组

前两个方法会将ThreadGroup中的active线程全部复制到Thread数组中
recurse = true; 递归复制——将所有子group中的active线程都递归到Thread数组中
recurse = false; 非递归复制
image.png

  • enumerate获取的仅仅是个预估值, 并不能100%地保证当前group的活跃线程.如: 在调用复制之后, 某个线程结束了生命周期/新的线程加入了进来, 均会导致数据的不准确

  • enumerate方法的返回值int较Thread[]的长度更为真实, 如: 定义了数组长度的Thread数组, 那么enumerate仅仅会将当前活跃的thread分别放进数组中, 而int代表真是的数量, 而并非数组的长度;

    6.3.2 复制ThreadGroup数组

    image.png

    6.4 ThreadGroup操作

    ThreadGroup不能提供对线程的管理, 主要功能是对线程 进行组织.

    6.4.1 ThreadGroup基操

    image.png
    image.png
    略.

    6.4.2 ThreadGroup的interrupt

    interrupt一个ThreadGroup会导致该group中所有的active线程都被interrupt, 即: 该group中每个线程的interrupt标识都被设置了

    6.4.3 ThreadGroup的destroy

    destroy用于销毁ThreadGroup, 只针对一个没有任何active线程的group进行一次destroy标记, 调用该方法的直接结果是在父group中将自己移除

    6.4.4 守护ThreadGroup

    像Thread一样, 但是将ThreadGroup设置为daemon, 并不会影响线程的daemon属性; 如果一个ThreadGroup的daemon被设置为true, 那么在group中没有任何active线程的时候该group将自动destroy

    7. Hook线程以及捕获线程异常执行

    7.1 获取线程运行时异常

  • _public static void _setDefaultUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh)

  • _public static _Thread.UncaughtExceptionHandler getDefaultUncaughtExceptionHandler()
  • _public _Thread.UncaughtExceptionHandler getUncaughtExceptionHandler()
  • _public void _setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh)

image.png
image.png

7.1.1 UncaughtExceptionHandler介绍

线程在执行单元中是不允许抛出checked异常的, 且线程运行在自己的上下文中, 派生它的线程将无法直接获得它运行中出现的异常信息; UncaughtExceptionHandler 接口: 当线程在运行过程中出现异常时, 会回调UncaughtExceptionHandler 接口, 从而得知是哪个线程在运行时出错, 以及出现了什么样的错误

java.lang.Thread#dispatchUncaughtException()调用了java.lang.Thread.UncaughtExceptionHandler#uncaughtException()
实际运行过程中出现异常, JVM会调用dispatchUncaughtException()方法, 该方法会将对应的线程实例以及异常信息传递给回调接口.

  1.     @FunctionalInterface
  2.     public interface UncaughtExceptionHandler {
  3.         /**
  4.          * Method invoked when the given thread terminates due to the
  5.          * given uncaught exception.
  6.          * <p>Any exception thrown by this method will be ignored by the
  7.          * Java Virtual Machine.
  8.          * @param t the thread
  9.          * @param e the exception
  10.          */
  11.         void uncaughtException(Thread t, Throwable e);
  12.     }
  15.     /**
  16.      * Dispatch an uncaught exception to the handler. This method is
  17.      * intended to be called only by the JVM.
  18.      */
  19.     private void dispatchUncaughtException(Throwable e) {
  20.         getUncaughtExceptionHandler().uncaughtException(this, e);
  21.     }

7.1.2 实例

  1. package com.zoro.concurrent.chapter07;
  3. import java.util.concurrent.TimeUnit;
  5. /**
  6.  * 捕获线程异常
  7.  *
  8.  * @author yx.jiang
  9.  * @date 2021/7/27 10:26
  10.  */
  11. public class CaptureThreadException {
  13.     /**
  14.      * 输出如下:
  15.      * Test-Threadoccur exception: 
  16.      * java.lang.ArithmeticException: / by zero
  17.      *     at com.zoro.concurrent.chapter07.CaptureThreadException.lambda$main$1(CaptureThreadException.java:27)
  18.      *     at java.lang.Thread.run(Thread.java:748)
  19.      */
  20.     public static void main(String[] args) {
  21.         //设置回调接口
  22.         Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
  23.             System.out.println(t.getName() + "occur exception: ");
  24.             e.printStackTrace();
  25.         });
  27.         final Thread thread = new Thread(() -> {
  28.             try {
  29.                 TimeUnit.SECONDS.sleep(2);
  30.             } catch (InterruptedException e) {
  31.                 //e.printStackTrace();
  32.             }
  33.             //unchecked异常
  34.             System.out.println(1 / 0);
  35.         }, "Test-Thread");
  37.         thread.start();
  38.     }
  39. }

7.1.3 分析sourceCode

java.lang.Thread#getUncaughtExceptionHandler

  1.     public UncaughtExceptionHandler getUncaughtExceptionHandler() {
  2.         return uncaughtExceptionHandler != null ?
  3.             uncaughtExceptionHandler : group;
  4.     }

java.lang.ThreadGroup#uncaughtException

  1.     public void uncaughtException(Thread t, Throwable e) {
  2.         if (parent != null) {
  3.             parent.uncaughtException(t, e);
  4.         } else {
  5.             Thread.UncaughtExceptionHandler ueh =
  6.                 Thread.getDefaultUncaughtExceptionHandler();
  7.             if (ueh != null) {
  8.                 ueh.uncaughtException(t, e);
  9.             } else if (!(e instanceof ThreadDeath)) {
  10.                 System.err.print("Exception in thread \""
  11.                                  + t.getName() + "\" ");
  12.                 e.printStackTrace(System.err);
  13.             }
  14.         }
  15.     }
  • 该ThreadGroup如果有父ThreadGroup, 则直接调用父ThreadGroup的uncaughtException方法
  • 如果设置了全局默认的UncaughtExceptionHandler, 则调用uncaughtException方法
  • 若既没有父ThreadGroup, 又没有设置全局默认的UncaughtExceptionHandler, 则会直接将异常的堆栈信息定向到System.err中 ```java package com.zoro.concurrent.chapter07;

import java.util.concurrent.TimeUnit;

/**

  • @author yx.jiang
  • @date 2021/7/27 10:26 */ public class EmptyExceptionHandler {
  1. /**
  2.  * 输出如下:
  3.  * mainGroup.getName() = main
  4.  * mainGroup.getParent() = java.lang.ThreadGroup[name=system,maxpri=10]
  5.  * mainGroup.getParent().getParent() = null
  6.  * Exception in thread "Test-Thread" java.lang.ArithmeticException: / by zero
  7.  *     at com.zoro.concurrent.chapter07.EmptyExceptionHandler.lambda$main$0(EmptyExceptionHandler.java:29)
  8.  *     at java.lang.Thread.run(Thread.java:748)
  9.  */
  10. public static void main(String[] args) {
  11.     //设置回调接口
  12.     ThreadGroup mainGroup = Thread.currentThread().getThreadGroup();
  13.     System.out.println("mainGroup.getName() = " + mainGroup.getName());
  14.     System.out.println("mainGroup.getParent() = " + mainGroup.getParent());
  15.     System.out.println("mainGroup.getParent().getParent() = " + mainGroup.getParent().getParent());
  17.     final Thread thread = new Thread(() -> {
  18.         try {
  19.             TimeUnit.SECONDS.sleep(2);
  20.         } catch (InterruptedException e) {
  21.             //e.printStackTrace();
  22.         }
  23.         //unchecked异常
  24.         System.out.println(1 / 0);
  25.     }, "Test-Thread");
  27.     thread.start();
  28. }

}

  1. ![](https://cdn.nlark.com/yuque/0/2021/jpeg/708204/1627541129891-79306ddd-43b2-4353-8ce9-efc5828d0948.jpeg)
  2. <a name="ac0bT"></a>
  3. ### 7.2 注入钩子线程
  4. <a name="t3szI"></a>
  5. #### 7.2.1 Hook线程介绍
  6. ```java
  7. package com.zoro.concurrent.chapter07;
  9. import java.util.concurrent.TimeUnit;
  11. /**
  12.  * 钩子线程
  13.  *
  14.  * @author yx.jiang
  15.  * @date 2021/7/27 10:26
  16.  */
  17. public class ThreadHook {
  19.     /**
  20.      * 输出如下:
  21.      * 程序即将关闭...
  22.      * 钩子线程1启动...
  23.      * 钩子线程2启动...
  24.      * 钩子线程1即将关闭
  25.      * 钩子线程2即将关闭
  26.      */
  27.     public static void main(String[] args) {
  28.         //注册第一个钩子线程
  29.         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  30.             System.out.println("钩子线程1启动...");
  31.             try {
  32.                 TimeUnit.SECONDS.sleep(2);
  33.             } catch (InterruptedException e) {
  34.                 e.printStackTrace();
  35.             }
  36.             System.out.println("钩子线程1即将关闭");
  37.         }));
  39.         //注册第二个钩子线程
  40.         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  41.             System.out.println("钩子线程2启动...");
  42.             try {
  43.                 TimeUnit.SECONDS.sleep(2);
  44.             } catch (InterruptedException e) {
  45.                 e.printStackTrace();
  46.             }
  47.             System.out.println("钩子线程2即将关闭");
  48.         }));
  50.         System.out.println("程序即将关闭...");
  51.     }
  52. }

7.2.2 Hook实战

  1. package com.zoro.concurrent.chapter07;
  3. import java.io.File;
  4. import java.io.IOException;
  5. import java.nio.file.Files;
  6. import java.nio.file.Path;
  7. import java.nio.file.Paths;
  8. import java.nio.file.attribute.PosixFilePermission;
  9. import java.nio.file.attribute.PosixFilePermissions;
  10. import java.util.Set;
  11. import java.util.concurrent.TimeUnit;
  13. /**
  14.  * 防止重复
  15.  * <h>防止某个程序重复启动, 在进程启动时创建一个.lock文件, 进程收到中断信号的时候会删除这个lock文件</h>
  16.  *
  17.  * @author yx.jiang
  18.  * @date 2021/7/27 10:26
  19.  */
  20. public class PreventDuplicated {
  22.     private static final String LOCK_PATH = "D:\\idea tool box\\locks";
  24.     private static final String LOCK_FILE = ".lock";
  26.     private static final String PERMISSION = "rw-------";
  28.     public static void main(String[] args) throws IOException {
  29.         //注入Hook线程, 在程序退出时删除lock文件
  30.         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  31.             System.out.println("程序接收到 kill 信号.");
  32.             getLockFile().toFile().delete();
  33.         }));
  35.         //检查是否存在.lock文件
  36.         checkRunning();
  38.         //简单模拟当前进程正在运行
  39.         for (; ; ) {
  40.             try {
  41.                 TimeUnit.SECONDS.sleep(1);
  42.                 System.out.println("程序正在运行...");
  43.             } catch (InterruptedException e) {
  44.                 e.printStackTrace();
  45.             }
  46.         }
  47.     }
  49.     public static void checkRunning() throws IOException {
  50.         Path path = getLockFile();
  51.         if (path.toFile().exists()) {
  52.             throw new RuntimeException("程序正在运行...");
  53.         }
  55.         //WINDOWS
  56.         Path file1 = Files.createFile(path);
  57.         File file = file1.toFile();
  58.         file.setReadable(true);
  59.         file.setWritable(true);
  60.         System.out.println(file.exists() + "  " + file.getAbsolutePath());
  62. //        'posix:permissions' not supported as initial attribute
  63. //        UNIX
  64. //        Set<PosixFilePermission> perms = PosixFilePermissions.fromString(PERMISSION);
  65. //        Files.createFile(path, perms);
  67.     }
  69.     public static Path getLockFile() {
  70.         return Paths.get(LOCK_PATH, LOCK_FILE);
  71.     }
  72. }

image.png

7.2.3 注意事项

  • Hook线程只有在收到退出信号的时候会被执行, 如果在kill的时候使用了参数 **-9**, 那么Hook线程不会得到执行, 进程将会立即退出, 因此.lock文件将得不到清理.
  • Hook线程中也可以执行一些资源的释放工作, 比如关闭文件句柄, socket链接, 数据库connection等.

  • 尽量不要在Hook线程中执行一些耗时非常长的操作, 因为会导致程序迟迟不能退出

    8. 线程池原理及自定义线程池

    8.1 线程池原理

  • 任务队列: 用于缓存提交任务

  • 线程数量管理功能: 一个线程池必须能够很好的管理和控制线程数量, 可通过如下三个参数来实现, 比如创建线程池时初始的线程数量init; 线程池自动扩充时最大的核心线程数量max; 在线程池空闲时需要释放线程但是也要维护一定数量的活跃数量或者核心数量core. init <= core <= max;
  • 任务拒绝策略: 如果线程数量已经到上限且任务队列已满, 则需要有相应的拒绝策略来通知任务提交者;
  • 线程工厂: 主要用于个性化定制线程, 比如将线程设置为守护线程以及设置线程名称等;
  • QueueSize: 任务队列主要用于存放提交的Runnable, 但是为了防止内存溢出, 需要有limit数量对其进行控制
  • keepedAlive时间: 该时间主要决定线程各个重要参数自动维护的时间间隔.

    8.2 线程池实现

    image.png

    ThreadPool

    ```java package com.zoro.concurrent.chapter08.threadpool;

/**

  • 线程池接口 *
  • @author yx.jiang

  • @date 2021/7/27 10:26 / public interface ThreadPool { /*

    • 提交任务到线程池 *

    • @param runnable runnable */ void execute(Runnable runnable);

      /**

    • 关闭并销毁线程池 */ void shutdown();

      /**

    • 获取线程池的初始化大小 *

    • @return */ int getInitSize();

      /**

    • 获取线程池最大的线程数 *

    • @return */ int getMaxSize();

      /**

    • 获取线程池的核心线程数量 *

    • @return */ int getCoreSize();

      /**

    • 获取线程池中用于缓存任务队列的大小 *

    • @return */ int getQueueSize();

      /**

    • 获取线程池中活跃线程的数量 *

    • @return */ int getActiveCount();

      /**

    • 查看线程池是否已经被销毁(shutdown) *
    • @return */ boolean isShutdown(); }
  1. <a name="KWWWf"></a>
  2. #### ThreadFactory
  3. ```java
  4. package com.zoro.concurrent.chapter08.threadpool;
  6. /**
  7.  * 创建Thread的工厂
  8.  *
  9.  * @author yx.jiang
  10.  * @date 2021/7/27 10:26
  11.  */
  12. @FunctionalInterface
  13. public interface ThreadFactory {
  14.     /**
  15.      * 创建线程
  16.      *
  17.      * @param runnable 执行单元/任务
  18.      * @return 线程
  19.      */
  20.     Thread createThread(Runnable runnable);
  21. }

RunnableQueue

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. /**
  4.  * 任务队列
  5.  *
  6.  * @author yx.jiang
  7.  * @date 2021/7/27 10:26
  8.  */
  9. public interface RunnableQueue {
  11.     /**
  12.      * 将任务提交至队列中
  13.      *
  14.      * @param runnable
  15.      */
  16.     void offer(Runnable runnable);
  18.     /**
  19.      * 工作线程通过take获取Runnable
  20.      */
  21.     Runnable take() throws InterruptedException;
  23.     /**
  24.      * 获取任务队列中任务的数量
  25.      *
  26.      * @return
  27.      */
  28.     int size();
  29. }

DenyPolicy

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. /**
  4.  * 拒绝策略
  5.  *
  6.  * @author yx.jiang
  7.  * @date 2021/7/27 10:26
  8.  */
  9. @FunctionalInterface
  10. public interface DenyPolicy {
  11.     /**
  12.      * 拒绝方法
  13.      *
  14.      * @param runnable   任务/执行单元
  15.      * @param threadPool 线程池
  16.      */
  17.     void reject(Runnable runnable, ThreadPool threadPool);
  19.     /**
  20.      * 直接丢弃Runnable任务
  21.      */
  22.     class DiscardDenyPolicy implements DenyPolicy {
  23.         @Override
  24.         public void reject(Runnable runnable, ThreadPool threadPool) {
  25.             //do nothing
  26.         }
  27.     }
  29.     /**
  30.      * 抛出{@link RunnableDenyException}异常
  31.      */
  32.     class AbortDenyPolicy implements DenyPolicy {
  33.         @Override
  34.         public void reject(Runnable runnable, ThreadPool threadPool) {
  35.             throw new RunnableDenyException("执行单元/任务: " + runnable + "将被放弃");
  36.         }
  37.     }
  39.     /**
  40.      * 交由调用者的线程直接运行Runnable, 而不会被加入到线程池中
  41.      */
  42.     class RunnerDenyPolicy implements DenyPolicy {
  43.         @Override
  44.         public void reject(Runnable runnable, ThreadPool threadPool) {
  45.             //do nothing
  46.         }
  47.     }
  48. }

RunnableDenyException

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. /**
  4.  * 可运行拒绝异常
  5.  *
  6.  * @author yx.jiang
  7.  * @date 2021/7/27 10:26
  8.  */
  9. public class RunnableDenyException extends RuntimeException {
  10.     public RunnableDenyException(String message) {
  11.         super(message);
  12.     }
  13. }

LinkedRunnableQueue

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. import java.util.LinkedList;
  5. /**
  6.  * 任务队列实现
  7.  *
  8.  * @author yx.jiang
  9.  * @date 2021/7/27 10:26
  10.  */
  11. public class LinkedRunnableQueue implements RunnableQueue {
  12.     /**
  13.      * 任务队列最大容量
  14.      */
  15.     private final int limit;
  17.     /**
  18.      * 任务队列满后,需执行拒绝策略
  19.      */
  20.     private final DenyPolicy denyPolicy;
  22.     private final LinkedList<Runnable> runnableList = new LinkedList<>();
  24.     private final ThreadPool threadPool;
  26.     public LinkedRunnableQueue(int limit, DenyPolicy denyPolicy, ThreadPool threadPool) {
  27.         this.limit = limit;
  28.         this.denyPolicy = denyPolicy;
  29.         this.threadPool = threadPool;
  30.     }
  32.     @Override
  33.     public void offer(Runnable runnable) {
  34.         synchronized (runnableList) {
  35.             if (runnableList.size() >= limit) {
  36.                 //无法容纳新的任务时执行拒绝策略
  37.                 denyPolicy.reject(runnable, threadPool);
  38.             } else {
  39.                 //将任务加入到队尾, 并唤醒阻塞中的线程
  40.                 runnableList.addLast(runnable);
  41.                 runnableList.notifyAll();
  42.             }
  43.         }
  44.     }
  46.     @Override
  47.     public Runnable take() throws InterruptedException {
  48.         synchronized (runnableList) {
  49.             while (runnableList.isEmpty()) {
  50.                 try {
  51.                     //如果任务队列中没有可执行的任务,则当前线程会挂起,进入runnableList关联的monitor waitSet 中等待唤醒(新的任务加入)
  52.                     runnableList.wait();
  53.                 } catch (InterruptedException e) {
  54.                     throw e;
  55.                 }
  56.             }
  57.             //从任务队列头部移除一个任务
  58.             return runnableList.removeFirst();
  59.         }
  60.     }
  62.     @Override
  63.     public int size() {
  64.         synchronized (runnableList) {
  65.             //返回当前任务队列中的任务数
  66.             return runnableList.size();
  67.         }
  68.     }
  69. }

InternalTask

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. /**
  4.  * 内部任务(线程池内)
  5.  *
  6.  * @author yx.jiang
  7.  * @date 2021/7/27 10:26
  8.  */
  9. public class InternalTask implements Runnable {
  11.     private final RunnableQueue runnableQueue;
  13.     private volatile boolean running = true;
  15.     public InternalTask(RunnableQueue runnableQueue) {
  16.         this.runnableQueue = runnableQueue;
  17.     }
  19.     @Override
  20.     public void run() {
  21.         //如果当任务为running并且没有被中断, 则其将不断的从queue中获取runnable, 然后执行run方法
  22.         while (running && !Thread.currentThread().isInterrupted()) {
  23.             try {
  24.                 Runnable task = runnableQueue.take();
  25.                 task.run();
  26.             } catch (InterruptedException e) {
  27.                 running = false;
  28.                 break;
  29.             }
  30.         }
  31.     }
  33.     /**
  34.      * 停止当前任务, 在线程池的shutdown方法中使用
  35.      */
  36.     public void stop() {
  37.         this.running = false;
  38.     }
  39. }

BasicThreadPool

  1. package com.zoro.concurrent.chapter08.threadpool;
  3. import java.util.ArrayDeque;
  4. import java.util.Queue;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.atomic.AtomicInteger;
  8. /**
  9.  * 初始化线程池
  10.  * FIXME: 这里不建议 extends Thread; 用组合的方式更好些;
  11.  *
  12.  * @author yx.jiang
  13.  * @date 2021/7/27 10:26
  14.  */
  15. public class BasicThreadPool extends Thread implements ThreadPool {
  17.     public static final String THREAD_POOL_DESTROYED = "线程池已销毁";
  18.     /**
  19.      * 初始化线程数量
  20.      */
  21.     private final int initSize;
  22.     /**
  23.      * 最大线程数量
  24.      */
  25.     private final int maxSize;
  26.     /**
  27.      * 核心线程数量
  28.      */
  29.     private final int coreSize;
  30.     /**
  31.      * 当前活跃线程数量
  32.      */
  33.     private int activeCount;
  34.     /**
  35.      * 创建线程所需工厂
  36.      */
  37.     private final ThreadFactory threadFactory;
  38.     /**
  39.      * 任务队列
  40.      */
  41.     private final RunnableQueue runnableQueue;
  42.     /**
  43.      * 线程池是否销毁
  44.      */
  45.     private volatile boolean isShutdown = false;
  46.     /**
  47.      * 工作队列
  48.      */
  49.     private final Queue<ThreadTask> threadQueue = new ArrayDeque<>();
  50.     /**
  51.      * 默认拒绝策略
  52.      */
  53.     private static final DenyPolicy DEFAULT_DENY_POLICY = new DenyPolicy.DiscardDenyPolicy();
  55.     /**
  56.      * 默认线程工厂
  57.      */
  58.     private static final ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory();
  60.     private final long keepAliveTime;
  62.     private final TimeUnit timeUnit;
  64.     public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory, int queueSize, DenyPolicy denyPolicy, long keepAliveTime, TimeUnit timeUnit) {
  65.         this.initSize = initSize;
  66.         this.maxSize = maxSize;
  67.         this.coreSize = coreSize;
  68.         this.threadFactory = threadFactory;
  69.         this.runnableQueue = new LinkedRunnableQueue(queueSize, denyPolicy, this);
  70.         this.keepAliveTime = keepAliveTime;
  71.         this.timeUnit = timeUnit;
  73.         this.init();
  74.     }
  76.     public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) {
  77.         this(initSize, maxSize, coreSize, DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENY_POLICY, 10, TimeUnit.SECONDS);
  78.     }
  80.     /**
  81.      * 初始化时 —— 创建initSize个线程
  82.      */
  83.     private void init() {
  84.         start();
  85.         for (int i = 0; i < initSize; i++) {
  86.             newThread();
  87.         }
  88.     }
  90.     private void newThread() {
  91.         //创建线程并启动
  92.         InternalTask internalTask = new InternalTask(runnableQueue);
  93.         Thread thread = threadFactory.createThread(internalTask);
  94.         ThreadTask threadTask = new ThreadTask(thread, internalTask);
  95.         threadQueue.add(threadTask);
  96.         this.activeCount++;
  97.         thread.start();
  98.     }
  100.     private void removeThread() {
  101.         //从线程池中移除某个线程
  102.         ThreadTask threadTask = threadQueue.remove();
  103.         threadTask.internalTask.stop();
  104.         this.activeCount--;
  105.     }
  107.     /**
  108.      * 同步代码块：为了阻止在线程维护过程中线程池销毁引起的数据不一致问题
  109.      */
  110.     @Override
  111.     public void run() {
  112.         //该方法继承自Thread，主要用于维护线程数量，如：扩容、回收工作
  113.         while (!isShutdown && !isInterrupted()) {
  114.             try {
  115.                 timeUnit.sleep(keepAliveTime);
  116.             } catch (InterruptedException e) {
  117.                 isShutdown = true;
  118.                 break;
  119.             }
  121.             synchronized (this) {
  122.                 if (isShutdown) {
  123.                     break;
  124.                 }
  125.                 //当前的队列中有任务尚未处理，且activeCount < coreSize，继续扩容
  126.                 if (runnableQueue.size() > 0 && activeCount < coreSize) {
  127.                     for (int i = initSize; i < coreSize; i++) {
  128.                         newThread();
  129.                     }
  130.                     //目的：不想让线程的扩容直接到达maxSize
  131.                     continue;
  132.                 }
  133.                 //当前的对列中有任务尚未处理，且activeCount < maxSize，继续扩容
  134.                 if (runnableQueue.size() > 0 && activeCount < maxSize) {
  135.                     for (int i = coreSize; i < maxSize; i++) {
  136.                         newThread();
  137.                     }
  138.                 }
  139.                 //若任务队列无任务，需要回收，回收至coreSize即可
  140.                 if (runnableQueue.size() == 0 && activeCount > coreSize) {
  141.                     for (int i = coreSize; i < activeCount; i++) {
  142.                         removeThread();
  143.                     }
  144.                 }
  145.             }
  146.         }
  147.     }
  149.     @Override
  150.     public void execute(Runnable runnable) {
  151.         if (this.isShutdown) {
  152.             throw new IllegalStateException(THREAD_POOL_DESTROYED);
  153.         }
  154.         //向任务队列插入Runnable
  155.         this.runnableQueue.offer(runnable);
  156.     }
  158.     /**
  159.      * 同步机制保护：防止与线程池本身的维护线程引起数据冲突
  160.      * 停止BasicThreadPool线程，停止线程池中的活动线程，将isShutdown开关改为true
  161.      */
  162.     @Override
  163.     public void shutdown() {
  164.         synchronized (this) {
  165.             if (isShutdown) {
  166.                 return;
  167.             }
  168.             isShutdown = true;
  169.             threadQueue.forEach(threadTask -> {
  170.                 threadTask.internalTask.stop();
  171.                 threadTask.thread.interrupt();
  172.             });
  173.             this.interrupt();
  174.         }
  175.     }
  177.     @Override
  178.     public int getInitSize() {
  179.         if (isShutdown) {
  180.             throw new IllegalStateException("线程池已销毁");
  181.         }
  182.         return this.initSize;
  183.     }
  185.     @Override
  186.     public int getMaxSize() {
  187.         if (isShutdown) {
  188.             throw new IllegalStateException("线程池已销毁");
  189.         }
  190.         return this.maxSize;
  191.     }
  193.     @Override
  194.     public int getCoreSize() {
  195.         if (isShutdown) {
  196.             throw new IllegalStateException("线程池已销毁");
  197.         }
  198.         return this.coreSize;
  199.     }
  201.     @Override
  202.     public int getQueueSize() {
  203.         if (isShutdown) {
  204.             throw new IllegalStateException("线程池已销毁");
  205.         }
  206.         return runnableQueue.size();
  207.     }
  209.     @Override
  210.     public int getActiveCount() {
  211.         synchronized (this) {
  212.             return this.activeCount;
  213.         }
  214.     }
  216.     @Override
  217.     public boolean isShutdown() {
  218.         return this.isShutdown;
  219.     }
  221.     /**
  222.      * {@link Thread} 和 {@link InternalTask}的一个组合
  223.      */
  224.     private static class ThreadTask {
  225.         Thread thread;
  226.         InternalTask internalTask;
  228.         public ThreadTask(Thread thread, InternalTask internalTask) {
  229.             this.thread = thread;
  230.             this.internalTask = internalTask;
  231.         }
  232.     }
  234.     /**
  235.      * 默认线程工厂
  236.      */
  237.     private static class DefaultThreadFactory implements ThreadFactory {
  238.         private static final AtomicInteger GROUP_COUNTER = new AtomicInteger(1);
  240.         private static final ThreadGroup group = new ThreadGroup("MyThreadPool-" + GROUP_COUNTER.getAndIncrement());
  242.         private static final AtomicInteger COUNTER = new AtomicInteger(0);
  244.         @Override
  245.         public Thread createThread(Runnable runnable) {
  246.             return new Thread(group, runnable, "thread-pool-" + COUNTER.getAndIncrement());
  247.         }
  248.     }
  249. }

ThreadPoolTest

  1. package com.zoro.concurrent.chapter08;
  3. import com.zoro.concurrent.chapter08.threadpool.BasicThreadPool;
  5. import java.util.concurrent.TimeUnit;
  7. /**
  8.  * 线程池测试类
  9.  *
  10.  * @author yx.jiang
  11.  * @date 2021/7/27 10:26
  12.  */
  13. public class ThreadPoolTest {
  14.     public static void main(String[] args) throws InterruptedException {
  15.         BasicThreadPool basicThreadPool = new BasicThreadPool(2, 6, 4, 1000);
  16.         for (int i = 0; i < 20; i++) {
  17.             basicThreadPool.execute(() -> {
  18.                 try {
  19.                     TimeUnit.SECONDS.sleep(10);
  20.                     System.out.println(Thread.currentThread().getName() + " 已启动");
  21.                 } catch (InterruptedException e) {
  22.                     e.printStackTrace();
  23.                 }
  24.             });
  25.         }
  27.         //验证线程池中的动态扩展情况及任务执行情况；最终activeCount = coreSize
  28.         /*for (; ;) {
  29.             System.out.println("basicThreadPool.getActiveCount() = " + basicThreadPool.getActiveCount());
  30.             System.out.println("basicThreadPool.getQueueSize() = " + basicThreadPool.getQueueSize());
  31.             System.out.println("basicThreadPool.getCoreSize() = " + basicThreadPool.getCoreSize());
  32.             System.out.println("basicThreadPool.getMaxSize() = " + basicThreadPool.getMaxSize());
  33.             System.out.println("==========");
  34.             TimeUnit.SECONDS.sleep(5);
  35.         }*/
  37.         TimeUnit.SECONDS.sleep(15);
  38.         //12s后shutdown
  39.         basicThreadPool.shutdown();
  40.         TimeUnit.SECONDS.sleep(1);
  41.         System.out.println("isShutdown:" + basicThreadPool.isShutdown());
  42.         //java.lang.IllegalStateException: 线程池已销毁
  43. //        System.out.println("QueueSize:" + basicThreadPool.getQueueSize());
  44.         //阻塞线程 - jStack -定位原因
  45.         Thread.currentThread().join();
  46.     }
  47. }

Test-Out

  1. //线程池动态扩张情况及任务执行情况： 输出
  3. basicThreadPool.getActiveCount() = 2
  4. basicThreadPool.getQueueSize() = 18
  5. basicThreadPool.getCoreSize() = 4
  6. basicThreadPool.getMaxSize() = 6
  7. ==========
  8. basicThreadPool.getActiveCount() = 2
  9. basicThreadPool.getQueueSize() = 18
  10. basicThreadPool.getCoreSize() = 4
  11. basicThreadPool.getMaxSize() = 6
  12. ==========
  13. basicThreadPool.getActiveCount() = 4
  14. thread-pool-0已启动
  15. thread-pool-1已启动
  16. basicThreadPool.getQueueSize() = 16
  17. basicThreadPool.getCoreSize() = 4
  18. basicThreadPool.getMaxSize() = 6
  19. ==========
  20. basicThreadPool.getActiveCount() = 4
  21. basicThreadPool.getQueueSize() = 14
  22. basicThreadPool.getCoreSize() = 4
  23. basicThreadPool.getMaxSize() = 6
  24. ==========
  25. thread-pool-2已启动
  26. thread-pool-3已启动
  27. thread-pool-1已启动
  28. thread-pool-0已启动
  29. basicThreadPool.getActiveCount() = 6
  30. basicThreadPool.getQueueSize() = 8
  31. basicThreadPool.getCoreSize() = 4
  32. basicThreadPool.getMaxSize() = 6
  33. ==========
  34. basicThreadPool.getActiveCount() = 6
  35. basicThreadPool.getQueueSize() = 8
  36. basicThreadPool.getCoreSize() = 4
  37. basicThreadPool.getMaxSize() = 6
  38. ==========
  39. thread-pool-2已启动
  40. thread-pool-5已启动
  41. thread-pool-3已启动
  42. thread-pool-4已启动
  43. thread-pool-0已启动
  44. thread-pool-1已启动
  45. basicThreadPool.getActiveCount() = 6
  46. basicThreadPool.getQueueSize() = 2
  47. basicThreadPool.getCoreSize() = 4
  48. basicThreadPool.getMaxSize() = 6
  49. ==========
  50. basicThreadPool.getActiveCount() = 6
  51. basicThreadPool.getQueueSize() = 2
  52. basicThreadPool.getCoreSize() = 4
  53. basicThreadPool.getMaxSize() = 6
  54. ==========
  55. thread-pool-4已启动
  56. thread-pool-3已启动
  57. thread-pool-2已启动
  58. thread-pool-5已启动
  59. thread-pool-1已启动
  60. thread-pool-0已启动
  61. basicThreadPool.getActiveCount() = 6
  62. basicThreadPool.getQueueSize() = 0
  63. basicThreadPool.getCoreSize() = 4
  64. basicThreadPool.getMaxSize() = 6
  65. ==========
  66. basicThreadPool.getActiveCount() = 6
  67. basicThreadPool.getQueueSize() = 0
  68. basicThreadPool.getCoreSize() = 4
  69. basicThreadPool.getMaxSize() = 6
  70. ==========
  71. thread-pool-4已启动
  72. thread-pool-3已启动
  73. basicThreadPool.getActiveCount() = 5
  74. basicThreadPool.getQueueSize() = 0
  75. basicThreadPool.getCoreSize() = 4
  76. basicThreadPool.getMaxSize() = 6
  77. ==========
  78. basicThreadPool.getActiveCount() = 5
  79. basicThreadPool.getQueueSize() = 0
  80. basicThreadPool.getCoreSize() = 4
  81. basicThreadPool.getMaxSize() = 6
  82. ==========
  83. basicThreadPool.getActiveCount() = 4
  84. basicThreadPool.getQueueSize() = 0
  85. basicThreadPool.getCoreSize() = 4
  86. basicThreadPool.getMaxSize() = 6
  87. ==========
  88. basicThreadPool.getActiveCount() = 4
  89. basicThreadPool.getQueueSize() = 0
  90. basicThreadPool.getCoreSize() = 4
  91. basicThreadPool.getMaxSize() = 6
  92. ==========
  95. //线程池销毁 - 输出
  96. thread-pool-0 已启动
  97. thread-pool-1 已启动
  98. java.lang.InterruptedException: sleep interrupted
  99.     at java.lang.Thread.sleep(Native Method)
  100.     at java.lang.Thread.sleep(Thread.java:340)
  101.     at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
  102.     at com.zoro.concurrent.chapter08.ThreadPoolTest.lambda$main$0(ThreadPoolTest.java:19)
  103.     at com.zoro.concurrent.chapter08.threadpool.InternalTask.run(InternalTask.java:25)
  104.     at java.lang.Thread.run(Thread.java:748)
  105. java.lang.InterruptedException: sleep interrupted
  106.     at java.lang.Thread.sleep(Native Method)
  107.     at java.lang.Thread.sleep(Thread.java:340)
  108.     at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
  109.     at com.zoro.concurrent.chapter08.ThreadPoolTest.lambda$main$0(ThreadPoolTest.java:19)
  110.     at com.zoro.concurrent.chapter08.threadpool.InternalTask.run(InternalTask.java:25)
  111.     at java.lang.Thread.run(Thread.java:748)
  112. java.lang.InterruptedException: sleep interrupted
  113.     at java.lang.Thread.sleep(Native Method)
  114.     at java.lang.Thread.sleep(Thread.java:340)
  115.     at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
  116.     at com.zoro.concurrent.chapter08.ThreadPoolTest.lambda$main$0(ThreadPoolTest.java:19)
  117.     at com.zoro.concurrent.chapter08.threadpool.InternalTask.run(InternalTask.java:25)
  118.     at java.lang.Thread.run(Thread.java:748)
  119. java.lang.InterruptedException: sleep interrupted
  120.     at java.lang.Thread.sleep(Native Method)
  121.     at java.lang.Thread.sleep(Thread.java:340)
  122.     at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
  123.     at com.zoro.concurrent.chapter08.ThreadPoolTest.lambda$main$0(ThreadPoolTest.java:19)
  124.     at com.zoro.concurrent.chapter08.threadpool.InternalTask.run(InternalTask.java:25)
  125.     at java.lang.Thread.run(Thread.java:748)
  126. isShutdown:true
  127. Exception in thread "main" java.lang.IllegalStateException: 线程池已销毁
  128.     at com.zoro.concurrent.chapter08.threadpool.BasicThreadPool.getQueueSize(BasicThreadPool.java:204)
  129.     at com.zoro.concurrent.chapter08.ThreadPoolTest.main(ThreadPoolTest.java:42)

最后

存在问题 【待优化项】

  • BasicThreadPool和Thread不应为继承关系，组合会更恰当；这样可以避免使用者直接调用Thread中的方法。
  • 线程池的销毁功能并未返回未被处理的任务，这样会导致未被处理的任务丢弃。
  • BasicThreadPool的构造函数参数过多，建议使用builder和设计模式对其封装或提供工厂方法进行构造。
  • 线程池中的数量控制没有进行合法性校验，如：initSize不应该大于maxSize。
  • 其他……