并发编程——线程池

╰+攻爆jí腚メ 2023-10-16 11:57 178阅读 0赞

文章目录

  • 线程池介绍
  • 线程池核心属性
  • 核心方法
    • execute方法
    • addWorker添加工作线程
    • runWorker执行任务
    • getTask工作线程排队拿任务

线程池介绍

Java构建线程的方式

  • new Thread
  • new Runnable
  • new Callable

为了避免频繁创建和销毁线程造成不必要的性能,一般在使用线程时,会采用线程池。

线程池使用方式:

  1. public static void main(String[] args) {
  2. // 线程池的核心线程数如何设置
  3. // 任务可以分为两种:CPU密集,IO密集。
  4. ThreadPoolExecutor executor = new ThreadPoolExecutor(
  5. 1,
  6. 2,
  7. 1,
  8. TimeUnit.SECONDS,
  9. new ArrayBlockingQueue<>(1),
  10. new ThreadFactory() {
  11. @Override
  12. public Thread newThread(Runnable r) {
  13. Thread t = new Thread(r);
  14. // ...
  15. return t;
  16. }
  17. },
  18. new ThreadPoolExecutor.AbortPolicy()
  19. );
  20. executor.execute(任务);
  21. executor.submit(有返回结果的任务);
  22. }

线程池提供的拒绝策略:

  • AbortPolicy:抛出异常。
  • CallerRunsPolicy:让提交任务的线程处理这个任务。
  • DiscardPolicy:丢弃任务
  • DiscardOldestPolicy:扔掉队列最前面的任务,尝试把当前任务添加进去。

线程池核心属性

  1. // AtomicInteger,就是一个int,写操作用CAS实现,保证了原子性
  2. // ctl维护线程池的2个核心内容:
  3. // 1:线程池状态(高3位,维护着线程池状态)
  4. // 2:工作线程数量(核心线程+非核心线程,低29位,维护着工作线程个数)
  5. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  6. // COUNT_BITS=29
  7. private static final int COUNT_BITS = Integer.SIZE - 3;
  8. // 工作线程的最大个数
  9. // 00100000 00000000 00000000 00000000 - 1
  10. // 000111111111111111111111111111111
  11. private static final int CAPACITY = (1 << COUNT_BITS) - 1;
  12. private static final int RUNNING = -1 << COUNT_BITS;
  13. private static final int SHUTDOWN = 0 << COUNT_BITS;
  14. private static final int STOP = 1 << COUNT_BITS;
  15. private static final int TIDYING = 2 << COUNT_BITS;
  16. private static final int TERMINATED = 3 << COUNT_BITS;
  17. // 拿到线程池状态
  18. private static int runStateOf(int c) {
  19. return c & ~CAPACITY; }
  20. // 拿到工作线程个数
  21. private static int workerCountOf(int c) {
  22. return c & CAPACITY; }

线程池状态

  1. RUNNING:正常运行的状态,接受新的任务,处理等待队列中的任务。线程池的初始化状态是RUNNING。一旦线程池被创建,就处于RUNNING状态,并且线程池中的任务数为0。
  2. SHUTDOWN:不接受新的任务提交,但是会继续处理等待队列中的任务。调用线程池的shutdown()方法时,线程池由RUNNING变为SHUTDOWN。
  3. STOP:不接受新的任务提交,不再处理等待队列中的任务,中断正在执行任务的线程。调用线程池的shutdownNow()方法时,线程池由(RUNNING or SHUTDOWN)变为STOP。
  4. TIDYING:所有的任务都销毁了,workCount为0,线程池的状态在转换为TIDYING状态时,会执行钩子方法terminated()。
  5. TERMINATED:线程池彻底终止。

核心方法

execute方法

通过execute方法,可以看到线程池处理任务的整体执行流程

  1. public void execute(Runnable command) {
  2. if (command == null)
  3. throw new NullPointerException();
  4. // 拿到ctl
  5. int c = ctl.get();
  6. // 通过ctl获取当前工作线程个数
  7. if (workerCountOf(c) < corePoolSize) {
  8. // true:代表是核心线程,false:代表是非核心线程
  9. if (addWorker(command, true))
  10. // 如果添加核心线程成功,return结束掉
  11. return;
  12. // 如果添加失败,重新获取ctl
  13. c = ctl.get();
  14. }
  15. // 核心线程数已经到了最大值、添加时,线程池状态变为SHUTDOWN/STOP
  16. // 判断线程池是否是运行状态 && 添加任务到工作队列
  17. if (isRunning(c) && workQueue.offer(command)) {
  18. // 再次获取ctl的值
  19. int recheck = ctl.get();
  20. // 再次判断线程池状态。 DCL
  21. // 如果状态不是RUNNING,把任务从工作队列移除。
  22. if (! isRunning(recheck) && remove(command))
  23. // 走一波拒绝策略。
  24. reject(command);
  25. // 线程池状态是RUNNING。
  26. // 判断工作线程数是否是0个。
  27. // 可以将核心线程设置为0,所有工作线程都是非核心线程。
  28. // 核心线程也可以通过keepAlived超时被销毁,所以如果恰巧核心线程被销毁,也会出现当前效果
  29. else if (workerCountOf(recheck) == 0)
  30. // 添加空任务的非核心线程去处理工作队列中的任务
  31. addWorker(null, false);
  32. }
  33. // 可能工作队列中的任务存满了,没添加进去,到这就要添加非核心线程去处理任务
  34. else if (!addWorker(command, false))
  35. // 执行拒绝策略!
  36. reject(command);
  37. }

这个方法的主要逻辑是:

  1. 如果当前运行的线程数少于corePoolSize,尝试启动一个新线程并将给定的任务作为它的第一个任务。这个调用会原子性地检查runState和workerCount,以防止在不应该添加线程时发出错误警报。
  2. 如果任务可以成功排队,那么我们仍然需要再次检查是否应该添加一个新线程(因为自上次检查以来,现有的线程可能已经死亡),或者自进入此方法以来,线程池是否已经关闭。所以我们重新检查状态,并在必要时回滚排队,如果已停止,则启动一个新线程。
  3. 如果我们无法排队任务,那么我们尝试添加一个新线程。如果失败,我们知道已经关闭或饱和,所以拒绝任务。

addWorker添加工作线程

  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2. xxx:
  3. for (;;) {
  4. int c = ctl.get();
  5. int rs = runStateOf(c);
  6. // 判断线程池状态
  7. if (rs >= SHUTDOWN &&
  8. // 判断如果线程池的状态为SHUTDOWN,还要处理工作队列中的任务
  9. // 如果你添加工作线程的方式,是任务的非核心线程,并且工作队列还有任务
  10. ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
  11. return false;
  12. // 判断工作线程个数
  13. for (;;) {
  14. int wc = workerCountOf(c);
  15. // 判断1:工作线程是否已经 == 工作线程最大个数
  16. // 判断2-true判断:判断是核心线程么?如果是判断是否超过核心线程个数
  17. // 判断2-false判断:如果是非核心线程,查看是否超过设置的最大线程数
  18. if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
  19. return false;
  20. // 对工作线程进行 + 1操作
  21. if (compareAndIncrementWorkerCount(c))
  22. // +1成功,跳出外层循环,执行添加工作线程的业务
  23. // 以CAS方式,对ctl+1,多线程并发操作,只有会有一个成功
  24. break xxx;
  25. // 重新拿ctl,
  26. c = ctl.get();
  27. // 判断线程池状态是否有变化
  28. if (runStateOf(c) != rs)
  29. continue xxx;
  30. }
  31. }
  32. // 添加工作线程的业务
  33. boolean workerStarted = false;
  34. boolean workerAdded = false;
  35. // Worker就是工作线程
  36. Worker w = null;
  37. try {
  38. // 创建工作线程,将任务传到Worker中
  39. w = new Worker(firstTask);
  40. final Thread t = w.thread;
  41. // 只有你写的线程工厂返回的是null,这里才会为null
  42. if (t != null) {
  43. // 获取锁资源
  44. final ReentrantLock mainLock = this.mainLock;
  45. // 加锁。 因为我要在启动这个工作线程时,避免线程池状态发生变化,加锁。
  46. mainLock.lock();
  47. try {
  48. // 重新获取ctl,拿到线程池状态
  49. int rs = runStateOf(ctl.get());
  50. // DCL i think you know~~~
  51. if (rs < SHUTDOWN ||
  52. (rs == SHUTDOWN && firstTask == null)) {
  53. // 判断Worker中的thread是否已经启动了,一般不会启动,除非你在线程工厂把他启动了
  54. if (t.isAlive())
  55. throw new IllegalThreadStateException();
  56. // 将工作线程存储到hashSet中
  57. workers.add(w);
  58. // 获取工作线程个数,判断是否需要修改最大工作线程数记录。
  59. int s = workers.size();
  60. if (s > largestPoolSize)
  61. largestPoolSize = s;
  62. // 工作线程添加成功 0
  63. workerAdded = true;
  64. }
  65. } finally {
  66. mainLock.unlock();
  67. }
  68. // 如果添加成功
  69. if (workerAdded) {
  70. // 启动工作线程
  71. t.start();
  72. // 设置标识为true
  73. workerStarted = true;
  74. }
  75. }
  76. } finally {
  77. // 如果工作线程启动失败
  78. if (! workerStarted)
  79. addWorkerFailed(w);
  80. }
  81. return workerStarted;
  82. }
  83. // 如果添加工作线程失败,执行
  84. private void addWorkerFailed(Worker w) {
  85. final ReentrantLock mainLock = this.mainLock;
  86. mainLock.lock();
  87. try {
  88. // 说明worker可能存放到了workers的hashSet中。
  89. if (w != null)
  90. // 移除!
  91. workers.remove(w);
  92. // 减掉workerCount的数值 -1
  93. decrementWorkerCount();
  94. // 尝试干掉自己
  95. tryTerminate();
  96. } finally {
  97. mainLock.unlock();
  98. }
  99. }

这个方法的主要逻辑是:

  1. 检查是否正在关闭或已经关闭,如果是,根据是否有任务以及队列是否为空来决定是否返回false。
  2. 如果没有关闭,那么检查当前线程数是否超过了核心线程数或最大线程数,如果超过了则返回false。
  3. 如果没有超过线程数限制,那么通过CAS操作增加workerCount,如果成功则跳出循环。如果失败,则重新读取ctl并重新计算runState,如果runState发生变化则继续循环。如果未发生变化,则继续尝试增加workerCount。
  4. 创建Worker对象,并启动对应的线程。如果启动成功,则返回true。如果启动失败,则调用addWorkerFailed()方法处理失败情况。

runWorker执行任务

  1. final void runWorker(Worker w) {
  2. // 拿到当前线程对象
  3. Thread wt = Thread.currentThread();
  4. // 拿到worker中存放的Runnable
  5. Runnable task = w.firstTask;
  6. // 将worker中的任务清空
  7. w.firstTask = null;
  8. boolean completedAbruptly = true;
  9. try {
  10. // 如果Worker自身携带任务,直接执行
  11. // 如果Worker携带的是null,通过getTask去工作队列获取任务
  12. while (task != null || (task = getTask()) != null) {
  13. w.lock();
  14. // 判断线程池状态是否大于等于STOP,如果是要中断当前线程
  15. if ((runStateAtLeast(ctl.get(), STOP) ||
  16. // 中断当前线程(DCL)
  17. (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted())
  18. wt.interrupt();
  19. try {
  20. // 前置钩子
  21. beforeExecute(wt, task);
  22. Throwable thrown = null;
  23. try {
  24. // 执行任务
  25. task.run();
  26. } catch (RuntimeException x) {
  27. thrown = x; throw x;
  28. } catch (Error x) {
  29. thrown = x; throw x;
  30. } catch (Throwable x) {
  31. thrown = x; throw new Error(x);
  32. } finally {
  33. // 后置钩子
  34. afterExecute(task, thrown);
  35. }
  36. } finally {
  37. task = null;
  38. // 当前工作执行完一个任务,就++
  39. w.completedTasks++;
  40. w.unlock();
  41. }
  42. }
  43. completedAbruptly = false;
  44. } finally {
  45. processWorkerExit(w, completedAbruptly);
  46. }
  47. }

这个方法的主要逻辑是:

  1. 获取当前线程和任务。如果任务不为空,则执行任务。如果任务为空,则从任务队列中获取任务。如果获取任务失败,则退出循环。
  2. 在执行任务前,检查线程是否中断,如果中断则停止执行任务。如果未中断,则执行任务。如果抛出运行时异常或错误,则捕获并记录异常。如果抛出其他异常,则转换为错误并抛出。
  3. 在执行任务后,将任务设置为null,增加已完成任务数,并解锁worker。
  4. 如果循环结束时,任务为空,则说明任务队列已关闭或已空,将completedAbruptly设置为false。否则,说明任务队列中还有任务,将completedAbruptly保持为true。
  5. 最后,处理worker退出,根据completedAbruptly的值决定是否重置线程池状态。

getTask工作线程排队拿任务

  1. private Runnable getTask() {
  2. boolean timedOut = false; // Did the last poll() time out?
  3. for (;;) {
  4. int c = ctl.get();
  5. int rs = runStateOf(c);
  6. // 线程池状态判断
  7. // 如果线程池状态为SHUTDOWN && 工作队列为空
  8. // 如果线程池状态为STOP
  9. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  10. // 对工作线程个数--
  11. decrementWorkerCount();
  12. return null;
  13. }
  14. // 对数量的判断。
  15. int wc = workerCountOf(c);
  16. // 判断核心线程是否允许超时?
  17. // 工作线程个数是否大于核心线程数
  18. boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  19. // 判断工作线程是否超过了最大线程数 && 工作队列为null
  20. if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
  21. // 工作线程数有问题,必须-1,干掉当前工作线程
  22. // 工作线程是否超过了核心线程,如果超时,就干掉当前线程
  23. // 对工作线程个数--
  24. if (compareAndDecrementWorkerCount(c))
  25. return null;
  26. continue;
  27. }
  28. try {
  29. // 如果是非核心,走poll,拉取工作队列任务,
  30. // 如果是核心线程,走take一直阻塞,拉取工作队列任务
  31. Runnable r = timed ?
  32. workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  33. // 当工作队列没有任务时,这时就会被Condition通过await阻塞线程
  34. // 当有任务添加到工作线程后,这是添加完任务后,就会用过Condition.signal唤醒阻塞的线程
  35. workQueue.take();
  36. if (r != null)
  37. return r;
  38. // 执行的poll方法,并且在指定时间没拿到任务,
  39. timedOut = true;
  40. } catch (InterruptedException retry) {
  41. timedOut = false;
  42. }
  43. }
  44. }

该方法的主要逻辑是:

  1. 检查线程池状态,如果线程池已经关闭或停止,并且任务队列为空,则减少worker数量并返回null。
  2. 检查worker数量是否超过了最大线程数或核心线程数,如果是,则减少worker数量并返回null。
  3. 如果worker数量未超过最大线程数或核心线程数,则从任务队列中获取任务。如果获取任务失败,则返回null。如果获取任务成功,则返回任务。
  4. 如果在获取任务时发生中断,则重新尝试获取任务。如果在指定的时间内无法获取任务,则将timedOut标志设置为true。如果在指定的时间内无法获取任务并且允许超时,则减少worker数量并返回null。

发表评论

表情:
评论列表 (有 0 条评论,178人围观)

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

相关阅读

    相关 并发编程-线

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

    相关 并发编程线线

    1. 线程 线程(英语:thread)是操作系统能够进行运算调度的最小单位。它被包含在进程之中,是进程中的实际运作单位。 线程是调度CPU资源的最小单位,线程模型分为K

    相关 并发编程线

    为什么使用线程池 有时候,系统需要处理非常多的执行时间很短的请求,如果每一个请求都开启一个新线程的话,系统就要不断的进行线程的创建和销毁,有时花在创建和销毁线程上的时间会