当前位置：网站首页>多线程之自定义线程池

多线程之自定义线程池

2022-08-10 00:44:00 【七国的天下，我要九十九】

多线程之自定义线程池

1 自义定线程池

在这里插入图片描述

1定义

自定义线程池组成部分: 核心线程数,等待超时时间,等待时间单位,任务队列,拒绝策略.

2 组成部分

1 自定义拒绝策略接口

// 拒绝策略
@FunctionalInterface 
interface RejectPolicy<T> {
    
     void reject(BlockingQueue<T> queue, T task);
}

拒绝策略接口,主要用来线程池做拒绝任务实现,根据不同实现的子类,来定义拒绝的策略,接口本身定义为函数式接口,更加灵活由调用者控制.

2 自定义任务队列

class BlockingQueue<T> {
    
     // 1. 任务队列
     private Deque<T> queue = new ArrayDeque<>();
     // 2. 锁
     private ReentrantLock lock = new ReentrantLock();
     // 3. 生产者条件变量
     private Condition fullWaitSet = lock.newCondition();
     // 4. 消费者条件变量
     private Condition emptyWaitSet = lock.newCondition();
     // 5. 容量
     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.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 {
    
         log.debug("等待加入任务队列 {} ...", task);
         fullWaitSet.await();
         } catch (InterruptedException e) {
    
             e.printStackTrace();
         }
     }
         log.debug("加入任务队列 {}", 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() == capcity) {
    
             try {
    
                 if(nanos <= 0) {
    
                 return false;
                 }
                 log.debug("等待加入任务队列 {} ...", task);
                 nanos = fullWaitSet.awaitNanos(nanos);

             } catch (InterruptedException e) {
    
                 e.printStackTrace();
             }
         }
         log.debug("加入任务队列 {}", 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() == capcity) {
    
                 rejectPolicy.reject(this, task);
             } else {
     
                 // 有空闲
                 log.debug("加入任务队列 {}", 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 void execute(Runnable task) {
    
         // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
         // 如果任务数超过 coreSize 时，加入任务队列暂存
         synchronized (workers) {
    
             if(workers.size() < coreSize) {
    
                 Worker worker = new Worker(task);
                 log.debug("新增 worker{}, {}", worker, task);
                 workers.add(worker);
                 worker.start();
             } else {
    
                // taskQueue.put(task);
                 // 1) 死等
                 // 2) 带超时等待
                 // 3) 让调用者放弃任务执行
                 // 4) 让调用者抛出异常
                 // 5) 让调用者自己执行任务
                 taskQueue.tryPut(rejectPolicy, task);
             }
         }
     }
    
 public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, 
RejectPolicy<Runnable> rejectPolicy) {
    
     this.coreSize = coreSize;
     this.timeout = timeout;
     this.timeUnit = timeUnit;
     this.taskQueue = new BlockingQueue<>(queueCapcity);
     this.rejectPolicy = rejectPolicy;
 }
    
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.take()) != null) {
    
         while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
    
             try {
    
                 log.debug("正在执行...{}", task);

                 task.run();
             } catch (Exception e) {
    
                 e.printStackTrace();
             } finally {
    
                 task = null;
             }
         }

         synchronized (workers) {
    
         log.debug("worker 被移除{}", this);
         workers.remove(this);
         }
     }
}
}

线程池,主要是用来获取执行任务的对象.根据自定义的任务队列,拒绝策略,构造出线程池, 执行任务时,从线程池中,获取线程,执行任务,当任务数大于核心线程数,则会添加到任务队列, 当任务队列已满,则会执行拒绝策略.

4 测试

public static void main(String[] args) {
    
     ThreadPool threadPool = new ThreadPool(1,
     1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{
    
         // 1. 死等
        // queue.put(task);
         // 2) 带超时等待
        // queue.offer(task, 1500, TimeUnit.MILLISECONDS);
         // 3) 让调用者放弃任务执行
        // log.debug("放弃{}", task);
         // 4) 让调用者抛出异常
        // throw new RuntimeException("任务执行失败 " + task);
         // 5) 让调用者自己执行任务
         task.run();
     });
     for (int i = 0; i < 4; i++) {
    
         int j = i;
         threadPool.execute(() -> {
    
             try {
    
                 Thread.sleep(1000L);
             } catch (InterruptedException e) {
    
                 e.printStackTrace();
             }
             log.debug("{}", j);
         });
     }
 }

执行任务时,根据调用者不同的需求,来自定义不同的拒绝策略,起到既满足业务需求,又能提高代码效率. (如JDK代码中, 也可将不同拒绝策略定义成枚举,然后更加不同枚举走不同的拒绝策略)

原网站

版权声明
本文为[七国的天下，我要九十九]所创，转载请带上原文链接，感谢
https://blog.csdn.net/ABestRookie/article/details/126255940

边栏推荐

猜你喜欢

随机推荐