自定义线程池

自定义线程池类图
image.pngimage.png

ThreadPool

ThreadPool是核心线程池，线程池在初始化时传入参数，构建任务队列，线程池大小，队列容量，以及队列满时的拒绝策略等。其中Worker是工作线程，用来执行传入的任务，

  1. public class ThreadPool {
  3.     /**
  4.      * 线程池核心线程数量
  5.      */
  6.     private int coreSize;
  8.     /**
  9.      * 提交的任务队列
  10.      */
  11.     private BlockQueue<Runnable> queue;
  13.     /**
  14.      * 拒接策略
  15.      */
  16.     private RejectPolicy<Runnable> rejectPolicy;
  18.     /**
  19.      * 任务的超时时间
  20.      */
  21.     private long timeOut;
  23.     /**
  24.      * 超时时间单位
  25.      */
  26.     private TimeUnit timeUnit;
  28.     /**
  29.      * 工作线程集合
  30.      */
  31.     private HashSet<Worker> workers = new HashSet();
  33.     public ThreadPool(int coreSize, int queueSize, long timeOut, TimeUnit timeUnit) {
  34.         this.coreSize = coreSize;
  35.         this.queue = new BlockQueue<>(queueSize);
  36.         this.timeOut = timeOut;
  37.         this.timeUnit = timeUnit;
  38.     }
  40.     public ThreadPool(int coreSize, int queueSize, long timeOut, TimeUnit timeUnit, RejectPolicy<Runnable> rejectPolicy) {
  41.         this.coreSize = coreSize;
  42.         this.queue = new BlockQueue<>(queueSize);
  43.         this.rejectPolicy = rejectPolicy;
  44.         this.timeOut = timeOut;
  45.         this.timeUnit = timeUnit;
  46.     }
  48.     /**
  49.      * 执行任务
  50.      */
  51.     public void execute(Runnable task) {
  52.         if (workers.size() < coreSize) {
  53.             log.debug("线程池空间未满,直接创建线程...");
  54.             Worker worker = new Worker(task);
  55.             workers.add(worker);
  56.             worker.start();
  57.         } else {
  58.             log.debug("线程池空间已满,提交任务进入队列等待...");
  59.             log.info("线程池已满,任务放入队列...");
  60.             //            queue.put(task);
  61.             //            queue.offer(task, 500, TimeUnit.MILLISECONDS);
  62.             queue.tryPut(rejectPolicy, task);
  63.         }
  64.     }
  65. }

Worker

Worker是一个工作线程，用来执行传入的任务，执行任务前需要判断任务是否为空，或者从队列中取出一个不为空的任务执行，如果没有任务，则线程结束。

  1.     /**
  2.      * 工作线程
  3.      */
  4.     class Worker extends Thread {
  5.         /**
  6.          * 执行的任务
  7.          */
  8.         private Runnable task;
  10.         public Worker(Runnable task) {
  11.             this.task = task;
  12.         }
  14.         /**
  15.          * 执行任务
  16.          * 1) 当 task 不为空，执行任务
  17.          * 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
  18.          */
  19.         @Override
  20.         public void run() {
  21.             //            while (null != task || (task = queue.take()) != null) {
  22.             while (null != task || (task = queue.poll(500, TimeUnit.MILLISECONDS)) != null) {
  23.                 try {
  24.                     log.debug("{}正在执行...", task);
  25.                     task.run();
  26.                 } catch (Exception e) {
  27.                     log.debug("任务执行异常...");
  28.                     e.printStackTrace();
  29.                 } finally {
  30.                     log.debug("{}执行结束...", task);
  31.                     task = null;
  32.                 }
  33.             }
  34.             synchronized (workers) {
  35.                 log.debug("所有任务都被执行完毕,工作线程结束,workers是共享变量,需要保护");
  36.                 workers.remove(this);
  37.             }
  38.         }
  39.     }

BlockQueue

BlockQueue是自定义的一个队列，队列中有相应的存和取的方法，用于存放和消费任务，因为对列会被多个线程共享，所以需要对其进行线程安全保护，所以额外加了锁保护，以及等待队列。

  1. public class BlockQueue<T> {
  3.     /**
  4.      * 存放任务的队列
  5.      */
  6.     private Deque<T> queue;
  8.     /**
  9.      * 保证线程安全的锁
  10.      */
  11.     private ReentrantLock lock = new ReentrantLock();
  13.     /**
  14.      * 队列已满的等待集合
  15.      */
  16.     private Condition fullCondition = lock.newCondition();
  18.     /**
  19.      * 对列空闲的等待集合
  20.      */
  21.     private Condition emptyCondition = lock.newCondition();
  23.     /**
  24.      * 队列的最大容量
  25.      */
  26.     private int capcity;
  28.     public BlockQueue(int capcity) {
  29.         this.capcity = capcity;
  30.         this.queue = new ArrayDeque<>(capcity);
  31.     }
  33.     /**
  34.      * 带超时时间的加入任务,如果在规定时间内未完成队列加入,则丢弃任务
  35.      */
  36.     public void offer(T t, long timeOut, TimeUnit timeUnit) {
  37.         lock.lock();
  38.         try {
  39.             log.debug("统一时间单位...");
  40.             long nanos = timeUnit.toNanos(timeOut);
  41.             while (queue.size() == capcity) {
  42.                 log.debug("队列已满,等待任务被执行后再加入...");
  43.                 nanos = fullCondition.awaitNanos(nanos);
  44.                 if (nanos < 0) {
  45.                     log.debug("超时结束等待,丢弃任务,awaitNanos 返回的是,入如果当前线程被虚假唤醒,则当前线程还应该等多久");
  46.                     return;
  47.                 }
  48.             }
  49.             log.debug("队列还有空间...");
  50.             queue.addLast(t);
  51.             log.debug("通知消费线程取任务...");
  52.             emptyCondition.signalAll();
  53.         } catch (InterruptedException e) {
  54.             log.debug("阻塞被打断...");
  55.             e.printStackTrace();
  56.         } finally {
  57.             lock.unlock();
  58.         }
  60.     }
  62.     /**
  63.      * @throws
  64.      * @description 带拒绝策略的put
  65.      * @author SongHongWei
  66.      * @params
  67.      * @updateTime 2021/10/4 17:23
  68.      */
  69.     public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
  70.         lock.lock();
  71.         try {
  72.             if (queue.size() == capcity) {
  73.                 log.info("队列已满,等待任务被执行后再加入...");
  74.                 rejectPolicy.reject(this, task);
  75.             } else {
  76.                 log.debug("队列还有空间...");
  77.                 queue.addLast(task);
  78.                 log.debug("通知消费线程取任务...");
  79.                 emptyCondition.signalAll();
  80.             }
  81.         } finally {
  82.             lock.unlock();
  83.         }
  84.     }
  86.     /**
  87.      * 在对列里加入新任务
  88.      * 阻塞添加
  89.      *
  90.      * @param t
  91.      */
  92.     public void put(T t) {
  93.         lock.lock();
  94.         try {
  95.             //不用if的原因是防止线程被虚假唤醒,即条件不满足时被唤醒
  96.             while (queue.size() == capcity) {
  97.                 log.info("队列已满,等待任务被执行后再加入...");
  98.                 fullCondition.await();
  99.             }
  100.             log.debug("队列还有空间...");
  101.             queue.addLast(t);
  102.             log.debug("通知消费线程取任务...");
  103.             emptyCondition.signalAll();
  104.         } catch (InterruptedException e) {
  105.             log.debug("阻塞被打断...");
  106.             e.printStackTrace();
  107.         } finally {
  108.             lock.unlock();
  109.         }
  110.     }
  112.     /**
  113.      * 获取任务,需要考虑线程安全
  114.      * 阻塞获取
  115.      *
  116.      * @return
  117.      */
  118.     public T take() {
  119.         lock.lock();
  120.         try {
  121.             while (queue.isEmpty()) {
  122.                 log.info("队列已空,等待任务加入...");
  123.                 try {
  124.                     emptyCondition.await();
  125.                 } catch (InterruptedException e) {
  126.                     log.debug("阻塞被打断...");
  127.                     e.printStackTrace();
  128.                 }
  129.             }
  130.             T first = queue.removeFirst();
  131.             log.debug("唤醒因队列满了导致在阻塞中的put线程");
  132.             fullCondition.signalAll();
  133.             return first;
  134.         } finally {
  135.             lock.unlock();
  136.         }
  137.     }
  139.     /**
  140.      * 带超时时间的任务获取
  141.      *
  142.      * @param timeOut
  143.      * @param timeUnit
  144.      * @return
  145.      */
  146.     public T poll(long timeOut, TimeUnit timeUnit) {
  147.         lock.lock();
  148.         try {
  149.             long nanos = timeUnit.toNanos(timeOut);
  150.             while (queue.isEmpty()) {
  151.                 log.info("队列已空,等待任务加入...");
  152.                 try {
  153.                     nanos = emptyCondition.awaitNanos(nanos);
  154.                     if (nanos < 0) {
  155.                         log.debug("超时结束等待,丢弃任务,awaitNanos 返回的是,入如果当前线程被虚假唤醒,则当前线程还应该等多久");
  156.                         return null;
  157.                     }
  158.                 } catch (InterruptedException e) {
  159.                     log.debug("阻塞被打断...");
  160.                     e.printStackTrace();
  161.                 }
  162.             }
  163.             T first = queue.removeFirst();
  164.             log.debug("唤醒因队列满了导致在阻塞中的put线程");
  165.             fullCondition.signalAll();
  166.             return first;
  167.         } finally {
  168.             lock.unlock();
  169.         }
  170.     }
  171. }

RejectPolicy

RejectPolicy是一个接口函数，用来指定当队列满的时候，采用什么样的方式任务，例如，继续等待直到队列有任务被消费了，或者带超时时间的等待，或者放弃任务，或者抛出异常等等。

  1. /**
  2.  * @author fmj
  3.  * 抽象一个拒接策略,用来处理当线程池的线程都处于busy状态,还有任务提交进来时的处理方式
  4.  * @date 2021 2021/9/24 15:06
  5.  */
  6. @FunctionalInterface
  7. public interface RejectPolicy<T> {
  8.     void reject(BlockQueue<T> queue, T task);
  9. }

PoolClient

PoolClient是用来调用线程池的，模拟调用场景

  1. /**
  2.  * @author fmj
  3.  * 模拟调用自定义线程池
  4.  * @date 2021 2021/9/24 15:07
  5.  */
  6. @Slf4j
  7. public class PoolClient {
  9.     public static void main(String[] args) {
  10.         ThreadPool pool = new ThreadPool(2, 3, 1000L, TimeUnit.MILLISECONDS);
  11.         ThreadPool rejectPool = new ThreadPool(2, 3, 1000L, TimeUnit.MILLISECONDS, (q, t) -> {
  12. //            log.info("1.队列满时一直等待,知道有线程执行完其他任务");
  13. //            q.put(t);
  14. //            log.info("2.队列满时执行带超时时间的等待");
  15. //            q.offer(t, 2000L, TimeUnit.SECONDS);
  16.             log.info("3.队列满时,放弃当前任务");
  17. //            log.info("4.队列满时,由调用者自己执行");
  18. //            t.run();
  19. //            log.info("5.队列满时,抛出异常");
  20. //            throw new RuntimeException("队列已满");
  21.         });
  22.         for (int i = 0; i < 10; i++) {
  23.             int j = i;
  24.             rejectPool.execute(() -> {
  25.                 try {
  26.                     Thread.sleep(1000);
  27.                     log.info("{}", j);
  28.                 } catch (InterruptedException e) {
  29.                     e.printStackTrace();
  30.                 }
  31.             });
  32.         }
  33.     }
  34. }