手写Java线程池(超详细解说)

线程池

问题背景

只是单纯使用
new Thread(runnable).start();的方式创建线程, 将会导致严重的程序性能问题: 1.线程创建, 销毁需要消耗很大的系统资源; 2.虚拟机创建线程的数量是有限的; 2.线程调度切换也将使程序性能下降; 针对这些问题, 对线程数量进行管理, 有效地重复利用线程, 将会很好地提高程序性能.

线程池原理

使用队列创建一定数量的线程, 当有任务的时候, 使用队列中线程执行任务(如果任务过多, 就将其放入任务队列, 进入等待执行状态), 任务执行完就自动回收线程队列中的线程(任务过少或者任务数量小于线程数量, 超出的线程将会销毁, 做到线程队列具有伸缩性);
根据上面描述, 我们自己的线程池将具有一下特点:
1.内部使用队列来管理线程, 管理提交的任务.
2.控制线程数量, 做到线程队列具有良好的伸缩性.
3.当任务数过多, 或者任务队列已经饱和, 将使用任务拒绝策略, 告诉对应的任务提交者.
4.使用线程工厂定制线程队列中, 每个线程的名字, 状态, 是否为守护线程等等.

线程池类图结构

1.任务队列, 队列使用limit限制提交任务的大小, 实现RunnableQueue接口(RunnableQueue接口负责: 1.接收用户提交的任务; 2.获取任务队列中的任务; 3.查看任务队列大小), LinkedRunnableQueue实现RunnableQueue中的方法, 并且针对用户提交不同的任务以及线程池种类(ThreadPool)的不同, 决定是否执行拒绝策略(拒绝策略具有多个, 拒绝方式取决于用户自定义, 在线程池内部具有默认的拒绝策略实现);
在这里插入图片描述

任务队列描述
2.实现Runnable接口, 在run方法中获取RunnableQueue中的任务, 然后执行RunnQueue中的任务, InternalTask中的run方法是一个while循环循环结束条件取决于是否关闭该线程(关闭线程据需要设置flag变量, 当flage为false, 线程run方法结束, 自动结束生命), 而不是当前用户提交的任务是否执行完!!!; InternalTask主要是对RunnableQueue的一种封装; stop方法主要是设置线程flag(flag主要判断当前线程是否关闭)
在这里插入图片描述

实现Runnable接口描述

线程池原型:

1.实现Runnable接口(此处注明, BasicThreadPool是继承Thread, 但是Thread内容太多了不能很好地在UML图中显示, 所以我就把他删除了, 只留下了实现Runnable接口), 因为线程池自身执行也需要一个线程, 所以继承Thread, 这样可以在BasicThreadPool的构造方法中执行start(), run方法中执行创建线程的操作(线程池内部执行任务的线程); 创建线程取决于线程池设置的最大线程数, 核心线程数, 初始化线程数, 用户提交的任务数;
2.实现ThreadPool接口(该接口主要用于定义线程池的基本操作, 比如执行任务, 获取线程池的一些基本属性) ;
3.内部具有2个内部类(ThreadTask负责对InternalTask进行封装, DefaultThreadFactory主要定义默认的线程创建方式), 不同的线程池中拥有不同的默认创建方式, 因此将线程创建方式设置为内部类;
4.在BasicThreadPool中使用newThread方法创建线程(这些线程用于执行ThreadTask中的任务);
5.线程池原型中具有2个队列, 第一个是刚才上面提的RunnQueue(负责执行的任务), 第二个是ThreadQueue(负责存储创建的每一个线程, 使用ArrayQueue实现, 这样很好地维护管理了线程, 做到资源重用)
6.removeThread方法: 删除多余的线程, 当用户提交的任务数量小于线程池中创建的线程数量, 那么就删除一定数量的线程, 这样才不会浪费线程资源.
7.在构造方法中设置基本属性, 以及当前线程池的拒绝策略.
在这里插入图片描述

结构图

每个接口, 类的详细定义

ThreadPool(interface 定义线程池基本操作)

package com.concurrent.customthreadpool;

/**
 * 线程池接口
 * @author regotto
 */
public interface ThreadPool {

    /**
     * 执行提交的Runnable任务
     * @param runnable
     */
    void execute(Runnable runnable);

    /**
     * 关闭线程池
     */
    void shutdown();

    /**
     * 获得线程池初始化大小
     * @return initSize
     */
    int getInitSize();

    /**
     * 获得线程池最大线程数
     * @return maxSize
     */
    int getMaxSize();

    /**
     * 获取线程池核心线程数
     * @return coreSize
     */
    int getCoreSize();

    /**
     * 获取线程池中用于缓存任务队列的大小
     * @return queueSize
     */
    int getQueueSize();

    /**
     * 获取线程池中国活跃的线程数量
     * @return activeCount
     */
    int getActiveCount();

    /**
     * 查看线程池是否shutdown
     * @return boolan
     */
    boolean isShutdown();

}

RunnableQueue(interface 任务队列)

package com.concurrent.customthreadpool;

/**
 * 存放提交的Runnable, 使用BlockedQueue, 设置limit
 * @author regotto
 */
public interface RunnableQueue {

    /**
     * 缓存提交到线程池中的任务
     * @param runnable
     */
    void offer(Runnable runnable);

    /**
     * 从缓存中获取Runnable任务
     * 如果没有任务, 调用者线程挂起, 在某些特定的时候抛出中断异常
     * @throws InterruptedException
     * @return runnable
     */
    Runnable take() throws InterruptedException;

    /**
     * 缓冲区大小
     * @return size
     */
    int size();

}

LinkedRunnableQueue(class 对RunnableQueue的封装, 用户提交任务, 线程执行任务, 此过程使用生产者-消费者模式实现)

package com.concurrent.customthreadpool;

import java.util.LinkedList;

/**
 * 线程池的内部线程队列, 缓冲区
 * @author regotto
 */
public class LinkedRunnableQueue implements RunnableQueue{

    /**
     * limit: 限制当前runnableList中还能存放多少内容
     * denyPolicy: 拒绝策略
     * runnableList: 存放runnable的缓冲区
     * threadPool: 线程池
     */
    private final int limit;
    private  final RunnableDenyPolicy denyPolicy;
    private final LinkedList<Runnable> runnableList = new LinkedList<>();
    private final ThreadPool threadPool;

    public LinkedRunnableQueue(int limit, RunnableDenyPolicy denyPolicy, ThreadPool threadPool){
        this.limit = limit;
        this.denyPolicy = denyPolicy;
        this.threadPool = threadPool;
    }

    @Override
    public void offer(Runnable runnable) {
        synchronized (runnableList) {
            if (runnableList.size() >= limit) {
                   //用户提交的任务大于限制条件, 执行对应的拒绝策略
                System.out.println(runnableList.size() + " >= " + limit + " execute deny policy");
                denyPolicy.reject(runnable, threadPool);
            } else {
                //添加任务到任务队列尾部, 有任务存在, 唤醒刚才wait的线程
                runnableList.addLast(runnable);
                runnableList.notifyAll();
            }
        }
    }

    @Override
    public Runnable take() throws InterruptedException{
        synchronized (runnableList) {
            while (runnableList.isEmpty()) {
                try {
                    //从RunnableQueue中取出任务, 如果任务为空, 使当前线程wait
                    runnableList.wait();
                } catch (InterruptedException e) {
                    throw e;
                }
            }
            //移除任务缓冲区的第一个
            return runnableList.removeFirst();
        }
    }

    @Override
    public int size() {
        synchronized (runnableList) {
            return runnableList.size();
        }
    }
}

InternalTask(class 对RunnableQueue中任务的执行)

package com.concurrent.customthreadpool;

/**
 * 用于线程池内部, 获取runnableQueue中的runnable
 * @author regotto
 */
public class InternalTask implements Runnable {

    private final RunnableQueue runnableQueue;
    private volatile boolean running = true;

    public InternalTask(RunnableQueue runnableQueue){
        this.runnableQueue = runnableQueue;
    }

    @Override
    public void run() {
        //如果线程没有关闭, 就让该线程死循环, 处理每一个提交的任务
        while (running && !Thread.currentThread().isInterrupted()){
            try {
                //处于中断时候的线程不做处理
                //获取RunnableQueue中任务, 然后执行
                Runnable take = runnableQueue.take();
                System.out.println("runnableQueue.take(): " + take.toString());
                take.run();
            } catch (InterruptedException e) {
                running = false;
                break;
            }
        }
    }

    /**
     * 停止当前任务, 设置其running为false, 在shutdown中处理
     */
    public void stop(){
        this.running = false;
    }

}

RunnableDenyPolicy(interface 任务拒绝策略)

package com.concurrent.customthreadpool;

/**
 * 当任务数提交超过缓冲区limit, 执行对应的任务拒绝策略
 * @author regotto
 */
@FunctionalInterface
public interface RunnableDenyPolicy {

    /**
     * 对提交到threadPool的runnable是否执行reject
     * @param runnable
     * @param threadPool
     */
    void reject(Runnable runnable, ThreadPool threadPool);

    /**
     * 该策略使用空方法直接丢弃任务
     */
    class DiscardDenyPolicy implements RunnableDenyPolicy {

        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
            System.out.println(runnable + "不做处理");
        }
    }

    /**
     * 该策略抛出一个RunnableDenyException
     */
    class AbortDenyPolicy implements RunnableDenyPolicy {

        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
            throw new RunnableDenyException("The" + runnable + "will be abort");
        }
    }

    /**
     *该策略Runnable给提交者所在的线程中运行, 不加入到线程中
     */
    class RunnerDenyPolicy implements RunnableDenyPolicy{

        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
            if (threadPool.isShutdown()) {
                runnable.run();
            }
        }
    }

}
RunnableDenyException(class 处理RunnableDenyPolicy抛出的运行时异常)
package com.concurrent.customthreadpool;

public class RunnableDenyException extends RuntimeException {
    public RunnableDenyException(String message) {
        super(message);
    }
}

ThreadFactory(interface 定义创建线程的接口)

package com.concurrent.customthreadpool;

/**
 * 创建线程接口, 定制线程属于哪一个group, 是否为守护线程, 优先级, 名字等
 * @author regotto
 */
public interface ThreadFactory {

    /**
     * 创建定制化线程
     * @param runnable
     * @return thread
     */
    Thread createThread(Runnable runnable);

}

BasicThreadPool(class 线程池的实现)

package com.concurrent.customthreadpool;

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 默认的自定义的线程池, 内部使用Queue进行维护
 * @author regotto
 */
public class BasicThreadPool extends Thread implements ThreadPool{

    /**initSize: 初始化线程数
     * maxSize: 线程池最大线程数
     * coreSize: 线程核心数
     * activeCount: 当前活跃线程数
     * threadFactory: 线程工厂, 配置线程创建需要的参数
     * runnableQueue: 任务队列
     * isShutdown: 是否关闭线程池
     * threadQueue: 工作线程队列
     * DEFAULT_THREAD_FACTORY: 默认的线程工厂
     * keepAliveTime: 线程存活时间
     */
    private final int initSize;
    private final int maxSize;
    private final int coreSize;
    private int activeCount;
    private final ThreadFactory threadFactory;
    private final RunnableQueue runnableQueue;
    private volatile boolean isShutdown = false;
    private Queue<ThreadTask> threadQueue = new ArrayDeque<>();
    private final static RunnableDenyPolicy DEFAULT_DENY_POLICY = new RunnableDenyPolicy.DiscardDenyPolicy();
    private final static ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory();
    private final static long DEFAULT_KEEP_ALIVE_TIME = 10;
    private final long keepAliveTime;
    private final TimeUnit timeUnit;

    public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) {
        this(initSize, maxSize, coreSize, DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENY_POLICY,
                DEFAULT_KEEP_ALIVE_TIME, TimeUnit.SECONDS);
    }

    public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory, int queueSize,
                           RunnableDenyPolicy denyPolicy, long keepAliveTime, TimeUnit timeUnit) {
        this.initSize = initSize;
        this.maxSize = maxSize;
        this.coreSize = coreSize;
        this.threadFactory = threadFactory;
        this.runnableQueue = new LinkedRunnableQueue(queueSize, denyPolicy, this);
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = timeUnit;
        this.init();
    }

    /**
     * 初始化线程池, 创建initThread
     */
    private void init() {
        start();
        for (int i = 0; i < initSize; ++i) {
            newThread();
        }
    }

    /**
     * 创建线程添加到线程队列, 然后用该线程执行ThreadTask任务(层层封装, 封装用户提交的任务)
     */
    private void newThread() {
        InternalTask internalTask = new InternalTask(runnableQueue);
        //使用自定义的线程工厂创建线程
        Thread thread = this.threadFactory.createThread(internalTask);
        ThreadTask threadTask = new ThreadTask(thread, internalTask);
        System.out.println(threadTask.thread.getName() + "被添加");
        //添加到线程队列
        threadQueue.offer(threadTask);
        this.activeCount++;
        //被添加后的线程执行start
        thread.start();
    }

    @Override
    public void execute(Runnable runnable) {
        if (this.isShutdown) {
            throw new IllegalStateException("The thread pool id destroy");
        }
        //将用户提交的任务放到runnableQueue中, 等待线程队列中线程执行
        this.runnableQueue.offer(runnable);
    }

    private void removeThread() {
        //ArrayDeque的remove就是removeFirst
        ThreadTask threadTask = threadQueue.remove();
        //设置当前线程flag, 在InternalTask中跳出循环自动结束线程生命
        threadTask.internalTask.stop();
        this.activeCount--;
    }

    @Override
    public void run() {
        while (!isShutdown && !isInterrupted()) {
            try {
                timeUnit.sleep(keepAliveTime);
            } catch (InterruptedException e) {
                isShutdown = true;
                break;
            }
            synchronized (this) {
                if (isShutdown) {
                    break;
                }
                //当前队列中有任务还没有处理, 且activeCount < coreSize
                if (runnableQueue.size() > 0 && activeCount < coreSize) {
                    //此处i曾写做i=0,导致多创建了一个线程,在没有任务的时候该线程一直保持wait
                    //因为关闭pool,该线程没有add到threadQueue,导致Interrupt失败,最终导致线程一直运行中
                    for (int i = initSize; i < coreSize; ++i) {
                        newThread();
                    }
                    //防止后面的if判断创建线程数超过coreSize, 在coreSize还没有满的时候, 只执行当前的if
                    continue;
                }
                //当上面if中创建的线程数不足的时候, 就扩大线程池线程数, 直到maxSize
                if (runnableQueue.size() > 0 && activeCount < maxSize) {
                    for (int i = coreSize; i < maxSize; ++i) {
                        newThread();
                    }
                }
                //当没有任务, 但是activeCount线程数超出coreSize大小, 回收超出coreSize的线程
                if (runnableQueue.size() == 0 && activeCount > coreSize) {
                    for (int i = coreSize; i < activeCount; ++i) {
                        removeThread();
                    }
                }
            }
        }
    }

    @Override
    public void shutdown() {
        synchronized (this) {
            if (!isShutdown) {
                isShutdown = true;
                System.out.println("threadQueue size:" + threadQueue.size());
                threadQueue.forEach(threadTask -> {
                    //调用internalTask中stop, 设置当前线程运行标志为false
                    threadTask.internalTask.stop();
                    //设置线程中断状态
                    threadTask.thread.interrupt();
                    System.out.println(threadTask.thread.getName());
                });
                System.out.println("threadQueue中线程已经关闭");
                //当前线程池自己也要关闭
                this.interrupt();
            }
        }
    }

    @Override
    public int getInitSize() {
        if (isShutdown) {
            throw new IllegalStateException("The thread pool is destroy");
        }
        return this.initSize;
    }

    @Override
    public int getMaxSize() {
        if (isShutdown) {
            throw new IllegalStateException("The thread pool is destroy");
        }
        return this.maxSize;
    }

    @Override
    public int getCoreSize() {
        if (isShutdown) {
            throw new IllegalStateException("The thread pool is destroy");
        }
        return this.coreSize;
    }

    @Override
    public int getQueueSize() {
        if (isShutdown) {
            throw new IllegalStateException("The thread pool is destroy");
        }
        return this.runnableQueue.size();
    }

    @Override
    public int getActiveCount() {
        if (isShutdown) {
            throw new IllegalStateException("The thread pool is destroy");
        }
        return this.activeCount;
    }

    @Override
    public boolean isShutdown() {
        return this.isShutdown;
    }

    /**
     * 内部类, 定义自己默认的线程工厂
     */
    private static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger GROUP_COUNTER = new AtomicInteger(1);
        //设置线程组
        private static final ThreadGroup GROUP = new ThreadGroup("MyThreadPool-" + GROUP_COUNTER.getAndDecrement());
        private static final AtomicInteger COUNTER = new AtomicInteger(0);
        @Override
        public Thread createThread(Runnable runnable) {
            //创建定制化的线程
            return new Thread(GROUP, runnable, " thread-pool-" + COUNTER.getAndDecrement());
        }
    }

    /**
     * 封装InternalTask, 与每次创建的线程绑定在一起
     */
    private class ThreadTask {
        Thread thread;
        InternalTask internalTask;

        ThreadTask(Thread thread, InternalTask internalTask){
            this.thread = thread;
            this.internalTask = internalTask;
        }
    }
}

ThreadPoolTest(class 测试类)

package com.concurrent.customthreadpool;

import java.util.concurrent.TimeUnit;

/**
 * 用于测试自定的线程池
 * @author regotto
 */
public class ThreadPoolTest {
    public static void main(String[] args) {
        //初始化线程数:2, 最大线程数:6, 核心线程数:4, 任务队列大小:1000
        final BasicThreadPool threadPool = new BasicThreadPool(2, 6, 4, 1000);
        //创建20个任务提交进行执行
        for (int i = 0; i < 20; ++i) {
            threadPool.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName() + "is running and done.");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        //此处用于测试线程池运行时基本信息状态
//        for (int j = 0; j < 1000; ++j) {
//            System.out.println("getActiveCount: " + threadPool.getActiveCount());
//            System.out.println("getQueueSize: " + threadPool.getQueueSize());
//            System.out.println("getCoreSize: " + threadPool.getCoreSize());
//            System.out.println("getMaxSize: " + threadPool.getMaxSize());
//            try {
//                TimeUnit.SECONDS.sleep(3);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
        try {
            TimeUnit.SECONDS.sleep(25);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("shutdown");
        //测试线程池shutdown功能
        threadPool.shutdown();
    }
}

总结

不要让BasicThreadPool继承Thread, 这样会导致直接调用BasicThreadPool中start方法, 导致线程池崩溃, 应该将Thread设置成组合, 隐藏start();

修改拒绝策略, 做到可以让用户自我实现拒绝方式, 此处拒绝策略可以使用策略模式;

使用工厂模式定义ThreadPool;

对每个参数进行合法验证;

在自己写的过程中, 出现参数传递出错的问题, 导致线程一直wait, 如果遇到这种问题, 使用jconsole工具来查看线程状态, 定位出错的位置, 这样能很快解决遇到的问题.