自定义线程池简单介绍

       线程池的相关概念就不在这里说明了，百度一下有很多，这里简单表述一下如何实现一个自定义的线程池就行线程管理，我们如果要实现一个线程池对线程的管理，那么需要实现一下几点的思路:

1.如何管理线程

2.如何定义工作线程以及工作线程如何持续的保持运行状态

3.如何定义线程池大小及队列大小

4.如何提供接口给调用者使用

5.如何关闭线程池中的线程

接下来我们就一一的实现这几个问题。

1.我们需要定义一个队列来来管理线程，这里使用了LinkedBlockingQueue

// 1.定义一个存储线程队列
private LinkedBlockingQueue<Runnable> queue;

 2.因为是一个简单的测试，所以我们可以先定义一个内部类来实现工作线程

// 2.定义工作线程进行线程的执行
    class Worker extends Thread {

        private SelfThreadPoolExecutor threadPoolExecutor;

        public Worker(SelfThreadPoolExecutor poolExecutor) {
            this.threadPoolExecutor = poolExecutor;
        }

        @Override
        public void run() {
            Runnable task;
            while (threadPoolExecutor.receiveTask || threadPoolExecutor.queue.size() > 0) {
                try {
                    // 有线程则取出来，否则等待
                    System.out.println("准备消费线程");
                    task = threadPoolExecutor.queue.take();
                    if (task != null) {
                        task.run();
                        System.out.println("消费线程");
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

 SelfThreadPoolExecutor是外部定义的整体类名

3.使用有参的构造方法进行线程池大小的管理

// 3.存放工作线程的集合
    private List<Worker> workerList;

    // 4.线程池初始化
    public SelfThreadPoolExecutor(int coreSize, int queueSize) {
        if (coreSize <= 0 || queueSize <= 0) {
            throw new IllegalArgumentException("参数不正确");
        }
        this.queue = new LinkedBlockingQueue<>(queueSize);
        // 线程安全的集合
        this.workerList = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < coreSize; i++) {
            Worker worker = new Worker(this);
            worker.start();
            workerList.add(worker);
        }
    }

4.定义阻塞和非阻塞的方式提供对应的接口

// 5.非阻塞的方法接口
    public boolean offer(Runnable task) {
        if (receiveTask) {
            return queue.offer(task);
        } else {
            return false;
        }
    }

    // 6.阻塞的方法接口
    public void put(Runnable task) {
        try {
            if (receiveTask) {
                queue.put(task);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

6.进行线程池的关闭

// 7.线程池的关闭
    private boolean receiveTask = true;

    public void shutdown() {
        // 7.1.队列不再接收线程
        receiveTask = false;
        // 7.2.关闭处于wait或block的线程
        for (Thread thread : workerList) {
            if (Thread.State.BLOCKED.equals(thread.getState())
            || Thread.State.WAITING.equals(thread.getState())
            || Thread.State.TIMED_WAITING.equals(thread.getState())){
                thread.interrupt();
            }
        }
    }

 我们测试的方法如下：

public static void main(String [] args){
        SelfThreadPoolExecutor selfThreadPoolExecutor = new SelfThreadPoolExecutor(5,10);
        for(int i = 0;i < 20;i++){
            Runnable task = () ->{
                System.out.println("开启线程");
            };
            selfThreadPoolExecutor.put(task);
        }
        selfThreadPoolExecutor.shutdown();
    }

 运行结果是：

准备消费线程
准备消费线程
准备消费线程
准备消费线程
准备消费线程
开启线程
消费线程
准备消费线程
开启线程
消费线程
准备消费线程
开启线程
消费线程
准备消费线程
。。。。。。

 整体代码见附件。