【详解】Java并发之自定义线程池

版本一

需求：

实现一个简单的任务队列
实现初始化线程数目

任务队列实现

设计一个队列，存放着Runnable实例，这些实例就是任务
任务队列的读写操作在并发的情况下存在线程不安全的问题，如超出最毒的队列数目，或者多个线程读取到一个任务，所以必须加锁
每个线程都会尝试从任务队列中读取任务，如果抢到了任务队列中的任务，则执行该任务；如果没有抢到，则继续争抢，直到全部的任务执行完毕，陷入阻塞
给外界一个提交任务的接口，可以将Runnable任务写入到队列中（注意加锁），同时唤醒全部阻塞的线程

初始化线程数目实现

设计一个默认线程数，如果没有输入时，采用默认线程数
在构造方法调用初始化方法，将根据数目创建线程，并放入到线程集合中，方便后续的管理

拒绝策略实现

设置一个属性保存最大的任务队列数，如果没有赋值，默认为2000
设计一个接口用于让用户自定义拒绝策略，如果没有设置自定义的，则默认使用抛出异常
在提交任务的时候，判断是否超出了最大任务数。如果超出，则执行拒绝策略

实现销毁线程池

判断任务队列是否为空，如果不为空，则休眠一会继续判断
设置一个变量用于保存还没有消除的线程数量
判断线程状态，如果处于阻塞状态，则打断令其销毁
为了防止线程刚刚修改状态还没有进入等待状态，即会出现打断失败，打断的同时需要将其状态设置为死亡状态

线程池的的扩容与空闲回收

线程池的扩容以及回收，都需要两个重要的参数：当前的线程数、任务队列的大小
线程的扩容和回收应该由线程池自动处理，线程池就是一个线程，当没有销毁时，每隔一段时间判断一次
需要特别注意锁的控制：如扩容时，可能会与销毁出现冲突，所以需要加锁；回收时与防止线程突然唤醒接到任务正在执行


public class SimpleThreadPool extends Thread{

    private int size;//初始化线程数目

    private final int queueSize;//最大任务队列大小

    private final  DiscardPolicy policy;//拒绝策略

    private volatile boolean  destroy = false;//是否是销毁状态

    private int min;

    private int max;

    private int active;

    private final Object Lock = new Object();

    private static volatile int seq;//自增数字，模仿线程的默认命名方式

    private static final String THREAD_PREFIX = "THREAD_POOL";//线程名词前缀

    private final static ThreadGroup GROUP = new ThreadGroup("POOL_THREAD_GROUP");

    private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();//任务队列，装着要执行的任务

    private final static List<WorkerTask> THREAD_LIST = new ArrayList<>();//线程组

    private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;//线程队列默认最大值



    private final static  DiscardPolicy DEFAULT_DISCARD_POLICY = ()->{//默认拒绝策略
        throw new DiscardException("拒绝该任务！！！");
    };


    public SimpleThreadPool() {
        this( 4,8,12,DEFAULT_TASK_QUEUE_SIZE,DEFAULT_DISCARD_POLICY);
    }

    public SimpleThreadPool(int queuesize){
        this(4,8,12,queuesize,DEFAULT_DISCARD_POLICY);
    }

    public SimpleThreadPool(int min,int active,int max,int queueSize,DiscardPolicy policy) {//初始化线程数大小
        this.queueSize = queueSize;
        this.policy = policy;
        this.min = min;
        this.active = active;
        this.max = max;
        this.init();
    }

    /** * 初始化线程池 */

    private void init() {
        for (int i = 0; i < min; i++) {
            this.CreatWorkerTask();
        }
        this.size = min;
        this.start();
    }


    /** * 创建线程 */
    private void CreatWorkerTask() {
        WorkerTask workerTask = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
        workerTask.start();
        THREAD_LIST.add(workerTask);
    }

    /** * 进行扩容，以及输出当前线程池的状态 */
    @Override
    public  void run() {
        while (!destroy){
            System.out.printf("Pool#Min:%d,Active:%d,Max:%d,Current:%d,QueueSize:%d\n",
                    this.min, this.active, this.max, this.size, TASK_QUEUE.size());
            try {
                Thread.sleep(5_000);
                synchronized (Lock){
                    
                    if (TASK_QUEUE.size()>active  && size <active && !destroy){//线程扩容到active
                        for (int i = size; i < active; i++) {
                            this.CreatWorkerTask();
                        }
                        System.out.println("线程池扩容到----active");
                        size = active;
                    }else if (TASK_QUEUE.size()>max  && size <max && !destroy){//线程扩容到max
                        for (int i = size; i < max; i++) {
                            this.CreatWorkerTask();
                        }
                        System.out.println("线程池扩容到----max");
                        size = max;
                    }

                    if (TASK_QUEUE.isEmpty() && size > active && !destroy){//线程降级到active
                        synchronized (TASK_QUEUE){
                            int release = size - active;
                            for (Iterator<WorkerTask> it = THREAD_LIST.iterator();it.hasNext();) {
                                if (release<=0){
                                    break;
                                }
                                WorkerTask wt = it.next();
                                wt.close();
                                wt.interrupt();
                                it.remove();
                                release--;
                            }
                            size = active;
                        }
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    /** * 提交任务功能，同时唤醒正在阻塞的队列 * * @param runnable 提交的任务 */
    public void submit(Runnable runnable) {
        if (destroy){
            throw new IllegalStateException("线程池已经被销毁，不能再提交任务");
        }
        synchronized (TASK_QUEUE) {
            if (TASK_QUEUE.size()>queueSize){//当超出规定的最大任务数
                policy.discard();//执行拒绝策略
            }else  {
                TASK_QUEUE.addLast(runnable);
                TASK_QUEUE.notifyAll();//唤醒全部阻塞线程
            }
        }
    }

    /** * 销毁线程池 * @throws InterruptedException */

    public void shutdown() throws InterruptedException {
        synchronized (Lock){
            while (!TASK_QUEUE.isEmpty()){
                Thread.sleep(50);
            }

            int initVal  = THREAD_LIST.size();
            while (initVal>0){
                for (WorkerTask task : THREAD_LIST) {
                    if (task.getTaskState()==TaskState.BLOCKED){
                        task.interrupt();
                        task.close();//防止线程还没有wait的时候被打断，无法跳出循环
                        initVal--;
                    }else {
                        Thread.sleep(10);
                    }
                }
            }

            this.destroy =true;
            System.out.println("线程池已经关闭");
        }

    }

    public int getSize() {
        return size;
    }

    public int getQueuesize() {
        return queueSize;
    }

    public boolean isDestroy(){
        return this.destroy;
    }

    private enum TaskState {//线程生命周期的一些状态
        FARE, RUNNING, BLOCKED, DEDE
    }

    /** * 拒绝策略抛出的异常 */
    public static class DiscardException extends RuntimeException {

        public DiscardException(String message) {
            super(message);
        }
    }

    /** * 拒绝策略 */
    public interface DiscardPolicy {

        void discard() throws DiscardException;
    }




    /** * 自定义的工作线程 * 用于执行工作队列中的任务 */
    private static class WorkerTask extends Thread {

        private volatile TaskState taskState = TaskState.FARE;//默认是空闲状态

        public WorkerTask(ThreadGroup group, String name) {
            super(group, name);
        }


        public TaskState getTaskState() {
            return this.taskState;
        }

        @Override
        public void run() {
            OUTER:
            while (this.taskState != TaskState.DEDE) {//只要线程没有死亡
                Runnable runnable;
                synchronized (TASK_QUEUE) {
                    while (TASK_QUEUE.isEmpty()) {
                        try {
                            taskState = TaskState.BLOCKED;//此时没有任务，线程陷入阻塞
                            TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            taskState = TaskState.DEDE;
                            break OUTER;//直接跳出多级循环，防止唤醒后，线程已经死亡
                        }
                    }
                    runnable = TASK_QUEUE.removeFirst();

                }
                if (runnable != null) {//如果可以运行
                    taskState = TaskState.RUNNING;
                    runnable.run();
                    taskState = TaskState.FARE;
                }
            }
        }

        public void close() {//关闭线程
            this.taskState = TaskState.DEDE;
        }
    }


    /** * 测试 */
    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool pool = new SimpleThreadPool();

        Thread.sleep(5_000);
        new Thread(()->{
            for (int i = 0; i < 200; i++) {
                int finalI = i;
                pool.submit(() -> {
                    System.out.println("当前是" + Thread.currentThread().getName() + "线程为我服务......" );
                    try {
                        Thread.sleep(1_000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("当前是" + Thread.currentThread().getName() + "线程完成了服务！！！" + finalI);
                });
            }
        }).start();

        Thread.sleep(10_000);
        pool.shutdown();


    }

}