一、自定义线程池

在看JDK的线程池之前，我们先来自定义一个线程池：

1.图解：

线程池 - 图1

  • 阻塞队列中维护了由主线程（或者其他线程）所产生的的任务
  • 主线程类似于生产者，产生任务并放入阻塞队列中
  • 线程池类似于消费者，得到阻塞队列中已有的任务并执行

2.具体实现

  1. package panw.ThreadPool;
  3. import javafx.concurrent.Worker;
  4. import lombok.extern.slf4j.Slf4j;
  6. import java.util.ArrayDeque;
  7. import java.util.Deque;
  8. import java.util.HashSet;
  9. import java.util.concurrent.TimeUnit;
  10. import java.util.concurrent.locks.Condition;
  11. import java.util.concurrent.locks.ReentrantLock;
  12. @Slf4j
  13. public class ThreadPoolTest {
  14.     public static void main(String[] args) {
  15.         ThreadPool threadPool = new ThreadPool(1, 1, TimeUnit.SECONDS, 1,
  16.                 (queue,task)->{
  17. //                    queue.put(task);    //  死等
  18. //                    queue.offer(task,50,TimeUnit.MILLISECONDS); //超时
  19. //                    log.debug("放弃{}",task); //放弃,后面任务继续会执行
  20. //                    throw new RuntimeException("任务执行失败"+task);  //抛出异常，后面任务不执行
  21. //                    task.run();             //自己执行
  22.                 }
  23.                 );
  24.         for (int i = 0; i < 15; i++) {
  25.             int j=i;
  26.             threadPool.execute(()->{
  27.                 try {
  28.                     Thread.sleep(1000);
  29.                 } catch (InterruptedException e) {
  30.                     throw new RuntimeException(e);
  31.                 }
  32.                 log.debug("{}",j);
  33.             });
  34.         }
  35.     }
  36. }
  37. @Slf4j
  38. class ThreadPool{
  39.     private BlockingQueue<Runnable> queue ;
  40.     private HashSet<Worker> workers =new HashSet<>();
  41.     private int coreSize;
  42.     private long timeout;
  43.     private TimeUnit timeUnit;
  44.     private RejectPolicy<Runnable> rejectPolicy;
  46.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int capacity,RejectPolicy<Runnable> rejectPolicy) {
  47.         this.coreSize = coreSize;
  48.         this.timeout = timeout;
  49.         this.timeUnit = timeUnit;
  50.         queue=new BlockingQueue<>(capacity);
  51.         this.rejectPolicy=rejectPolicy;
  52.     }
  53.     public void execute(Runnable task){
  54.         synchronized (workers){
  55.             if (workers.size()<coreSize){
  56.                 Worker worker = new Worker(task);
  57.                 log.debug("新增 worker{}, {}", worker, task);
  58.                 workers.add(worker);
  59.                 worker.start();
  60.             }else {
  61. //                queue.put(task);
  62.                 queue.tryPut(rejectPolicy,task);
  63.             }
  64.         }
  65.     }
  67.     class Worker extends Thread{
  68.         private Runnable task;
  70.         public Worker(Runnable task) {
  71.             this.task = task;
  72.         }
  74.         @Override
  75.         public void run() {
  76.             while (task!=null||(task=queue.poll(timeout,timeUnit))!=null){
  77.                 try {
  78.                     log.debug("正在执行...{}", task);
  79.                     task.run();
  80.                 }catch (Exception e){
  82.                 }finally {
  83.                     task=null;
  84.                 }
  85.             }
  86.             synchronized (workers){
  87.                 log.debug("worker 被移除{}", this);
  88.                 workers.remove(this);
  89.             }
  92.         }
  93.     }
  96. }
  97. @Slf4j
  98. class BlockingQueue<T>{
  99.     private Deque<T> queue ;
  100.     private int capacity;
  101.     private ReentrantLock lock;
  102.     private Condition fullWait;
  103.     private Condition emptyWait;
  105.     public BlockingQueue(int capacity) {
  106.         this.queue=new ArrayDeque<>();
  107.         this.capacity=capacity;
  108.         this.lock= new ReentrantLock();
  109.         this.fullWait=lock.newCondition();
  110.         this.emptyWait= lock.newCondition();
  111.     }
  112.     //阻塞添加
  113.     public void put(T t){
  114.         lock.lock();
  115.         try {
  116.             while (queue.size()==capacity){
  117.                 log.debug("等待加入任务队列 {}", t);
  118.                 fullWait.await();
  119.             }
  121.             log.debug("加入任务队列 {}", t);
  122.             queue.addLast(t);
  123.             emptyWait.signal();
  124.         } catch (InterruptedException e) {
  125.             throw new RuntimeException(e);
  126.         } finally {
  127.             lock.unlock();
  128.         }
  129.     }
  130.     //带超时添加
  131.     public boolean offer(T t,long timeout,TimeUnit timeUnit){
  132.         lock.lock();
  133.         long nanos = timeUnit.toNanos(timeout);
  134.         try {
  135.             while (queue.size()==capacity){
  136.                 if (nanos<=0){
  137.                     log.debug("超时了...删除{}",t);
  138.                     return false;
  139.                 }
  140.                 log.debug("等待加入任务队列 {} ...", t);
  141.                 nanos = fullWait.awaitNanos(nanos);
  142.             }
  143.             log.debug("加入任务队列 {}", t);
  145.             queue.addLast(t);
  146.             emptyWait.signal();
  147.             return true;
  148.         } catch (InterruptedException e) {
  149.             throw new RuntimeException(e);
  150.         } finally {
  151.             lock.unlock();
  152.         }
  153.     }
  156.     //阻塞获取
  157.     public T take(){
  158.         lock.lock();
  159.         try {
  160.             while (queue.isEmpty()){
  161.                 emptyWait.await();
  162.             }
  163.             T t = queue.removeFirst();
  164.             fullWait.signal();
  165.             return t;
  166.         } catch (InterruptedException e) {
  167.             throw new RuntimeException(e);
  168.         } finally {
  169.             lock.unlock();
  170.         }
  171.     }
  173.     //带超时获取
  174.     public T poll(long timeout, TimeUnit timeUnit){
  175.         lock.lock();
  176.         long nanos = timeUnit.toNanos(timeout);
  177.         try {
  178.             while (queue.isEmpty()){
  179.                 if (nanos<=0){
  180.                     return null;
  181.                 }
  182.                 nanos = emptyWait.awaitNanos(nanos);
  183.             }
  184.             T t = queue.removeFirst();
  185.             fullWait.signal();
  186.             return t;
  188.         } catch (InterruptedException e) {
  189.             throw new RuntimeException(e);
  190.         } finally {
  191.             lock.unlock();
  192.         }
  193.     }
  195.     //容量
  196.     public int size(){
  197.         lock.lock();
  198.         try {
  199.             return queue.size();
  201.         }finally {
  202.             lock.unlock();
  203.         }
  204.     }
  206.     public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
  207.         lock.lock();
  208.         try {
  209.             //判断队列是否已满
  210.             if (queue.size()==capacity){
  211.                 rejectPolicy.reject(this,task);
  212.             }else{
  213.                 log.debug("加入阻塞队列{}",task);
  214.                 queue.addLast(task);
  215.                 emptyWait.signal();
  217.             }
  218.         }finally {
  219.             lock.unlock();
  220.         }
  221.     }
  222. }
  223. @FunctionalInterface
  224. interface RejectPolicy<T>{
  225.     void reject(BlockingQueue<T> queue,T task);
  226. }

二、ThreadPoolExecutor

1.继承关系

image.png

2.线程池状态

  1.  private static final int COUNT_BITS = Integer.SIZE - 3;
  3. private static final int RUNNING    = -1 << COUNT_BITS;
  4. private static final int SHUTDOWN   =  0 << COUNT_BITS;
  5. private static final int STOP       =  1 << COUNT_BITS;
  6. private static final int TIDYING    =  2 << COUNT_BITS;
  7. private static final int TERMINATED =  3 << COUNT_BITS;

从源码可以看出:

状态名称 高3位的值 描述
RUNNING 111 接收新任务，同时处理任务队列中的任务
SHUTDOWN 000 不接受新任务，但是处理任务队列中的任务
STOP 001 中断正在执行的任务，同时抛弃阻塞队列中的任务
TIDYING 010 任务执行完毕，活动线程为0时，即将进入终结阶段
TERMINATED 011 终结状态 
  1.  // 原子整数，前3位保存了线程池的状态，剩余位保存的是线程数量
  2. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  4. // 并不是所有平台的int都是32位。
  5. // 去掉前三位保存线程状态的位数，剩下的用于保存线程数量
  6. // 高3位为0，剩余位数全为1
  7. private static final int COUNT_BITS = Integer.SIZE - 3;
  9. // 2^COUNT_BITS次方，表示可以保存的最大线程数
  10. // CAPACITY 的高3位为 0
  11. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

线程池状态和线程池中线程的数量由一个原子整型ctl来共同表示

  • 使用一个数来表示两个值的主要原因是：可以通过一次CAS同时更改两个属性的值

线程池内部属性也是包含了上述自定义线程池的。

  1. // 工作线程，内部封装了Thread
  2. private final class Worker
  3.         extends AbstractQueuedSynchronizer
  4.         implements Runnable {
  5.     ...
  6. }
  8. // 阻塞队列，用于存放来不及被核心线程执行的任务
  9. private final BlockingQueue<Runnable> workQueue;
  11. // 锁
  12. private final ReentrantLock mainLock = new ReentrantLock();
  14. //  用于存放核心线程的容器，只有当持有锁时才能够获取其中的元素（核心线程）
  15. private final HashSet<Worker> workers = new HashSet<Worker>();

3.构造方法及参数

直接来看看最多参数的构造：

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                               int maximumPoolSize,
  3.                               long keepAliveTime,
  4.                               TimeUnit unit,
  5.                               BlockingQueue<Runnable> workQueue,
  6.                               ThreadFactory threadFactory,
  7.                               RejectedExecutionHandler handler)

参数解释

  • corePoolSize：核心线程数
  • maximumPoolSize：最大线程数
    • maximumPoolSize - corePoolSize = 救急线程数
  • keepAliveTime：救急线程空闲时的最大生存时间
  • unit：时间单位
  • workQueue：阻塞队列（存放任务）
    • 有界阻塞队列 ArrayBlockingQueue
    • 无界阻塞队列 LinkedBlockingQueue
    • 最多只有一个同步元素的 SynchronousQueue
    • 优先队列 PriorityBlockingQueue
  • threadFactory：线程工厂（给线程取名字）不重要

  • handler：拒绝策略 不重要

    工作流程：

  • 当一个任务传给线程池以后，可能有以下几种可能

    • 将任务分配给一个核心线程来执行
    • 核心线程都在执行任务，将任务放到阻塞队列workQueue中等待被执行
    • 阻塞队列满了，使用救急线程来执行任务
      • 救急线程用完以后，超过生存时间（keepAliveTime）后会被释放
    • 任务总数大于了 最大线程数（maximumPoolSize）与阻塞队列容量的最大值（workQueue.capacity），使用拒绝策略

      拒绝策略

      20201022194718.png

  • AbortPolicy：让调用者抛出 RejectedExecutionException 异常，这是默认策略

    1.  public static class AbortPolicy implements RejectedExecutionHandler {
    2.       /**
    3.        * Creates an {@code AbortPolicy}.
    4.        */
    5.       public AbortPolicy() { }
    7.       /**
    8.        * Always throws RejectedExecutionException.
    9.        *
    10.        * @param r the runnable task requested to be executed
    11.        * @param e the executor attempting to execute this task
    12.        * @throws RejectedExecutionException always
    13.        */
    14.       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    15.           throw new RejectedExecutionException("Task " + r.toString() +
    16.                                                " rejected from " +
    17.                                                e.toString());
    18.       }
    19.   }

  • DiscardPolicy：放弃本次任务

    1. public static class DiscardPolicy implements RejectedExecutionHandler {
    2.       /**
    3.        * Creates a {@code DiscardPolicy}.
    4.        */
    5.       public DiscardPolicy() { }
    7.       /**
    8.        * Does nothing, which has the effect of discarding task r.
    9.        *
    10.        * @param r the runnable task requested to be executed
    11.        * @param e the executor attempting to execute this task
    12.        */
    13.       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    14.       }
    15.   }

  • DiscardOldestPolicy：放弃队列中最早的任务，本任务取而代之

    1.   public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    2.       /**
    3.        * Creates a {@code DiscardOldestPolicy} for the given executor.
    4.        */
    5.       public DiscardOldestPolicy() { }
    7.       /**
    8.        * Obtains and ignores the next task that the executor
    9.        * would otherwise execute, if one is immediately available,
    10.        * and then retries execution of task r, unless the executor
    11.        * is shut down, in which case task r is instead discarded.
    12.        *
    13.        * @param r the runnable task requested to be executed
    14.        * @param e the executor attempting to execute this task
    15.        */
    16.       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    17.           if (!e.isShutdown()) {
    18.               e.getQueue().poll();
    19.               e.execute(r);
    20.           }
    21.       }
    22.   }

  • CallerRunsPolicy：让调用者运行任务

    1.   public static class CallerRunsPolicy implements RejectedExecutionHandler {
    2.       /**
    3.        * Creates a {@code CallerRunsPolicy}.
    4.        */
    5.       public CallerRunsPolicy() { }
    7.       /**
    8.        * Executes task r in the caller's thread, unless the executor
    9.        * has been shut down, in which case the task is discarded.
    10.        *
    11.        * @param r the runnable task requested to be executed
    12.        * @param e the executor attempting to execute this task
    13.        */
    14.       public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    15.           if (!e.isShutdown()) {
    16.               r.run();
    17.           }
    18.       }
    19.   }

    简单使用

    ```java package panw.ThreadPool;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger;

@Slf4j public class ThreadPoolExecutorTest { public static void main(String[] args) { // new ThreadPoolExecutor(1,4,1, TimeUnit.SECONDS,); AtomicInteger atomic = new AtomicInteger(0);

  1.     ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 4, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3), r -> new Thread(r,"MyThread-"+atomic.getAndIncrement()), new ThreadPoolExecutor.CallerRunsPolicy());
  2.     for (int i = 0; i < 10; i++) {
  3.         threadPoolExecutor.execute(()->{
  4.             log.debug(String.valueOf(Thread.currentThread()));
  5.             try {
  6.                 Thread.sleep(2000);
  7.             } catch (InterruptedException e) {
  8.                 throw new RuntimeException(e);
  9.             }
  10.         });
  11.     }
  12. }

}

  1. <a name="Pd20a"></a>
  2. ### 4.常见四种线程池
  3. <a name="Y0vYw"></a>
  4. #### FixedThreadPool()
  5. 定长线程池：
  7. - 可控制线程最大并发数（同时执行的线程数）
  8. - 超出的线程会在队列中等待
  9. ```java
  10. //nThreads => 最大线程数即maximumPoolSize
  11. ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads);
  13. ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads, ThreadFactory threadFactory);
  15. //构造
  16.  public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
  17.         return new ThreadPoolExecutor(nThreads, nThreads,
  18.                                       0L, TimeUnit.MILLISECONDS,
  19.                                       new LinkedBlockingQueue<Runnable>(),
  20.                                       threadFactory);
  21.     }

CachedThreadPool

可缓存线程池：

  • 线程数无限制
  • 有空闲线程则复用空闲线程，若无空闲线程则新建线程

  • 一定程序减少频繁创建/销毁线程，减少系统开销

    1. //创建
    2. ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
    3. //构造
    4. public static ExecutorService newCachedThreadPool() {
    5.   return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
    6.                                 60L, TimeUnit.SECONDS,
    7.                                 new SynchronousQueue<Runnable>());
    8. }

    SingleThreadExecutor()

    单线程化的线程池：

  • 有且仅有一个工作线程执行任务

  • 所有任务按照指定顺序执行，即遵循队列的入队出队规则

    1. //创建
    2. ExecutorService singleThreadPool = Executors.newSingleThreadPool();
    3. //构造
    4. public static ExecutorService newSingleThreadExecutor() {
    5.   return new FinalizableDelegatedExecutorService
    6.       (new ThreadPoolExecutor(1, 1,
    7.                               0L, TimeUnit.MILLISECONDS,
    8.                               new LinkedBlockingQueue<Runnable>()));
    9. }

    几个注意

  • SingleThread和自己创建一个线程来运行多个任务的区别

    • 当线程正在执行的任务发生错误时，如果是自己创建的线程，该任务和剩余的任务就无法再继续运行下去。而SingleThread会创建一个新线程，继续执行任务队列中剩余的任务。

  • SingleThread和newFixedThreadPool(1)的区别

    1. // 强转为ThreadPoolExecutor
    2. ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
    3. // 改变核心线程数
    4. threadPool.setCorePoolSize(2);
    • newFixedThreadPool(1)传值为1，可以将FixedThreadPool强转为ThreadPoolExecutor，然后通过setCorePoolSize改变核心线程数
    • 而SingleThread无法修改核心线程数,因为返回的是FinalizableDelegatedExecutorService
      1. public static ExecutorService newSingleThreadExecutor() {
      2.    return new FinalizableDelegatedExecutorService
      3.        (new ThreadPoolExecutor(1, 1,
      4.                                0L, TimeUnit.MILLISECONDS,
      5.                                new LinkedBlockingQueue<Runnable>()));
      6. }

      ScheduledThreadPool()

      定时线程池：

  • 支持定时及周期性任务执行 ```java //nThreads => 最大线程数即maximumPoolSize ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(int corePoolSize);

//构造 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); }

//ScheduledThreadPoolExecutor(): public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS, new DelayedWorkQueue()); }

  1. <a name="oGkND"></a>
  2. ##### 定时功能
  3. ```java
  4. ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
  5.         //定时，以固定频率，上一个任务开始的时间计时，一个period后，检测上一个任务是否执行完毕，
  6.         // 如果上一个任务执行完毕，则当前任务立即执行，
  7.         // 如果上一个任务没有执行完毕，则需要等上一个任务执行完毕后立即执行。
  9.         pool.scheduleAtFixedRate(()->{log.debug(LocalDateTime.now().toString());},1,5, TimeUnit.SECONDS);
  10.         //1表示首次执行任务的延迟时间，5表示每次执行任务的间隔时间，TimeUnit.SECONDS执行的时间间隔数值单位
  12.         //定时，以固定的延时 delay（延时）指的是一次执行终止和下一次执行开始之间的延迟。
  13.         pool.scheduleWithFixedDelay(()->{log.debug(LocalDateTime.now().toString());},1,5,TimeUnit.SECONDS);