如何自定义线程池工具类（ThreadPoolUtils）

Posted 2021-12-06 Tang.Mr

1.线程池对象

java.util.concurrent.ThreadPoolExecutor

2.线程池核心参数了解

在自定义线程池之前先来了解一下在创建并初始化线程池时有那些所需知的前置条件(线程池有那些核心参数)，详细入下代码：

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.acc = System.getSecurityManager() == null ?
    null :
  AccessController.getContext();
  this.corePoolSize = corePoolSize;
  this.maximumPoolSize = maximumPoolSize;
  this.workQueue = workQueue;
  this.keepAliveTime = unit.toNanos(keepAliveTime);
  this.threadFactory = threadFactory;
  this.handler = handler;
}

3.自定义线程池思路

创建和初始化线程池可以直接通过new ThreadPoolExecutor(...);设定线程池的核心参数来进行构建线程池对象；通过线程池对象调用ThreadPoolExecutor类中的execute方法去执行线程任务。但是如果以此种方式来做的话每次需要由线程池去执行请求任务的时候都需要先去创建和初始化线程池。这样以来在n个不同的地方使用，那么就会产生n个线程池。

实现思路：

• 封装线程创建的工厂，在线程池初始化时使用自己封装的线程构建工厂模式
• 封装饱和策略，在线程池初始化时使用自己封装的饱和策略
• 封装一个线程池工具类：该工具类设计模式为单例模式，主要是用来创建初始化线程池、对外提供执行线程任务、查看线程池状态、关闭运行中的线程、关闭线程池状态等等公共函数

4.代码

4.1.封装创建线程池的工厂对象

import java.util.concurrent.ThreadFactory;

/**
 * @Author ScholarTang
 * @Date 2021/11/3 下午12:38
 * @Desc 创建线程池的工厂对象
 */
public class MyThreadFactory implements ThreadFactory {

    /**
     * 该方法用来创建线程
     * @param r
     * @return
     */
    @Override
    public Thread newThread(Runnable r) {
        Thread newThread = new Thread(r);
        return newThread;
    }
}

4.2.封装线程池饱和策略

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author ScholarTang
 * @Date 2021/11/3 下午12:42
 * @Desc 线程池饱和策略
 */

public class ThreadRejectedExecutionHandler implements RejectedExecutionHandler{

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {}

    /**
     * 饱和策略一：调用者线程执行策略
     * 在该策略下，在调用者中执行被拒绝任务的run方法。除非线程池showdown，否则直接丢弃线程
     */
    public static class CallerRunsPolicy extends ThreadRejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            //判断线程池是否在正常运行，如果线程池在正常运行则由调用者线程执行被拒绝的任务。如果线程池停止运行，则直接丢弃该任务
            if (!executor.isShutdown()){
                r.run();
            }
        }
    }


    /**
     * 饱和策略二：终止策略
     * 在该策略下，丢弃被拒绝的任务，并抛出拒绝执行异常
     */
    public static class AbortPolicy extends ThreadRejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            throw new RejectedExecutionException("请求任务：" + r.toString() + "，线程池负载过高执行饱和终止策略！");
        }
    }


    /**
     * 饱和策略三：丢弃策略
     * 在该策略下，什么都不做直接丢弃被拒绝的任务
     */
    public static class DiscardPolicy extends ThreadRejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {}
    }

    /**
     * 饱和策略四：弃老策略
     * 在该策略下，丢弃最早放入阻塞队列中的线程，并尝试将拒绝任务加入阻塞队列
     */
    public static class DiscardOldestPolicy extends ThreadRejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            //判断线程池是否正常运行，如果线程池正常运行则弹出（或丢弃）最早放入阻塞队列中的任务，并尝试将拒绝任务加入阻塞队列。如果线程池停止运行，则直接丢弃该任务
            if (!executor.isShutdown()){
                executor.getQueue().poll();
                executor.execute(r);
            }
        }
    }
}

4.3.封装线程池工具类

import java.util.List;
import java.util.concurrent.*;

/**
 * @Author ScholarTang
 * @Date 2021/11/3 下午1:03
 * @Desc 单例的线程池工具类
 */

public class ThreadPoolUtils {

    /**
     * 系统可用计算资源
     */
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 核心线程数
     */
    private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));

    /**
     * 最大线程数
     */
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;

    /**
     * 空闲线程存活时间
     */
    private static final int KEEP_ALIVE_SECONDS = 30;

    /**
     * 工作队列
     */
    private static final BlockingQueue<Runnable> POOL_WORK_QUEUE = new LinkedBlockingQueue<>(128);

    /**
     * 工厂模式
     */
    private static final MyThreadFactory MY_THREAD_FACTORY = new MyThreadFactory();

    /**
     * 饱和策略
     */
    private static final ThreadRejectedExecutionHandler THREAD_REJECTED_EXECUTION_HANDLER = new ThreadRejectedExecutionHandler.CallerRunsPolicy();

    /**
     * 线程池对象
     */
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR;

    /**
     * 声明式定义线程池工具类对象静态变量，在所有线程中同步
     */
    private static volatile ThreadPoolUtils threadPoolUtils = null;


    /**
     * 初始化线程池静态代码块
     */
    static {
        THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(
                //核心线程数
                CORE_POOL_SIZE,
                //最大线程数
                MAXIMUM_POOL_SIZE,
                //空闲线程执行时间
                KEEP_ALIVE_SECONDS,
                //空闲线程执行时间单位
                TimeUnit.SECONDS,
                //工作队列（或阻塞队列）
                POOL_WORK_QUEUE,
                //工厂模式
                MY_THREAD_FACTORY,
                //饱和策略
                THREAD_REJECTED_EXECUTION_HANDLER
                );
    }


    /**
     * 线程池工具类空参构造方法
     */
    private ThreadPoolUtils() {}

    /**
     * 获取线程池工具类实例
     * @return
     */
    public static ThreadPoolUtils getNewInstance(){
        if (threadPoolUtils == null) {
            synchronized (ThreadPoolUtils.class) {
                if (threadPoolUtils == null) {
                    threadPoolUtils = new ThreadPoolUtils();
                }
            }
        }
        return threadPoolUtils;
    }


    /**
     * 执行线程任务
     * @param runnable 任务线程
     */
    public void executor(Runnable runnable) {
        THREAD_POOL_EXECUTOR.execute(runnable);
    }

    /**
     * 获取线程池状态
     * @return 返回线程池状态
     */
    public boolean isShutDown(){
        return THREAD_POOL_EXECUTOR.isShutdown();
    }

    /**
     * 停止正在执行的线程任务
     * @return 返回等待执行的任务列表
     */
    public List<Runnable> shutDownNow(){
        return THREAD_POOL_EXECUTOR.shutdownNow();
    }

    /**
     * 关闭线程池
     */
    public void showDown(){
        THREAD_POOL_EXECUTOR.shutdown();
    }


    /**
     * 关闭线程池后判断所有任务是否都已完成
     * @return
     */
    public boolean isTerminated(){
        return THREAD_POOL_EXECUTOR.isTerminated();
    }
}

4.4.任务线程类

/**
 * @Author ScholarTang
 * @Date 2021/11/3 下午2:14
 * @Desc 任务线程
 */

public class ThreadTask implements Runnable{
    @Override
    public void run() {
        System.out.println("thread task：" + this.toString() + "，task start！");
    }
}

5.测试

/**
 * @Author ScholarTang
 * @Date 2021/11/3 下午2:16
 * @Desc 测试类
 */

public class Test {
    public static void main(String[] args) {
        ThreadPoolUtils newInstance = ThreadPoolUtils.getNewInstance();
        newInstance.executor(new ThreadTask());
        newInstance.showDown();
    }
}