线程池

自定义线程池

image.png

步骤1：自定义拒绝策略接口

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

步骤2：自定义任务队列

  1. class BlockingQueue<T> {
  2.     // 1. 任务队列
  3.     private Deque<T> queue = new ArrayDeque<>();
  4.     // 2. 锁
  5.     private ReentrantLock lock = new ReentrantLock();
  6.     // 3. 生产者条件变量
  7.     private Condition fullWaitSet = lock.newCondition();
  8.     // 4. 消费者条件变量
  9.     private Condition emptyWaitSet = lock.newCondition();
  10.     // 5. 容量
  11.     private int capcity;
  12.     public BlockingQueue(int capcity) {
  13.         this.capcity = capcity;
  14.     }
  15.     // 带超时阻塞获取
  16.     public T poll(long timeout, TimeUnit unit) {
  17.         lock.lock();
  18.         try {
  19.             // 将 timeout 统一转换为 纳秒
  20.             long nanos = unit.toNanos(timeout);
  21.             while (queue.isEmpty()) {
  22.                 try {
  23.                     // 返回值是剩余时间
  24.                     if (nanos <= 0) {
  25.                         return null;
  26.                     }
  27.                     nanos = emptyWaitSet.awaitNanos(nanos);
  28.                 } catch (InterruptedException e) {
  29.                     e.printStackTrace();
  30.                 }
  31.             }
  32.             T t = queue.removeFirst();
  33.             fullWaitSet.signal();
  34.             return t;
  35.         } finally {
  36.             lock.unlock();
  37.         }
  38.     }
  39.     // 阻塞获取
  40.     public T take() {
  41.         lock.lock();
  42.         try {
  43.             while (queue.isEmpty()) {
  44.                 try {
  45.                     emptyWaitSet.await();
  46.                 } catch (InterruptedException e) {
  47.                     e.printStackTrace();
  48.                 }
  49.             }
  50.             T t = queue.removeFirst();
  51.             fullWaitSet.signal();
  52.             return t;
  53.         } finally {
  54.             lock.unlock();
  55.         }
  56.     }
  57.     // 阻塞添加
  58.     public void put(T task) {
  59.         lock.lock();
  60.         try {
  61.             while (queue.size() == capcity) {
  62.                 try {
  63.                     log.debug("等待加入任务队列 {} ...", task);
  64.                     fullWaitSet.await();
  65.                 } catch (InterruptedException e) {
  66.                     e.printStackTrace();
  67.                 }
  68.             }
  69.             log.debug("加入任务队列 {}", task);
  70.             queue.addLast(task);
  71.             emptyWaitSet.signal();
  72.         } finally {
  73.             lock.unlock();
  74.         }
  75.     }
  76.     // 带超时时间阻塞添加
  77.     public boolean offer(T task, long timeout, TimeUnit timeUnit) {
  78.         lock.lock();
  79.         try {
  80.             long nanos = timeUnit.toNanos(timeout);
  81.             while (queue.size() == capcity) {
  82.                 try {
  83.                     if(nanos <= 0) {
  84.                         return false;
  85.                     }
  86.                     log.debug("等待加入任务队列 {} ...", task);
  87.                     nanos = fullWaitSet.awaitNanos(nanos);
  88.                 } catch (InterruptedException e) {
  89.                     e.printStackTrace();
  90.                 }
  91.             }
  92.             log.debug("加入任务队列 {}", task);
  93.             queue.addLast(task);
  94.             emptyWaitSet.signal();
  95.             return true;
  96.         } finally {
  97.             lock.unlock();
  98.         }
  99.     }
  100.     public int size() {
  101.         lock.lock();
  102.         try {
  103.             return queue.size();
  104.         } finally {
  105.             lock.unlock();
  106.         }
  107.     }
  108.     public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
  109.         lock.lock();
  110.         try {
  111.             // 判断队列是否满
  112.             if(queue.size() == capcity) {
  113.                 rejectPolicy.reject(this, task);
  114.             } else { // 有空闲
  115.                 log.debug("加入任务队列 {}", task);
  116.                 queue.addLast(task);
  117.                 emptyWaitSet.signal();
  118.             }
  119.         } finally {
  120.             lock.unlock();
  121.         }
  122.     }
  123. }

步骤3：自定义线程池

  1. class ThreadPool {
  2.     // 任务队列
  3.     private BlockingQueue<Runnable> taskQueue;
  4.     // 线程集合
  5.     private HashSet<Worker> workers = new HashSet<>();
  6.     // 核心线程数
  7.     private int coreSize;
  8.     // 获取任务时的超时时间
  9.     private long timeout;
  10.     private TimeUnit timeUnit;
  11.     private RejectPolicy<Runnable> rejectPolicy;
  13.     // 执行任务
  14.     public void execute(Runnable task) {
  15.         // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
  16.         // 如果任务数超过 coreSize 时，加入任务队列暂存
  17.         synchronized (workers) {
  18.             if (workers.size() < coreSize) {
  19.                 Worker worker = new Worker(task);
  20.                 log.debug("新增 worker{}, {}", worker, task);
  21.                 workers.add(worker);
  22.                 worker.start();
  23.             } else {
  24. // taskQueue.put(task);
  25.                 // 1) 死等
  26.                 // 2) 带超时等待
  27.                 // 3) 让调用者放弃任务执行
  28.                 // 4) 让调用者抛出异常
  29.                 // 5) 让调用者自己执行任务
  30.                 taskQueue.tryPut(rejectPolicy, task);
  31.             }
  32.         }
  33.     }
  35.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,
  36.                       RejectPolicy<Runnable> rejectPolicy) {
  37.         this.coreSize = coreSize;
  38.         this.timeout = timeout;
  39.         this.timeUnit = timeUnit;
  40.         this.taskQueue = new BlockingQueue<>(queueCapcity);
  41.         this.rejectPolicy = rejectPolicy;
  42.     }
  44.     class Worker extends Thread {
  45.         private Runnable task;
  47.         public Worker(Runnable task) {
  48.             this.task = task;
  49.         }
  51.         @Override
  52.         public void run() {
  53.             // 执行任务
  54.             // 1) 当 task 不为空，执行任务
  55.             // 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
  56. // while(task != null || (task = taskQueue.take()) != null) {
  57.             while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
  58.                 try {
  59.                     log.debug("正在执行...{}", task);
  60.                     task.run();
  61.                 } catch (Exception e) {
  62.                     e.printStackTrace();
  63.                 } finally {
  64.                     task = null;
  65.                 }
  66.             }
  67.             synchronized (workers) {
  68.                 log.debug("worker 被移除{}", this);
  69.                 workers.remove(this);
  70.             }
  71.         }
  72.     }
  73. }

步骤四：测试

  1. public static void main(String[] args) {
  2.     ThreadPool threadPool = new ThreadPool(1,
  3.                                            1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{
  4.                                                // 1. 死等
  5.                                                // queue.put(task);
  6.                                                // 2) 带超时等待
  7.                                                // queue.offer(task, 1500, TimeUnit.MILLISECONDS);
  8.                                                // 3) 让调用者放弃任务执行
  9.                                                // log.debug("放弃{}", task);
  10.                                                // 4) 让调用者抛出异常
  11.                                                // throw new RuntimeException("任务执行失败 " + task);
  12.                                                // 5) 让调用者自己执行任务
  13.                                                task.run();
  14.                                            });
  15.     for (int i = 0; i < 4; i++) {
  16.         int j = i;
  17.         threadPool.execute(() -> {
  18.             try {
  19.                 Thread.sleep(1000L);
  20.             } catch (InterruptedException e) {
  21.                 e.printStackTrace();
  22.             }
  23.             log.debug("{}", j);
  24.         });
  25.     }
  26. }

自己实现的线程池

  1. @Slf4j(topic = "c.PoolTest1")
  2. public class PoolTest1 {
  3.     public static void main(String[] args) {
  4.         Pool pool = new Pool(2, 1000, TimeUnit.MILLISECONDS, 1, (queue, task) -> {
  5.             // 死等
  6.             // queue.push(task);
  8.             // 带超时等待
  9.             // queue.offer(task, 2000,TimeUnit.MILLISECONDS);
  11.             // 让调用者放弃任务执行
  12.             // log.info("啥都不写，就是放弃{}!!!", task);
  14.             // 让调用者抛出异常
  15.             // throw new RuntimeException("抛出异常，后面的都不能运行");
  17.             // 让调用者自己操作
  18.             task.run();
  19.         });
  20.         for (int i = 0; i < 4; i++) {
  21.             int j = i;
  22.             pool.execute(() -> {
  23.                 try {
  24.                     Thread.sleep(1000);
  25.                 } catch (InterruptedException e) {
  26.                     e.printStackTrace();
  27.                 }
  28.                 log.info("{}", j);
  29.             });
  30.         }
  31.     }
  33. }
  35. @Slf4j(topic = "c.Pool")
  36. class Pool {
  38.     // 任务队列
  39.     private BlockingQueues<Runnable> taskQueue;
  41.     // 线程集合
  42.     private HashSet<Work> workers = new HashSet<>();
  44.     // 核心线程数
  45.     private int coreSize;
  47.     // 获取任务的超时时间
  48.     private long timeout;
  50.     // 队列的容量
  51.     private TimeUnit timeUnit;
  53.     // 拒绝策略的接口
  54.     private RejectPolicy<Runnable> rejectPolicy;
  56.     public Pool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapicity, RejectPolicy<Runnable> rejectPolicy) {
  57.         this.taskQueue = new BlockingQueues<>(queueCapicity);
  58.         this.coreSize = coreSize;
  59.         this.timeout = timeout;
  60.         this.timeUnit = timeUnit;
  61.         this.rejectPolicy = rejectPolicy;
  62.     }
  64.     // 执行任务
  65.     public void execute(Runnable task) {
  66.         // 锁住workers
  67.         synchronized (workers) {
  68.             if (workers.size() < coreSize) {
  69.                 // 创建新的工作线程
  70.                 Work work = new Work(task);
  71.                 log.info("新增worker:{}", work);
  72.                 // 放入工作线程集合
  73.                 workers.add(work);
  74.                 work.start();
  75.             } else {
  76.                 // 如果工作线程不够用，放入阻塞队列中
  77.                 // 死等
  78.                 // taskQueue.push(task);
  79.                 taskQueue.tryPut(rejectPolicy, task);
  80.             }
  81.         }
  82.     }
  84.     class Work extends Thread {
  86.         // 任务
  87.         private Runnable task;
  89.         public Work(Runnable task) {
  90.             this.task = task;
  91.         }
  93.         @Override
  94.         public void run() {
  95.             // 执行任务
  96.             // 1.如果任务不为空，直接执行
  97.             // 2.如果任务为空，那么从任务队列中获取
  98.             // while (task != null || (task = taskQueue.take()) != null) {  // 如果没加超时时间，线程会一直等待任务
  99.             while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
  100.                 try {
  101.                     log.info("正在执行...{}", task);
  102.                     task.run();
  103.                 } catch (Exception exception) {
  104.                     exception.printStackTrace();
  105.                 } finally {
  106.                     // 执行完毕，任务为空
  107.                     task = null;
  108.                 }
  109.             }
  110.             synchronized (workers) {
  111.                 log.info("一直没任务，移除线程:{}", this);
  112.                 workers.remove(this);
  113.             }
  114.         }
  115.     }
  116. }
  118. /**
  119.  * 阻塞队列
  120.  *
  121.  * @param <T>
  122.  */
  123. @Slf4j(topic = "c.BlockingQueues")
  124. class BlockingQueues<T> {
  125.     // 线程队列
  126.     private Deque<T> quque = new ArrayDeque<>();
  128.     // 锁
  129.     private ReentrantLock lock = new ReentrantLock();
  131.     // 生产者条件变量
  132.     private Condition proWaitSet = lock.newCondition();
  134.     // 消费者条件变量
  135.     private Condition conWaitSet = lock.newCondition();
  137.     // 容量
  138.     private int capicity;
  140.     public BlockingQueues(int capicity) {
  141.         this.capicity = capicity;
  142.     }
  144.     // 带超时的阻塞获取
  145.     public T poll(long waitTime, TimeUnit timeUnit) {
  146.         // 先加锁
  147.         lock.lock();
  148.         try {
  149.             // 将time统一转化为纳秒
  150.             long nanos = timeUnit.toNanos(waitTime);
  151.             while (quque.isEmpty()) {
  152.                 try {
  153.                     // 等待完毕
  154.                     if (nanos <= 0) {
  155.                         return null;
  156.                     }
  158.                     // 没有就等待
  159.                     // 此时重新赋值是为了防止虚假唤醒，唤醒之后，时间就变少了
  160.                     // awaitNanos方法会返回剩余时间
  161.                     nanos = conWaitSet.awaitNanos(nanos);
  162.                 } catch (InterruptedException e) {
  163.                     e.printStackTrace();
  164.                 }
  165.             }
  167.             return quque.poll();
  168.         } finally {
  169.             proWaitSet.signal();
  170.             lock.unlock();
  171.         }
  172.     }
  174.     // 阻塞获取
  175.     public T take() {
  176.         // 先加锁
  177.         lock.lock();
  178.         try {
  179.             while (quque.isEmpty()) {
  180.                 try {
  181.                     // 没有就等待
  182.                     conWaitSet.await();
  183.                 } catch (InterruptedException e) {
  184.                     e.printStackTrace();
  185.                 }
  186.             }
  188.             return quque.poll();
  189.         } finally {
  190.             proWaitSet.signal();
  191.             lock.unlock();
  192.         }
  193.     }
  195.     public void push(T t) {
  196.         // 先加锁
  197.         try {
  198.             lock.lock();
  199.             while (quque.size() == capicity) {
  200.                 try {
  201.                     log.info("任务队列已满，等待加入:{}", t);
  202.                     proWaitSet.await();
  203.                 } catch (InterruptedException e) {
  204.                     e.printStackTrace();
  205.                 }
  206.             }
  207.             quque.addLast(t);
  208.             log.info("加入任务队列:{}", t);
  209.             conWaitSet.signal();
  210.         } finally {
  211.             lock.unlock();
  212.         }
  213.     }
  215.     /**
  216.      * 带超时时间的阻塞添加
  217.      *
  218.      * @param task
  219.      * @param timeout
  220.      * @param timeUnit
  221.      * @return
  222.      */
  223.     public boolean offer(T task, int timeout, TimeUnit timeUnit) {
  224.         // 先加锁
  225.         try {
  226.             lock.lock();
  227.             long nanos = timeUnit.toNanos(timeout);
  228.             while (quque.size() == capicity) {
  229.                 try {
  230.                     if (nanos <= 0) {
  231.                         return false;
  232.                     }
  233.                     log.info("任务队列已满，等待加入:{}", task);
  234.                     nanos = proWaitSet.awaitNanos(nanos);
  235.                 } catch (InterruptedException e) {
  236.                     e.printStackTrace();
  237.                 }
  238.             }
  239.             quque.addLast(task);
  240.             log.info("加入任务队列:{}", task);
  241.             conWaitSet.signal();
  242.             return true;
  243.         } finally {
  244.             lock.unlock();
  245.         }
  246.     }
  248.     /**
  249.      * 带有拒接策略的放入任务
  250.      *
  251.      * @param rejectPolicy
  252.      * @param task
  253.      */
  254.     public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
  255.         try {
  256.             lock.lock();
  257.             // 队列已满
  258.             if (quque.size() == capicity) {
  259.                 // 调用操作者实现的操作
  260.                 rejectPolicy.reject(this, task);
  261.             } else {
  262.                 // 还有空闲
  263.                 quque.addLast(task);
  264.                 log.info("加入任务队列:{}", task);
  265.                 conWaitSet.signal();
  266.             }
  267.         } finally {
  268.             lock.unlock();
  269.         }
  270.     }
  272.     public int size() {
  273.         lock.lock();
  274.         try {
  275.             return quque.size();
  276.         } finally {
  277.             lock.unlock();
  278.         }
  279.     }
  281. }
  283. /**
  284.  * 拒接策略
  285.  *
  286.  * @param <T>
  287.  */
  288. @FunctionalInterface
  289. interface RejectPolicy<T> {
  290.     void reject(BlockingQueues<T> queues, T task);
  291. }

ThreadPoolExecutor

image.png

线程池状态

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量

image.png

从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING（因为是高三位，第一位是符号所以RUNNING是负数）

这些信息存储在一个原子变量 ctl 中，目的是将线程池状态与线程个数合二为一，这样就可以用一次 cas 原子操作进行赋值。

  1. // c 为旧值， ctlOf 返回结果为新值
  2. ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
  4. // rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
  5. private static int ctlOf(int rs, int wc) { return rs | wc; }

构造方法

  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 最大线程数目
  • keepAliveTime 生存时间 - 针对救急线程
  • unit 时间单位 - 针对救急线程
  • workQueue 阻塞队列
  • threadFactory 线程工厂 - 可以为线程创建时起个好名字
  • handler 拒绝策略

工作方式：

image.png

  • 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
  • 当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
  • 如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急。
  • 如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现
    • AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
    • CallerRunsPolicy 让调用者运行任务
    • DiscardPolicy 放弃本次任务
    • DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
    • Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方
    • 便定位问题
    • Netty 的实现，是创建一个新线程来执行任务
    • ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
    • PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
  • 当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

image.png

根据这个构造方法，JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池。

newFixedThreadPool

  1. public static ExecutorService newFixedThreadPool(int nThreads) {
  2.     return new ThreadPoolExecutor(nThreads, nThreads,
  3.             0L, TimeUnit.MILLISECONDS,
  4.             new LinkedBlockingQueue<Runnable>());
  5. }

特点

  • 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着
    • 全部都是救急线程（60s 后可以回收）
    • 救急线程可以无限创建
  • 队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱、一手交货） 
  1. SynchronousQueue<Integer> integers = new SynchronousQueue<>();
  2. new Thread(() -> {
  3.     try {
  4.         log.debug("putting {} ", 1);
  5.         integers.put(1);
  6.         log.debug("{} putted...", 1);
  7.         log.debug("putting...{} ", 2);
  8.         integers.put(2);
  9.         log.debug("{} putted...", 2);
  10.     } catch (InterruptedException e) {
  11.         e.printStackTrace();
  12.     }
  13. },"t1").start();
  14. sleep(1);
  15. new Thread(() -> {
  16.     try {
  17.         log.debug("taking {}", 1);
  18.         integers.take();
  19.     } catch (InterruptedException e) {
  20.         e.printStackTrace();
  21.     }
  22. },"t2").start();
  23. sleep(1);
  24. new Thread(() -> {
  25.     try {
  26.         log.debug("taking {}", 2);
  27.         integers.take();
  28.     } catch (InterruptedException e) {
  29.         e.printStackTrace();
  30.     }
  31. },"t3").start();

输出

  1. 11:48:15.500 c.TestSynchronousQueue [t1] - putting 1 
  2. 11:48:16.500 c.TestSynchronousQueue [t2] - taking 1 
  3. 11:48:16.500 c.TestSynchronousQueue [t1] - 1 putted... 
  4. 11:48:16.500 c.TestSynchronousQueue [t1] - putting...2 
  5. 11:48:17.502 c.TestSynchronousQueue [t3] - taking 2 
  6. 11:48:17.503 c.TestSynchronousQueue [t1] - 2 putted...

评价 整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况。

newSingleThreadExecutor

使用场景：

希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程

也不会被释放。

区别：

  • 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作
  • Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改
    • FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
  • Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改
    • 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

提交任务

  1. // 执行任务
  2. void execute(Runnable command);
  4. // 提交任务 task，用返回值 Future 获得任务执行结果
  5. <T> Future<T> submit(Callable<T> task);
  7. // 提交 tasks 中所有任务
  8. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
  9.  throws InterruptedException;
  11. // 提交 tasks 中所有任务，带超时时间
  12. <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
  13.         long timeout, TimeUnit unit)
  14.  throws InterruptedException;
  16. // 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
  17. <T> T invokeAny(Collection<? extends Callable<T>> tasks)
  18.  throws InterruptedException, ExecutionException;
  20. // 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
  21. <T> T invokeAny(Collection<? extends Callable<T>> tasks,
  22.         long timeout, TimeUnit unit)
  23.  throws InterruptedException, ExecutionException, TimeoutException;

关闭线程

shutdown

  1. /*
  2. 线程池状态变为 SHUTDOWN
  3. - 不会接收新任务
  4. - 但已提交任务会执行完
  5. - 此方法不会阻塞调用线程的执行
  6. */
  7. void shutdown();
  1. public void shutdown() {
  2.     final ReentrantLock mainLock = this.mainLock;
  3.     mainLock.lock();
  4.     try {
  5.         checkShutdownAccess();
  6.         // 修改线程池状态
  7.         advanceRunState(SHUTDOWN);
  8.         // 仅会打断空闲线程
  9.         interruptIdleWorkers();
  10.         onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
  11.     } finally {
  12.         mainLock.unlock();
  13.     }
  14.     // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
  15.     tryTerminate();
  16. }

shutdownNow

  1. /*
  2. 线程池状态变为 STOP
  3. - 不会接收新任务
  4. - 会将队列中的任务返回
  5. - 并用 interrupt 的方式中断正在执行的任务
  6. */
  7. List<Runnable> shutdownNow();
  1. public List<Runnable> shutdownNow() {
  2.     List<Runnable> tasks;
  3.     final ReentrantLock mainLock = this.mainLock;
  4.     mainLock.lock();
  5.     try {
  6.         checkShutdownAccess();
  7.         // 修改线程池状态
  8.         advanceRunState(STOP);
  9.         // 打断所有线程
  10.         interruptWorkers();
  11.         // 获取队列中剩余任务
  12.         tasks = drainQueue();
  13.     } finally {
  14.         mainLock.unlock();
  15.     }
  16.     // 尝试终结
  17.     tryTerminate();
  18.     return tasks; 
  19. }

其它方法

  1. // 不在 RUNNING 状态的线程池，此方法就返回 true
  2. boolean isShutdown();
  4. // 线程池状态是否是 TERMINATED
  5. boolean isTerminated();
  7. // 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待
  8. boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

模式之 Worker Thread

任务调度线程池

在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。

  1. public static void main(String[] args) {
  2.     Timer timer = new Timer();
  3.     TimerTask task1 = new TimerTask() {
  4.         @Override
  5.         public void run() {
  6.             log.debug("task 1");
  7.             sleep(2);
  8.         }
  9.     };
  10.     TimerTask task2 = new TimerTask() {
  11.         @Override
  12.         public void run() {
  13.             log.debug("task 2");
  14.         }
  15.     };
  16.     // 使用 timer 添加两个任务，希望它们都在 1s 后执行
  17.     // 但由于 timer 内只有一个线程来顺序执行队列中的任务，因此『任务1』的延时，影响了『任务2』的执行
  18.     timer.schedule(task1, 1000);
  19.     timer.schedule(task2, 1000);
  20. }

输出

  1. 20:46:09.444 c.TestTimer [main] - start... 
  2. 20:46:10.447 c.TestTimer [Timer-0] - task 1 
  3. 20:46:12.448 c.TestTimer [Timer-0] - task 2

使用 ScheduledExecutorService 改写：

  1. ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
  2. // 添加两个任务，希望它们都在 1s 后执行
  3. executor.schedule(() -> {
  4.     System.out.println("任务1，执行时间：" + new Date());
  5.     try { Thread.sleep(2000); } catch (InterruptedException e) { }
  6. }, 1000, TimeUnit.MILLISECONDS);
  7. executor.schedule(() -> {
  8.     System.out.println("任务2，执行时间：" + new Date());
  9. }, 1000, TimeUnit.MILLISECONDS);

输出

  1. 任务1，执行时间：Thu Jan 03 12:45:17 CST 2019 
  2. 任务2，执行时间：Thu Jan 03 12:45:17 CST 2019

scheduleAtFixedRate 例子：

  1. ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
  2.     log.debug("start...");
  3. pool.scheduleAtFixedRate(() -> {
  4.      log.debug("running...");
  5. }, 1, 1, TimeUnit.SECONDS);

输出

  1. 21:45:43.167 c.TestTimer [main] - start... 
  2. 21:45:44.215 c.TestTimer [pool-1-thread-1] - running... 
  3. 21:45:45.215 c.TestTimer [pool-1-thread-1] - running... 
  4. 21:45:46.215 c.TestTimer [pool-1-thread-1] - running... 
  5. 21:45:47.215 c.TestTimer [pool-1-thread-1] - running...

scheduleAtFixedRate 例子（任务执行时间超过了间隔时间）：

  1. ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
  2. log.debug("start...");
  3. pool.scheduleAtFixedRate(() -> {
  4.      log.debug("running...");
  5.      sleep(2);
  6. }, 1, 1, TimeUnit.SECONDS);

输出分析：一开始，延时 1s，接下来，由于任务执行时间 > 间隔时间，间隔被『撑』到了 2s

  1. 21:44:30.311 c.TestTimer [main] - start... 
  2. 21:44:31.360 c.TestTimer [pool-1-thread-1] - running... 
  3. 21:44:33.361 c.TestTimer [pool-1-thread-1] - running... 
  4. 21:44:35.362 c.TestTimer [pool-1-thread-1] - running... 
  5. 21:44:37.362 c.TestTimer [pool-1-thread-1] - running

scheduleWithFixedDelay 例子：

  1. ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
  2. log.debug("start...");
  3. pool.scheduleWithFixedDelay(()-> {
  4.      log.debug("running...");
  5.      sleep(2);
  6. }, 1, 1, TimeUnit.SECONDS);

输出分析：一开始，延时 1s，scheduleWithFixedDelay 的间隔是 上一个任务结束 <-> 延时 <-> 下一个任务开始 所以间隔都是 3s

  1. 21:40:55.078 c.TestTimer [main] - start... 
  2. 21:40:56.140 c.TestTimer [pool-1-thread-1] - running... 
  3. 21:40:59.143 c.TestTimer [pool-1-thread-1] - running... 
  4. 21:41:02.145 c.TestTimer [pool-1-thread-1] - running... 
  5. 21:41:05.147 c.TestTimer [pool-1-thread-1] - running...

评价 整个线程池表现为：线程数固定，任务数多于线程数时，会放入无界队列排队。任务执行完毕，这些线程也不会被释放。用来执行延迟或反复执行的任务。

正确处理执行任务异常

方法1：主动捉异常

  1. pool.submit(() -> {
  2.     try {
  3.         log.debug("task1");
  4.         int i = 1 / 0;
  5.     } catch (Exception e) {
  6.         log.error("error:", e);
  7.     }
  8. });

输出

  1. 21:59:04.558 c.TestTimer [pool-1-thread-1] - task1 
  2. 21:59:04.562 c.TestTimer [pool-1-thread-1] - error: 
  3. java.lang.ArithmeticException: / by zero 
  4.      at cn.itcast.n8.TestTimer.lambda$main$0(TestTimer.java:28) 
  5.      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 
  6.      at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
  7.      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 
  8.      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 
  9.      at java.lang.Thread.run(Thread.java:748)

方法2：使用 Future

  1. ExecutorService pool = Executors.newFixedThreadPool(1);
  2. ExecutorService pool = Executors.newFixedThreadPool(1);
  3. Future<Boolean> f = pool.submit(() -> {
  4.     log.debug("task1");
  5.     int i = 1 / 0;
  6.     return true;
  7. });
  8. log.debug("result:{}", f.get());

输出

  1. 21:54:58.208 c.TestTimer [pool-1-thread-1] - task1 
  2. Exception in thread "main" java.util.concurrent.ExecutionException: 
  3. java.lang.ArithmeticException: / by zero 
  4.      at java.util.concurrent.FutureTask.report(FutureTask.java:122) 
  5.      at java.util.concurrent.FutureTask.get(FutureTask.java:192) 
  6.      at cn.itcast.n8.TestTimer.main(TestTimer.java:31) 
  7. Caused by: java.lang.ArithmeticException: / by zero 
  8.      at cn.itcast.n8.TestTimer.lambda$main$0(TestTimer.java:28) 
  9.      at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
  10.      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 
  11.      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) 
  12.      at java.lang.Thread.run(Thread.java:748)

应用之定时任务

如何让每周四 18:00:00 定时执行任务？

  1. public class TestSchedule {
  2.     // 如何每周四18:00:00定时执行任务？
  3.     public static void main(String[] args) {
  4.         // 获取当前时间
  5.         LocalDateTime now = LocalDateTime.now();
  6.         // 获取周四时间
  7.         LocalDateTime time = now.withHour(18).withMinute(0).withSecond(0).withNano(0).with(DayOfWeek.THURSDAY);
  8.         // 如果当前时间大于本周周四，必须找到下周四
  9.         if (now.compareTo(time) > 0) {
  10.             time = time.plusWeeks(1);
  11.         }
  13.         // initailDelay 代表当前时间和周四的时间差
  14.         // period 一周间隔的时间
  15.         long initailDelay = Duration.between(now, time).toMillis();
  16.         long period = 1000 * 60 * 60 * 24 * 7;
  17.         ScheduledThreadPoolExecutor pool = new ScheduledThreadPoolExecutor(1);
  18.         pool.scheduleAtFixedRate(() -> {
  19.             System.out.println("running ... ");
  20.         }, initailDelay, period, TimeUnit.SECONDS);
  22.     }
  23. }

Tomcat线程池

Tomcat 在哪里用到了线程池呢

  • LimitLatch 用来限流，可以控制最大连接个数，类似 J.U.C 中的 Semaphore 后面再讲
  • Acceptor 只负责【接收新的 socket 连接】
  • Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】
  • 一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
  • Executor 线程池中的工作线程最终负责【处理请求】

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同

  • 如果总线程数达到 maximumPoolSize
    • 这时不会立刻抛 RejectedExecutionException 异常
    • 而是再次尝试将任务放入队列，如果还失败，才抛出 RejectedExecutionException 异常

tomcat7的源码

  1.     submittedCount.incrementAndGet();
  2.     try {
  3.         super.execute(command);
  4.     } catch (RejectedExecutionException rx) {
  5.         if (super.getQueue() instanceof TaskQueue) {
  6.             final TaskQueue queue = (TaskQueue)super.getQueue();
  7.             try {
  8.                 if (!queue.force(command, timeout, unit)) {
  9.                     submittedCount.decrementAndGet();
  10.                     throw new RejectedExecutionException("Queue capacity is full.");
  11.                 }
  12.             } catch (InterruptedException x) {
  13.                 submittedCount.decrementAndGet();
  14.                 Thread.interrupted();
  15.                 throw new RejectedExecutionException(x);
  16.             }
  17.         } else {
  18.             submittedCount.decrementAndGet();
  19.             throw rx;
  20.         }
  21.     }
  22. }

TaskQueue.java

  1. public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
  2.     if ( parent.isShutdown() )
  3.         throw new RejectedExecutionException(
  4.                 "Executor not running, can't force a command into the queue"
  5.         );
  6.     return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task 
  7.     is rejected
  8. }

Connector 配置

image.png

Executor 线程配置

image.png

image.png

Fork/Join

概念

Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 cpu 密集型运算

所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解

Fork/Join 在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率

Fork/Join 默认会创建与 cpu 核心数大小相同的线程池

使用

提交给 Fork/Join 线程池的任务需要继承 RecursiveTask（有返回值）或 RecursiveAction（没有返回值），例如下面定义了一个对 1~n 之间的整数求和的任务。

  1. class AddTask1 extends RecursiveTask<Integer> {
  2.     int n;
  3.     public AddTask1(int n) {
  4.         this.n = n;
  5.     }
  6.     @Override
  7.     public String toString() {
  8.         return "{" + n + '}';
  9.     }
  10.     @Override
  11.     protected Integer compute() {
  12.         // 如果 n 已经为 1，可以求得结果了
  13.         if (n == 1) {
  14.             log.debug("join() {}", n);
  15.             return n;
  16.         }
  18.         // 将任务进行拆分(fork)
  19.         AddTask1 t1 = new AddTask1(n - 1);
  20.         t1.fork();
  21.         log.debug("fork() {} + {}", n, t1);
  23.         // 合并(join)结果
  24.         int result = n + t1.join();
  25.         log.debug("join() {} + {} = {}", n, t1, result);
  26.         return result;
  27.     }
  28. }

然后提交给 ForkJoinPool 来执行

  1. public static void main(String[] args) {
  2.      ForkJoinPool pool = new ForkJoinPool(4);
  3.      System.out.println(pool.invoke(new AddTask3(1, 10)));
  4. }

结果

  1. [ForkJoinPool-1-worker-0] - fork() 2 + {1} 
  2. [ForkJoinPool-1-worker-1] - fork() 5 + {4} 
  3. [ForkJoinPool-1-worker-0] - join() 1 
  4. [ForkJoinPool-1-worker-0] - join() 2 + {1} = 3 
  5. [ForkJoinPool-1-worker-2] - fork() 4 + {3} 
  6. [ForkJoinPool-1-worker-3] - fork() 3 + {2} 
  7. [ForkJoinPool-1-worker-3] - join() 3 + {2} = 6 
  8. [ForkJoinPool-1-worker-2] - join() 4 + {3} = 10 
  9. [ForkJoinPool-1-worker-1] - join() 5 + {4} = 15 
  10. 15

image.png

改进

  1. class AddTask3 extends RecursiveTask<Integer> {
  3.     int begin;
  4.     int end;
  5.     public AddTask3(int begin, int end) {
  6.         this.begin = begin;
  7.         this.end = end;
  8.     }
  9.     @Override
  10.     public String toString() {
  11.         return "{" + begin + "," + end + '}';
  12.     }
  13.     @Override
  14.     protected Integer compute() {
  15.         // 5, 5
  16.         if (begin == end) {
  17.             log.debug("join() {}", begin);
  18.             return begin;
  19.         }
  20.         // 4, 5
  21.         if (end - begin == 1) {
  22.             log.debug("join() {} + {} = {}", begin, end, end + begin);
  23.             return end + begin;
  24.         }
  26.         // 1 5
  27.         int mid = (end + begin) / 2; // 3
  28.         AddTask3 t1 = new AddTask3(begin, mid); // 1,3
  29.         t1.fork();
  30.         AddTask3 t2 = new AddTask3(mid + 1, end); // 4,5
  31.         t2.fork();
  32.         log.debug("fork() {} + {} = ?", t1, t2);
  33.         int result = t1.join() + t2.join();
  34.         log.debug("join() {} + {} = {}", t1, t2, result);
  35.         return result;
  36.     }
  37. }

结果

  1. [ForkJoinPool-1-worker-0] - join() 1 + 2 = 3 
  2. [ForkJoinPool-1-worker-3] - join() 4 + 5 = 9 
  3. [ForkJoinPool-1-worker-0] - join() 3 
  4. [ForkJoinPool-1-worker-1] - fork() {1,3} + {4,5} = ? 
  5. [ForkJoinPool-1-worker-2] - fork() {1,2} + {3,3} = ? 
  6. [ForkJoinPool-1-worker-2] - join() {1,2} + {3,3} = 6 
  7. [ForkJoinPool-1-worker-1] - join() {1,3} + {4,5} = 15 
  8. 15

image.png