并发编程（十六）：线程池

小鱼儿 2023-06-04 11:50 178阅读 0赞

一，ThreadPoolExecutor 概述

1，线程池优势

  1. 在Java中，如果每个请求到达就创建一个线程，创建线程和销毁线程对系统资源的消耗都非常大，甚至可能比实际业务处理消耗的资源都大。同时，如果在JVM中创建太多的线程，也可能由于过度消耗内存或调度切换从而导致系统资源不足。
  3. 为了解决上面提出的问题，就有了线程池的概念。线程池，就是在一个线程容器中提前放置一定量的初始化线程，如果业务需要创建线程进行业务处理，则直接从线程池中获取一个线程进行执行，执行完成后归还线程到线程池，同时线程池也会对创建线程进行限制，不会无休止的创建下去。

使用线程池后，可以后下面几点优势：

* 减低创建线程和销毁线程的开销

* 提高响应速度，当有新任务需要执行时不需要等待线程创建时就可以直接执行（如果有空闲线程）

* 合理的设置线程池的大小可以避免因为硬件瓶颈带来的性能问题

2，类图

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70

3，线程池常用API

  1. // 线程池初始化
  2. public ThreadPoolExecutor(
  3.     // 核心线程数
  4.     int corePoolSize,
  5.     // 最大线程数
  6.     int maximumPoolSize,
  7.     // 线程空闲保留时间
  8.     long keepAliveTime,
  9.     // 线程保留单位
  10.     TimeUnit unit,
  11.     // 线程任务阻塞容器
  12.     BlockingQueue<Runnable> workQueue,
  13.     // 线程工厂，一般取默认
  14.     ThreadFactory threadFactory,
  15.     // 拒绝策略
  16.     RejectedExecutionHandler handler);
  18. // 线程执行，不带返回值
  19. public void execute(Runnable command);
  21. // 线程执行，带返回值
  22. public <T> Future<T> submit(Callable<T> task);
  23. public <T> Future<T> submit(Runnable task, T result);
  24. public Future<?> submit(Runnable task);
  26. // 关闭线程池
  27. public void shutdown();

4，线程池常量解析，分为线程池常量和Future常量两部分

  1. /********************* 线程池常量 **********************/
  2. // 数量为，该值是29
  3. // 线程池把 Integer 的32位拆分为高3位和低29位，
  4. // 通过高三位表示状态，低29位表示正在执行的线程数量，
  5. private static final int COUNT_BITS = Integer.SIZE - 3;
  7. // 最大允许执行的线程数量，因为高三位表示状态，所以天然限制为29位
  8. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
  10. // 线程池运行状态
  11. // -1 的二进制是 11111111 11111111 11111111 11111111
  12. // 右移29位就是  11100000 00000000 00000000 00000000
  13. // 高三位表示状态，所以 RUNNING 的状态就是 111
  14. private static final int RUNNING    = -1 << COUNT_BITS;
  16. // 线程池关闭状态，不接收新任务，但是执行队列中的人物
  17. // 状态为 0
  18. private static final int SHUTDOWN   =  0 << COUNT_BITS;
  20. // 线程池停止状态，不接受新任务，不执行队列人物，终止执行任务
  21. // 状态为 1
  22. private static final int STOP       =  1 << COUNT_BITS;
  24. // 所有任务都已经结束，线程数量为0，处于该状态的线程池即将调用 terminated()方法
  25. // 状态为 10
  26. private static final int TIDYING    =  2 << COUNT_BITS;
  28. // terminated()方法执行完成
  29. // 状态为 11
  30. private static final int TERMINATED =  3 << COUNT_BITS;
  32. /********************* Future常量 **********************/
  33. // Future 常量主要是对 state 状态的几种情况分析
  34. // NEW 新建状态，表示这个 FutureTask还没有开始运行
  35. private static final int NEW = 0; 
  37. // COMPLETING 完成状态，表示 FutureTask 任务已经计算完毕了
  38. // 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
  39. private static final int COMPLETING = 1;
  41. // FutureTask 任务完结，正常完成，没有发生异常
  42. private static final int NORMAL = 2;
  44. // FutureTask 任务完结，因为发生异常。
  45. private static final int EXCEPTIONAL = 3;
  47. // FutureTask 任务完结，因为取消任务
  48. private static final int CANCELLED = 4;
  50. // FutureTask 任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
  51. private static final int INTERRUPTING = 5;
  53. // FutureTask 任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
  54. private static final int INTERRUPTED = 6;

5，JDK提供的几种常用线程池，阿里开发手册不提倡用内置的线程池，建议通过构造器自行初始化，后续自行构造的执行流程会分析，该部分参考即可；跟源码可以发下，各种初始化方式最终也是通过构造器初始化！

  1. // 初始化定长线程池
  2. // 内置指定的核心线程数和最大线程数，并按照线程池流程执行
  3. public static ExecutorService newFixedThreadPool(int nThreads);
  5. // 初始化缓存的线程池
  6. // 没有核心线程数，默认最大线程数为 Integer.MAX_VALUE，阻塞队列为 SynchronousQueue，不保存数据
  7. // 所有对于缓存的线程池来说，接收一个任务的同时就需要执行一个任务
  8. public static ExecutorService newCachedThreadPool();
  10. // 初始化单例的线程池
  11. // 一次最多执行一个线程任务，多线程竞争时，添加到阻塞队列等候
  12. public static ExecutorService newSingleThreadExecutor();
  14. // 初始化定时的线程池
  15. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);

6，功能DEMO

  1. package com.gupao.concurrent;
  3. import java.util.concurrent.*;
  5. /**
  6.  * @author pj_zhang
  7.  * @create 2019-10-31 21:56
  8.  **/
  9. public class ThreadPoolTest {
  11.     private static ThreadPoolExecutor executor =
  12.             new ThreadPoolExecutor(20, 20,
  13.                     0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
  15.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  16.         Thread thread = new Thread(() -> {
  17.             System.out.println("THREAD 线程执行");
  18.         });
  20.         Runnable runnable = () -> {
  21.             System.out.println("RUNNABLE 线程执行");
  22.         };
  24.         Callable<String> callable = () -> {
  25.             Thread.sleep(2000);
  26.             return "Callable 线程执行";
  27.         };
  29.         executor.execute(thread);
  30.         executor.execute(runnable);
  31.         // 此处打印时间，是为了演示 Future 的阻塞获取结果
  32.         System.out.println("callable执行前：" + System.currentTimeMillis());
  33.         Future<String> future = executor.submit(callable);
  34.         String callableResult = future.get();
  35.         System.out.println("callable执行后：" + System.currentTimeMillis() + ", " + callableResult);
  36.     }
  38. }

2019103122063457.png

二，源码分析

1，底层方法分析

1.1，ctlOf(int rs, int wc)：获取线程池状态+数量的 Integer 值，

  1. private static int ctlOf(int rs, int wc) { 
  2.     return rs | wc; 
  3. }

1.2，workerCountOf(int c)：获取工作线程数量

  1. private static int workerCountOf(int c)  {
  2.     // 根据 ctlOf 获取到的值，用低29位进行与运算，获取到线程数量
  3.     return c & CAPACITY; 
  4. }

1.3，runStateOf(int c)：获取线程状态

  1. private static int runStateOf(int c)     { 
  2.     return c & ~CAPACITY; 
  3. }

2，ThreadPoolExecutor 初始化及执行流程

2.1，ThreadPoolExecutor()

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                           int maximumPoolSize,
  3.                           long keepAliveTime,
  4.                           TimeUnit unit,
  5.                           BlockingQueue<Runnable> workQueue,
  6.                           ThreadFactory threadFactory,
  7.                           RejectedExecutionHandler handler) {
  8.     if (corePoolSize < 0 ||
  9.         maximumPoolSize <= 0 ||
  10.         maximumPoolSize < corePoolSize ||
  11.         keepAliveTime < 0)
  12.         throw new IllegalArgumentException();
  13.     if (workQueue == null || threadFactory == null || handler == null)
  14.         throw new NullPointerException();
  15.     this.acc = System.getSecurityManager() == null ?
  16.             null :
  17.             AccessController.getContext();
  18.     // 核心线程数
  19.     this.corePoolSize = corePoolSize;
  20.     // 最大线程数
  21.     this.maximumPoolSize = maximumPoolSize;
  22.     // 阻塞队列
  23.     this.workQueue = workQueue;
  24.     // 保持存活时间
  25.     this.keepAliveTime = unit.toNanos(keepAliveTime);
  26.     // 线程工厂，这个直接取默认
  27.     this.threadFactory = threadFactory;
  28.     // 拒绝策略
  29.     this.handler = handler;
  30. }

2.2，线程池执行流程

  1. \* 接收到线程任务后，首先判断核心线程有没有被全部占用；没有被全部占用，随机构建一个线程执行任务
  3. \* 如果核心线程全部被占用，查看阻塞队列是否已满；如果没有满，添加到阻塞队列
  5. \* 如果阻塞队列已满，继续看最大线程数有没有被全部占用；如果存在空闲，构建线程执行任务
  7. \* 如果最大线程数已经全部占用，根据定义的拒绝策略进行拒绝操作

2.3，线程池拒绝策略

  1. \* 线程池提供了四种拒绝策略，分别是 *RejectedExecutionHandler* 接口的四种实现
  3. // java.util.concurrent.ThreadPoolExecutor.AbortPolicy
  4. // 异常处理
  5. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  6.     throw new RejectedExecutionException("Task " + r.toString() +
  7.                                          " rejected from " +
  8.                                          e.toString());
  9. }
  11. // java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy
  12. // 丢弃最前面的任务，重新添加执行
  13. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  14.     if (!e.isShutdown()) {
  15.         e.getQueue().poll();
  16.         e.execute(r);
  17.     }
  18. }
  20. // java.util.concurrent.ThreadPoolExecutor.DiscardPolicy
  21. // 什么都不做，即丢弃当前任务
  22. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  23. }
  25. // java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
  26. // 只要线程池没有关闭，则直接开线程运行，该策略建议慎用，不收控制
  27. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
  28.     if (!e.isShutdown()) {
  29.         r.run();
  30.     }
  31. }
  33. // 最后，业务可以自定义拒绝方式，只需要实现 RejectedExecutionHandler 接口，然后重写其接口方法

3，execute()

  1. \* execute(Runnable command)：刚才对线程池的大概执行流程进行了分析，该方法内可以看出代码实现
  3. public void execute(Runnable command) {
  4.     if (command == null)
  5.         throw new NullPointerException();
  6.     // 获取状态位 + 数量位的 int 对象
  7.     int c = ctl.get();
  8.     // 获取工作线程数量，首先判断核心线程数
  9.     if (workerCountOf(c) < corePoolSize) {
  10.         // 存在空闲的核心线程，进行线程执行
  11.         if (addWorker(command, true))
  12.             return;
  13.         // 如果执行失败，重新对 c 赋值，说明存在线程竞争
  14.         c = ctl.get();
  15.     }
  16.     // isRunning(c)：线程池依旧运行状态
  17.     // workQueue.offer(command)：添加到队列成功
  18.     // 次数是判断加入到阻塞队列是否成功
  19.     if (isRunning(c) && workQueue.offer(command)) {
  20.         int recheck = ctl.get();
  21.         // 如果线程池不处于运行状态，则从队列中移除当前任务
  22.         // 并执行拒绝策略
  23.         if (! isRunning(recheck) && remove(command))
  24.             reject(command);
  25.         // 如果之前的线程已经被销毁完，新建一个线程
  26.         else if (workerCountOf(recheck) == 0)
  27.             addWorker(null, false);
  28.     }
  29.     // 阻塞队列添加失败，唤醒最大线程执行，
  30.     else if (!addWorker(command, false))
  31.         // 最大线程执行失败，进行拒绝策略处理
  32.         reject(command);
  33. }
  35. \* addWorker(Runnable firstTask, boolean core)
  37. // Runnable firstTask：当前线程任务
  38. // boolean core：是否核心线程
  39. private boolean addWorker(Runnable firstTask, boolean core) {
  40.     retry:
  41.     for (;;) {
  42.         // 获取线程执行状态
  43.         int c = ctl.get();
  44.         // 这部分计算后续搞明白再填充 TODO
  45.         int rs = runStateOf(c);
  46.         // 线程池已经关闭，不再接受新任务
  47.         // SHUTDOWN 状态不接受新任务，但仍然会执行已经加入任务队列的任务
  48.         // 所以当进入 SHUTDOWN 状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的,
  49.         // 如果把这个条件取反，就表示不允许添加 worker
  50.         if (rs >= SHUTDOWN &&
  51.             ! (rs == SHUTDOWN &&
  52.                firstTask == null &&
  53.                ! workQueue.isEmpty()))
  54.             return false;
  55.         // 自旋添加任务
  56.         for (;;) {
  57.             // 获取工作线程
  58.             int wc = workerCountOf(c);
  59.             // 首先判断是否大于最大允许数量
  60.             // 然后根据是否核心线程判断是否大于核心线程数或者最大线程数
  61.             if (wc >= CAPACITY ||
  62.                 wc >= (core ? corePoolSize : maximumPoolSize))
  63.                 return false;
  64.             // 对c递增，也就是工作线程数递增
  65.             if (compareAndIncrementWorkerCount(c))
  66.                 break retry;
  67.             // 递增失败，说明存在线程竞争或者状态变更，继续自旋处理
  68.             c = ctl.get(); 
  69.             // 此处不等于，说明存在线程池状态变更
  70.             // 等于，说明只是存在线程竞争造成的CAS失败
  71.             if (runStateOf(c) != rs)
  72.                 continue retry;
  73.         }
  74.     }
  76.     // 上半部分基本是对线程池状态及工作线程数进行判断，并最终对工作线程+1，表示当前线程已经抢占到一个线程位置
  77.     boolean workerStarted = false;
  78.     boolean workerAdded = false;
  79.     Worker w = null;
  80.     try {
  81.         // 包装线程对象为 Worker 对象
  82.         w = new Worker(firstTask);
  83.         // 通过 ThreadFactory 构建一个新的线程
  84.         final Thread t = w.thread;
  85.         if (t != null) {
  86.             // 此处加重入锁
  87.             final ReentrantLock mainLock = this.mainLock;
  88.             mainLock.lock();
  89.             try {
  90.                 // 继续获取线程池状态进行判断
  91.                 int rs = runStateOf(ctl.get());
  93.                 if (rs < SHUTDOWN ||
  94.                     (rs == SHUTDOWN && firstTask == null)) {
  95.                     // 如果线程已经运行中，则说明存在问题，此处线程还没有启用
  96.                     if (t.isAlive())
  97.                         throw new IllegalThreadStateException();
  98.                     // 添加线程包装后的Worker对象到列表中
  99.                     workers.add(w);
  100.                     int s = workers.size();
  101.                     if (s > largestPoolSize)
  102.                         largestPoolSize = s;
  103.                     // 表示工作线程创建成功
  104.                     workerAdded = true;
  105.                 }
  106.             } finally {
  107.                 mainLock.unlock();
  108.             }
  109.             if (workerAdded) {
  110.                 // 线程启动，
  111.                 // 此处注意，Worker 实现了 Runnable接口，则此处是调用 Worker.run()
  112.                 t.start();
  113.                 workerStarted = true;
  114.             }
  115.         }
  116.     } finally {
  117.         // 如果添加失败，则递减工作线程数
  118.         if (! workerStarted)
  119.             addWorkerFailed(w);
  120.     }
  121.     return workerStarted;
  122. }
  124. \* runWorker(Worker w)：Worker的 run() 方法内部调用该方法
  126. final void runWorker(Worker w) {
  127.     Thread wt = Thread.currentThread();
  128.     // 获取初始化 Worker时传递的线程
  129.     // 此处分析的是 execute() 触发，如果是 submit() 触发，此处的Runnable实现类应该为 FutureTask，
  130.     // task.run() 最终调用 FutureTask.run()方法，此处会对 Future的状态进行处理，实现阻塞获取的功能
  131.     Runnable task = w.firstTask;
  132.     w.firstTask = null;
  133.     // unlock，表示当前 worker 线程允许中断，因为 new Worker 默认的 state=-1,
  134.     // 此处是调用Worker 类的 tryRelease()方法，将 state 置为 0， 
  135.     // 而 interruptIfStarted()中只有 state>=0 才允许调用中断
  136.     w.unlock();
  137.     boolean completedAbruptly = true;
  138.     try {
  139.         // 任务不为空，则持续执行
  140.         // getTask()：此处表示不断从阻塞队列中获取元素
  141.         while (task != null || (task = getTask()) != null) {
  142.             w.lock();
  143.             // 线程池状态为stop时不接受新任务，并中断正在执行的人物
  144.             // (Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP)确保线程中断标志位为 true 且是 stop 状态以上，接着清除了中断标志
  145.             // !wt.isInterrupted()则再一次检查保证线程需要设置中断标志位
  146.             if ((runStateAtLeast(ctl.get(), STOP) ||
  147.                  (Thread.interrupted() &&
  148.                   runStateAtLeast(ctl.get(), STOP))) &&
  149.                 !wt.isInterrupted())
  150.                 wt.interrupt();
  151.             try {
  152.                 // 默认没有实现功能，前置处理
  153.                 beforeExecute(wt, task);
  154.                 Throwable thrown = null;
  155.                 try {
  156.                     // 此处注意 task.run() 的task的不同实现
  157.                     // 在submit()触发的功能中，表示FutureTask
  158.                     task.run();
  159.                 } catch (RuntimeException x) {
  160.                     thrown = x; throw x;
  161.                 } catch (Error x) {
  162.                     thrown = x; throw x;
  163.                 } catch (Throwable x) {
  164.                     thrown = x; throw new Error(x);
  165.                 } finally {
  166.                     // 后置执行，没有处理
  167.                     afterExecute(task, thrown);
  168.                 }
  169.             } finally {
  170.                 task = null;
  171.                 w.completedTasks++;
  172.                 w.unlock();
  173.             }
  174.         }
  175.         completedAbruptly = false;
  176.     } finally {
  177.         // 将入参 worker 从数组 workers 里删除掉；
  178.         // 根据布尔值 allowCoreThreadTimeOut 来决定是否补充新的 Worker 进数组workers
  179.         processWorkerExit(w, completedAbruptly);
  180.     }
  181. }
  183. \* getTask()：从阻塞队列中获取下一个有效任务。线程池定义的超时处理再该部分实现
  185. private Runnable getTask() {
  186.     boolean timedOut = false;
  188.     for (;;) {
  189.         int c = ctl.get();
  190.         int rs = runStateOf(c);
  192.         // 校验线程池状态
  193.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  194.             decrementWorkerCount();
  195.             return null;
  196.         }
  198.         // 获取工作线程
  199.         int wc = workerCountOf(c);
  200.         // 对超时线程进行时间控制
  201.         // allowCoreThreadTimeOut默认为false，表示核心线程不收控制
  202.         // wc > corePoolSize：超过核心线程，即最大线程数，则触发控制
  203.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  205.         // timedOut为true，说明上次阻塞操作已经超时，则工作线程数-1，并返回null
  206.         if ((wc > maximumPoolSize || (timed && timedOut))
  207.             && (wc > 1 || workQueue.isEmpty())) {
  208.             if (compareAndDecrementWorkerCount(c))
  209.                 return null;
  210.             continue;
  211.         }
  213.         try {
  214.             // 此处就是对超时控制的处理，在从队列中获取数据时，阻塞获取
  215.             Runnable r = timed ?
  216.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  217.                 workQueue.take();
  218.             // 如果拿到任务，则直接返回去进行处理
  219.             if (r != null)
  220.                 return r;
  221.             // 走到这一步，说明超时，在下一步时候进行回收处理
  222.             timedOut = true;
  223.         } catch (InterruptedException retry) {
  224.             timedOut = false;
  225.         }
  226.     }
  227. }
  229. \* addWorkerFailed(Worker w)：添加工作线程失败
  231. private void addWorkerFailed(Worker w) {
  232.     final ReentrantLock mainLock = this.mainLock;
  233.     mainLock.lock();
  234.     try {
  235.         // 从列表中移除当前 Worker
  236.         if (w != null)
  237.             workers.remove(w);
  238.         // 工作线程数递减
  239.         decrementWorkerCount();
  240.         // 尝试修改线程状态为 Terminate
  241.         tryTerminate();
  242.     } finally {
  243.         mainLock.unlock();
  244.     }
  245. }
  247. private void decrementWorkerCount() {
  248.     do {} while (! compareAndDecrementWorkerCount(ctl.get()));
  249. }
  251. \* reject(Runnable command)：拒绝策略
  253. final void reject(Runnable command) {
  254.     // 直接执行拒绝策略
  255.     handler.rejectedExecution(command, this);
  256. }

4，submit()：同时对 Future 进行分析

  1. \* submit(Callable<T> task)：触发线程执行
  3. public <T> Future<T> submit(Callable<T> task) {
  4.     if (task == null) throw new NullPointerException();
  5.     // 初始化化一个 FutureTask，实现了 Runnable 接口
  6.     RunnableFuture<T> ftask = newTaskFor(task);
  7.     // 直接执行 execute 方法
  8.     execute(ftask);
  9.     return ftask;
  10. }
  12. protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
  13.     return new FutureTask<T>(callable);
  14. }
  16. \* 在 runWorker() 方法中，触发 task.run()，实际调用的是 Future.run()方法
  18. public void run() {
  19.     // 状态不为new，直接执行异常
  20.     if (state != NEW ||
  21.         !UNSAFE.compareAndSwapObject(this, runnerOffset,
  22.                                      null, Thread.currentThread()))
  23.         return;
  24.     try {
  25.         Callable<V> c = callable;
  26.         // 线程一切正常，准备执行
  27.         if (c != null && state == NEW) {
  28.             V result;
  29.             boolean ran;
  30.             try {
  31.                 // 直接调用Callable的call方法，并返回结果
  32.                 result = c.call();
  33.                 ran = true;
  34.             } catch (Throwable ex) {
  35.                 result = null;
  36.                 ran = false;
  37.                 setException(ex);
  38.             }
  39.             // 设置结果
  40.             if (ran)
  41.                 set(result);
  42.         }
  43.     } finally {
  44.         runner = null;
  45.         int s = state;
  46.         if (s >= INTERRUPTING)
  47.             handlePossibleCancellationInterrupt(s);
  48.     }
  49. }
  51. \* set(V v)：设置执行结果
  53. protected void set(V v) {
  54.     // 设置状态为完成
  55.     if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
  56.         outcome = v;
  57.         // 设置正常结束
  58.         UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
  59.         // 完成后续处理，唤醒等待节点
  60.         finishCompletion();
  61.     }
  62. }
  64. \* finishCompletion()：该方法主要是唤醒等待节点，去返回结果中拿数据
  66. private void finishCompletion() {
  67.     // 获取等待节点
  68.     for (WaitNode q; (q = waiters) != null;) {
  69.         if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
  70.             for (;;) {
  71.                 // 等待节点线程优先，直接唤醒
  72.                 Thread t = q.thread;
  73.                 if (t != null) {
  74.                     q.thread = null;
  75.                     LockSupport.unpark(t);
  76.                 }
  77.                 // 递归下一个等待节点同步处理
  78.                 WaitNode next = q.next;
  79.                 if (next == null)
  80.                     break;
  81.                 q.next = null; // unlink to help gc
  82.                 q = next;
  83.             }
  84.             break;
  85.         }
  86.     }
  88.     done();
  90.     callable = null;        // to reduce footprint
  91. }

5，Funture.get()

  1. \* get()：阻塞获取数据
  3. public V get() throws InterruptedException, ExecutionException {
  4.     // 获取Future对应的线程执行状态，如果没有执行完直接等待
  5.     int s = state;
  6.     if (s <= COMPLETING)
  7.         s = awaitDone(false, 0L);
  8.     // 执行完成后，解析结果集
  9.     return report(s);
  10. }
  12. \* awaitDone(boolean timed, long nanos)
  14. private int awaitDone(boolean timed, long nanos) throws InterruptedException {
  15.     final long deadline = timed ? System.nanoTime() + nanos : 0L;
  16.     WaitNode q = null;
  17.     boolean queued = false;
  18.     for (;;) {
  19.         // 线程中断，直接移除
  20.         if (Thread.interrupted()) {
  21.             removeWaiter(q);
  22.             throw new InterruptedException();
  23.         }
  25.         int s = state;
  26.         // 此处说明执行完成
  27.         if (s > COMPLETING) {
  28.             if (q != null)
  29.                 q.thread = null;
  30.             return s;
  31.         }
  32.         // 还有后续操作没有执行完成，暂时让出时间片段，稍后执行
  33.         else if (s == COMPLETING)
  34.             Thread.yield();
  35.         // 表示状态为null，构建等待节点，准备等待
  36.         else if (q == null)
  37.             q = new WaitNode();
  38.         // 使用CAS添加等待节点到等待队列
  39.         else if (!queued)
  40.             queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
  41.                                                  q.next = waiters, q);
  42.         // 此处表示设置了超时
  43.         else if (timed) {
  44.             // 如果超时时间没有获取到值，则直接退出
  45.             nanos = deadline - System.nanoTime();
  46.             if (nanos <= 0L) {
  47.                 removeWaiter(q);
  48.                 return state;
  49.             }
  50.             LockSupport.parkNanos(this, nanos);
  51.         }
  52.         else
  53.             // 添加等待队列成功后，线程挂起等待，等待执行完成后进行唤醒，完成那部分已经分析
  54.             LockSupport.park(this);
  55.     }
  56. }
  58. \* report(int s)：获取返回值
  60. private V report(int s) throws ExecutionException {
  61.     // 获取返回值
  62.     Object x = outcome;
  63.     if (s == NORMAL)
  64.         return (V)x;
  65.     if (s >= CANCELLED)
  66.         throw new CancellationException();
  67.     throw new ExecutionException((Throwable)x);
  68. }

#

发表评论

表情：
评论列表 （有 0 条评论，178人围观）

还没有评论，来说两句吧...

相关阅读

    相关 并发编程-线程池

    文章目录 线程池 线程池原理 线程池分类 线程池 为什么需要使用线程池？ 线程的创建和销毁都需要消耗系统资源，线程池可以复

    小灰灰小灰灰/ 0 赞/ 174 阅读