如何实现java简单的线程池

2023-06-20 16:06:04 213人浏览安东尼

摘要

这篇文章主要讲解了“如何实现java简单的线程池”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“如何实现java简单的线程池”吧！目录拆分实现流程实现方式拒绝策略阻塞队列线程池和工作线程策略模

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

拆分实现流程

请看下面这张图

如何实现java简单的线程池

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

Thread Pool 就是我们的线程池，t1，t2，t3代表三个线程
Blocking Queue代表阻塞队列
main代表main方法的线程
task1，task2，task3代表要执行的每个任务

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

如何实现java简单的线程池

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

线程池
工作线程
阻塞队列
拒绝策略（干嘛的？就是当线程数已经满了，并且阻塞队列也满了，还有任务想进入阻塞队列的时候，就可以拒绝这个任务）

实现方式

1.拒绝策略

@FunctionalInterfaceinterface 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.await();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }                        T t = queue.removeFirst();                        fullWaitSet.signal();            //返回任务            return t;        }finally {            lock.unlock();        }    }    //阻塞添加    public void put(T task){        lock.lock();        try {            while(queue.size() == capacity){    //任务队列满了                try {                    System.out.println("等待加入任务队列"+task);                                        fullWaitSet.await();                } catch (InterruptedException e) {                    e.printStackTrace();                }            }            //任务队列还未满            System.out.println("加入任务队列"+task);            //把任务加入阻塞队列            queue.addLast(task);                        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);            }        }    }}