如何实现java简单的线程池

发布时间:2021-07-16 11:52:53 作者:chen
来源:亿速云 阅读:144

这篇文章主要讲解了“如何实现java简单的线程池”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“如何实现java简单的线程池”吧!

目录

拆分实现流程

请看下面这张图

如何实现java简单的线程池

首先我们得对线程池进行一个功能拆分

现在我们梳理一下执行的流程,注意这里是简略版的,文章后面我会给出详细版的

如何实现java简单的线程池

所以此时,我们发现了需要创建几个类,或者说几个角色,分别是

实现方式

1.拒绝策略

/**
 * 拒绝策略
 */
@FunctionalInterface
interface RejectPolicy<T>{
	//queue就是我们自己实现的阻塞队列,task是任务
    void reject(BlockingQueue<T> queue,T task);
}

2.阻塞队列

我们需要实现四个方法,获取和添加,超时获取和超时添加,至于方法实现的细节,我都备注了大量的注释进行解释。

/**
 * 阻塞队列
 */
class BlockingQueue<T>{
    //阻塞队列
    private Deque<T> queue = new ArrayDeque<>();

    //锁
    private ReentrantLock lock = new ReentrantLock();

    //生产者条件变量
    private Condition fullWaitSet = lock.newCondition();

    //消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //容量
    private int capacity;

    public BlockingQueue(int capacity){
        this.capacity = capacity;
    }

    //带有超时阻塞获取
    public T poll(long timeout, TimeUnit timeUnit){
        lock.lock();
        try {
            //将timeout统一转换为纳秒
            long nanos = timeUnit.toNanos(timeout);
            while(queue.isEmpty()){
                try {
                    if(nanos <= 0){
                        //小于0,说明上次没有获取到,代表已经超时了
                        return null;
                    }
                    //返回值是剩余的时间
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            //通知生产者
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        lock.lock();
        try{
            while(queue.isEmpty()){ //如果任务队列为空,代表线程池没有可以执行的内容
                try {
                     /*
                    也就说此时进来的线程是执行不了任务的,所以此时emptyWaitSet消费者要进行阻塞状态
                    等待下一次唤醒,然后继续判断队列是否为空
                     */
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            /*
            代码执行到这里。说明任务队列不为空,线程池就从任务队列拿出一个任务出来执行
            也就是说把阻塞队列的一个任务出队
             */
            T t = queue.removeFirst();
            /*
            然后唤醒之前存放在生成者Condition休息室,因为由于之前阻塞队列已满,fullWaitSet才会进入阻塞状态
            所以当阻塞队列删除了任务,就要唤醒之前进入阻塞状态的fullWaitSet
             */
            fullWaitSet.signal();
            //返回任务
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞添加
    public void put(T task){
        lock.lock();
        try {
            while(queue.size() == capacity){    //任务队列满了
                try {
                    System.out.println("等待加入任务队列"+task);
                    /*
                    也就说此时进来的任务是进不了阻塞队列的,已经满了,所以此时生产者Condition要进入阻塞状态
                    等待下一次唤醒,然后继续判断队列是否为空
                     */
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //任务队列还未满
            System.out.println("加入任务队列"+task);
            //把任务加入阻塞队列
            queue.addLast(task);
            /*
            然后唤醒之前存放在消费者Condition休息室,因为由于之前阻塞队列为空,emptyWaitSet才会进入阻塞状态
            所以当阻塞队列加入了任务,就要唤醒之前进入阻塞状态的emptyWaitSet
             */
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    //带超时阻塞时间添加
    public boolean offer(T task,long timeout,TimeUnit timeUnit){
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while(queue.size() == capacity){
                try {
                    if(nanos < 0){
                        return false;
                    }
                    System.out.println("等待加入任务队列"+task);
                    //不会一直阻塞,超时就会继续向下执行
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("加入任务队列"+task);
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    //获取任务数量
    public int size(){
        lock.lock();
        try{
            return queue.size();
        }finally {
            lock.unlock();
        }
    }

    //尝试添加任务,如果阻塞队列已经满了,就使用拒绝策略
    public void tryPut(RejectPolicy<T> rejectPolicy, T task){
        lock.lock();
        try {
            //判断队列是否已满
            if(queue.size() == capacity){
                rejectPolicy.reject(this,task);
            }else{  //有空闲
                System.out.println("加入任务队列"+task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }
}

3.线程池和工作线程

我把工作线程当成线程池的内部类去实现。方便调用变量。

/**
 * 线程池
 */
class ThreadPool{
    //阻塞队列
    private BlockingQueue<Runnable> taskQueue;

    //线程集合
    private HashSet<Worker> workers = new HashSet<>();

    //核心线程数
    private int coreSize;

    //获取任务的超时时间
    private long timeout;

    private TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }

    //执行任务
    public void execute(Runnable task){
        synchronized (workers){
            if(workers.size() <= coreSize){  //当前的线程数小于核心线程数
                Worker worker = new Worker(task);
                workers.add(worker);
                //让线程开始工作,执行它的run方法
                worker.start();
            }else{
                // 1) 死等
                // 2) 带超时等待
                // 3) 让调用者放弃任务执行
                // 4) 让调用者抛出异常
                // 5) 让调用者自己执行任务
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }

    /**
     * 工作线程,也就是线程池里面的线程
     */
    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable task){
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            // 1) 当 task 不为空,执行任务
            // 2) 当 task 执行完毕,再接着从任务队列获取任务并执行
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    System.out.println("正在执行的任务" + task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    //代表这个任务已经执行完了
                    task = null;
                }
            }
            synchronized (workers) {
                System.out.println("worker 被移除" + this);
                workers.remove(this);
            }
        }
    }
}

策略模式

细心的小伙伴已经发现,我在拒绝策略这里使用了23种设计模式的策略模式,因为我没有将拒绝的方式写死,而是交给了调用者去实现。

对比JDK的线程池

下面是JDK自带的线程池

如何实现java简单的线程池

经典的七大核心参数

实际上我们自己实现的也大同小异,只不过JDK官方的更为复杂。

JDK线程执行的流程图

如何实现java简单的线程池

如何实现java简单的线程池

线程池的状态转化

线程我们知道在操作系统层面有5种状态

如何实现java简单的线程池

线程在Java API层面有6种状态

如何实现java简单的线程池

线程池有5种状态

如何实现java简单的线程池

感谢各位的阅读,以上就是“如何实现java简单的线程池”的内容了,经过本文的学习后,相信大家对如何实现java简单的线程池这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!

推荐阅读:
  1. Java的线程池
  2. java线程池如何实现的?

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

java 线程池

上一篇:asp.net中怎么声明公共变量

下一篇:Web开发中客户端跳转与服务器端跳转有什么区别

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》