Java多线程与并发编程(7)_自定义线程池

冷不防 2023-02-19 10:22 12阅读 0赞

# Java多线程与并发编程(7)\_自定义线程池 #

## 线程池架构 ##

![format_png][]

*  任务队列：存放任务
 *  线程集合：存放线程
 *  线程集合中的线程从任务队列中拉取任务进行消费

## 编写自定义拒绝策略 ##

当任务数目大于线程集合中线程数目的时候，拒绝添加任务

使用函数式接口，将拒绝权限下放，由调用者决定添加任务失败时线程池的回绝方式

> 1.  死等
> 2.  带超时等待
> 3.  让调用者放弃任务执行
> 4.  让调用者抛出异常
> 5.  让调用者自己执行任务

//步骤一：自定义拒绝策略接口
    @FunctionalInterface 
    interface RejectPolicy<T>{
        
        void reject(BlockingQueue<T> queue, T task) ;
    }

利用该接口可以由消费者自己决定采用何种策略

## 编写自定义阻塞队列 ##

阻塞队列用来暂存任务，超过线程池核心线程数的部分线程，将存放在阻塞队列中，如果阻塞队列也满了，将拒绝继续向阻塞队列中添加任务

//自定义阻塞队列
    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 capcity;
        //构造器
        public BlockingQueue(int capcity) {
        
            this.capcity = capcity;
        }
    
        //返回任务队列头部的元素给消费者（线程集合）
        // 超时机制版的阻塞获取，一定时间内无任务则不再进行获取
        public T poll(long timeout, TimeUnit unit) {
        
    
            lock.lock(); //获取操作需要同步
            try {
        
            //将 timeout 统一转换为 纳秒
            long nanos = unit.toNanos(timeout);
            //任务队列中已经空了（无任务）
            while (queue.isEmpty()) {
        
                try {
        
                    //等待时间超时，返回空值
                    if (nanos <= 0) {
        
                        return null;
                    }
                    //进入等待队列
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
            //任务队列不空，返回队列头部元素
            T t = queue.removeFirst();
              //从fullWaitSet唤醒因任务队列满了而等待的线程
              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() == capcity) {
        
                    try {
        
                        //进入等待fullWaitSet等待队列进行等待
                        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 unit){
        
            lock.lock();
            try {
        
                long nanos = unit.toNanos(timeout);
                //队列满
                while (queue.size()==capcity) {
        
                    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();
            }
        }
    
        //带拒绝策略的put，即如果任务队列无法添加元素，使用何种策略进行拒绝添加
        public void tryPut(RejectPolicy<T> rejectPolicy,T task){
        
            lock.lock();
            try {
        
                //判断队列是否满，如满，执行拒绝策略 --> 策略模式
                if (queue.size() == capcity) rejectPolicy.reject(this,task);
            else {
        
                    System.out.println("加入任务队列"+task);
                    queue.addLast(task);
                    emptyWaitSet.signal();
                }
            }finally {
        
                lock.unlock();
            }
        }
    
    }

小结：

![format_png 1][]

*  本代码实现了阻塞队列头元素获取及尾元素添加
 *  将阻塞队列中的各个任务存放于双端队列
 *  构造器将初始化该阻塞队列的核心容量
 *  添加与获取都提供了超时回空策略，避免长时间无添加与获取造成的性能浪费

## 自定义线程池 ##

//自定义线程池
    class ThreadPool{
        
        public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,RejectPolicy<Runnable> rejectPolicy) {
        
            //核心线程数
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.timeUnit = timeUnit;
            //拒绝策略
            this.rejectPolicy = rejectPolicy;
            //任务队列
            this.taskQueue = new BlockingQueue<>(queueCapcity) ;
        }
    
        //对核心线程的封装，使用Worker对线程进行了扩展
        class Worker extends Thread{
        
            private Runnable task ;
    
            public Worker(Runnable task) {
        
                this.task = task;
            }
            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);
                }
            }
        }
        //任务队列
        private BlockingQueue<Runnable> taskQueue ;
        //线程集合，集合中使用Worker对线程进行了扩展
        private HashSet<Worker> workers = new HashSet<>();
        //核心线程数
        private int coreSize ;
        //获取任务时的超时时间
        private long timeout ;
        //超时时间的单位规范
        private TimeUnit timeUnit ;
        //拒绝策略，由主线程自己决定，主线程取实现这个RejectPolicy，并传入本线程池即可
        private RejectPolicy<Runnable> rejectPolicy ;
    
        //执行任务
        public void execute(Runnable task){
        
            // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行，注意，Work执行完毕后不能立即消失，要被循环利用
            // 如果任务数超过 coreSize 时，加入任务队列暂存
            synchronized (workers){
        
    
                //如果线程集合中的线程数量少于核心线程数，创建Work对象来执行任务
                if (workers.size() < coreSize){
        
                      Worker worker = new Worker(task);
                      System.out.println("新增worker"+worker+"task"+task);
                      workers.add(worker) ;
                      worker.start();
                }else {
        
                    // 各种拒绝策略
                    // 1) 死等
                    // 2) 带超时等待
                    // 3) 让调用者放弃任务执行
                    // 4) 让调用者抛出异常
                    // 5) 让调用者自己执行任务
    
                    //加入任务队列，等待被执行
                    taskQueue.tryPut(rejectPolicy,task);
                }
            }
        }
    }

小结：

这就是融合了阻塞队列与拒绝策略的完整线程池了，包括

*  阻塞队列
 *  线程集合
 *  执行方法
    
     *  创建线程来执行任务
        
         *  如果线程集合中的线程数少于线程池容量，则创建线程直接执行任务
         *  如果线程集合中的线程数大于线程池容量，则将任务放入阻塞队列，由线程执行完任务后拉去阻塞队列中的任务来执行

##  ##

public class MyThreadPool {
        
        public static void main(String[] args) {
        
            ThreadPool threadPool = new ThreadPool(1,
                    1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{
        
                task.run();
            });
            for (int i = 0; i < 4; i++) {
        
                int j = i;
                threadPool.execute(() -> {
        
                        Thread.sleep(1000L);
                    System.out.println(j); //执行打印j
                });
            }
        }
    }

方便演示拒绝策略是否生效，我们指定：

*  线程池中只有1个线程，超时等待时间为1s，阻塞队列容量为1个

拒绝策略我们选择：

*  如果线程池拒绝接收任务，则调用线程池的线程自己把任务执行掉

> **结果**：

![format_png 2][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwMTIwMzAucG5n?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwMTI3MzcucG5n?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwMTQ4NDAucG5n?x-oss-process=image/format,png