线程池_03_源码分析

  1. 云栖社区>
  2. 博客>
  3. 正文

线程池_03_源码分析

坎布里奇 2020-02-27 00:35:06 浏览159
展开阅读全文

线程池_03_源码分析
  • 关键变量介绍:
    •  线程的状态
      • 5种
        • RUNNABLE:运行状态,接受新任务,持续处理任务队列里的任务
        • SHUTDOWN:不再接受新任务,但要处理任务队列里的任务
        • STOP:不接受新任务,不再处理任务队列里的任务,中断正在进行中的任务
        • TIDYING:表示线程池正在停止运作,中止所有任务,销毁所有工作线程
        • TERMINATED:表示线程池已停止运作,所有工作线程已被销毁,所有任务已被清空或执行完毕
      • 状态装换
        07245109-4128-4ba2-a9c5-84a525143230-5210562.jpg
      • 细节
        c99e8f16-0748-49f2-b843-5a67cb157357-5210562.jpg
        • 线程的状态使用ctl表示,默认为 running
        • private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        • ctl类型为AtomicInteger,那用一个基础如何表示以上五种状态以及线程池工作线程数量呢?int型变量占用4字节,共32位,因此采用位表示,可以解决上述问题。5种状态使用5种数值进行表示,需要占用3位,余下的29位就可以用来表示线程数。因此,高三位表示进程状态,低29位为线程数量,代码如下:
          • private staticfinal int COUNT_BITS = Integer.SIZE - 3; // 值为29
          • private staticfinal int CAPACITY   = (1 <<COUNT_BITS) - 1; //高三位全为0,低29位全为1,因此线程数量的表示范围为 0 ~ 2^29
          • 因为ctl分位来表示状态和数量,下面几个状态仅看有效位的值
            • private staticfinal int RUNNING    = -1 <<COUNT_BITS; // 有效值为 111
            • private staticfinal int SHUTDOWN   =  0 << COUNT_BITS; // 有效值为 000
            • private staticfinal int STOP       =  1 << COUNT_BITS; // 有效值为 001
            • private staticfinal int TIDYING    =  2 << COUNT_BITS; // 有效值为 010
            • private staticfinal int TERMINATED =  3 <<COUNT_BITS; // 有效值为 011
        • 采用int分位表示线程池状态和线程数量; 并且提供runStateOf(): 获取线程池状态 和 workerCountOf(): 获取工作线程数量两个方法,均为二进制操作。
    • 工作线程(Worker)
      •  线程池中的工作线程以Worker作为体现,真正工作的线程为Worker的成员变量,Worker即是Runnable,又是同步器。Worker从工作队列中取出任务来执行,并能通过Worker控制任务状态。
  • 执行任务,源码分析:
    • 1. execute() --提交线程任务;
      668b29d9-3487-4a67-9aa5-43f60f3d58a4-5210562.jpg
    • 2. addWorker()  -- 通过添加核心和非核心线程来执行任务
      • private boolean addWorker(Runnable firstTask, boolean core) {
        • int c = this.ctl.get();           // 获取当前ctl值
        • label253:
        • while(!runStateAtLeast(c, 0) || !runStateAtLeast(c, 536870912) && firstTask == null && !this.workQueue.isEmpty()) {
        • // workerCountOf(c)-获取线程数  <  core ? this.corePoolSize : this.maximumPoolSize 如果为true,核心线程;否则,非核心线程
          • while(workerCountOf(c) < ((core ? this.corePoolSize : this.maximumPoolSize) & 536870911)) {
          • // CAS操作增加线程数,跳出循环
          • if (this.compareAndIncrementWorkerCount(c)) {
            • boolean workerStarted = false;
            • boolean workerAdded = false;
            • ThreadPoolExecutor.Worker w = null;
            • try {
              • w = new ThreadPoolExecutor.Worker(firstTask);
              • Thread t = w.thread;
              • if (t != null) {
                • ReentrantLock mainLock = this.mainLock;
                • mainLock.lock();
                • try {
                  • int c = this.ctl.get();
                    • if (isRunning(c) || runStateLessThan(c, 536870912) && firstTask == null) {
                      • if (t.isAlive()) {
                        • throw new IllegalThreadStateException();
                      • }
                      • this.workers.add(w);
                      • int s = this.workers.size();
                      • if (s > this.largestPoolSize) {
                        • this.largestPoolSize = s;
                      • }
                      • workerAdded = true;
                    • }
                  • } finally {
                    • mainLock.unlock();
                  • }
                  • if (workerAdded) {
                    • t.start();
                    • workerStarted = true;
                  • }
                • }
              • } finally {
                • if (!workerStarted) {
                  • this.addWorkerFailed(w);
                • }
              • }
              • return workerStarted;
            • }
            • c = this.ctl.get();
            • // 上面的CAS操作没成功,检查线程池状态与开始是否一致;如果一致,继续执行此for循环,否则重新执行retry代码块;自旋以期CAS成功,后续才能添加线程
            • if (runStateAtLeast(c, 0)) {
              • continue label253;
            • }
          • }
          • return false;
        • }
          • return false;
        • }
      • addWorkerFailed(w)
        • 首先是将Worker移除,然后通过CAS操作更新ctl,最后调用tryTerminate()操作尝试中止线程池。
    • 4. runWorker()
      • 线程首个任务为firstTask,之后通过getTask()就从阻塞队列里任务。线程池提供了beforeExecute()和afterExecute()通知子类任务执行前后的回调,让子类有时机能执行自己的事情。如果线程池已没有任务了,工作线程达到了可退出的状态,则将线程退出。
      • 线程池执行的任务的线程,也就是Workder里的Thread。因此在addWorker()中执行new ThreadPoolExecutor.Worker(firstTask)后;执行的是Worker.run(),run()则调用了ThreadPoolExecutor.runWorker()
        d9f8af34-6084-4ebe-9aa7-72d1bbc5a256-5210562.jpg
    • 5. getTask()
      • 线程池里的线程从阻塞队列里拿任务,如果存在非核心线程,假设阻塞队列里没有任务,那么非核心线程也要在等到keepAliveTime时间后才会释放。
      • 如果当前仅有核心线程存在,如果允许释放核心线程的话,也就和非核线程的处理方式一样,反之,则通过take()一直阻塞直到拿到任务,这也就是线程池里的核心线程为什么不死的原因。
        1662b122-3448-48f5-b8d0-bab47984fb13-5210562.jpg
    • 6. processWorkerExit() 
      • 在线程没有拿到任务后,退出线程
      • 释放工作线程也并没有区分核心与非核心,也是随机进行的。所谓随机,就是在前面所说的区间范围内,根据释放策略,哪个线程先达到获取不到任务的状态,就释放哪个线程。
        11eb2680-5719-47ee-9abe-88af2232abda-5210562.jpg
      • tryTerminate()
        • 发现可以中止线程池时,中止,并调用terminated()进行通知。如果线程池处于RUNNABLE状态,什么也不做,否则尝试中断一个线程。中断线程通过interruptIdleWorker()完成。

网友评论

登录后评论
0/500
评论
坎布里奇
+ 关注