乐趣区

关于后端:Java并发编程阿里最喜欢问的几道线程池的面试题

jiezi

引言

上一篇文章咱们有介绍过线程池的一个根本执行流程《【Java 并发编程】面试必备之线程池》以及它的 7 个外围参数，以及每个参数的作用、以及如何去应用线程池
还留了几个小问题。。倡议看这篇文章之前可先看下后面那篇文章。这篇文章咱们就来剖析下上篇文章的几个小问题

  • 线程池是否辨别外围线程和非核心线程？
  • 如何保障外围线程不被销毁？
  • 线程池的线程是如何做到复用的？

咱们先看最初一个问题个别一个线程执行完工作之后就完结了，Thread.start()只能调用一次，一旦这个调用完结，则该线程就到了 stop 状态，不能再次调用 start。如果你对一个曾经启动的线程对象再调用一次start 办法的话, 会产生:IllegalThreadStateException异样，然而 Thread 的run办法是能够反复调用的。所以这里也会有一个面试常常问到的问题：Thread 类中 run()和 start()办法的有什么区别？
上面咱们就从 jdk 的源码来一起看看如何实现线程复用的：
线程池执行工作的 ThreadPoolExecutor#execute 办法为入口

 public void execute(Runnable command) {if (command == null)
         throw new NullPointerException();
         
     int c = ctl.get();
     // 线程池以后线程数小于 corePoolSize 时进入 if 条件调用 addWorker 创立外围线程来执行工作
     if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))
             return;
         c = ctl.get();}
     // 线程池以后线程数大于或等于 corePoolSize , 就将工作增加到 workQueue 中
     if (isRunning(c) && workQueue.offer(command)) {
         // 获取到以后线程的状态, 赋值给 recheck , 是为了从新查看状态
         int recheck = ctl.get();
         // 如果 isRunning 返回 false , 那就 remove 掉这个工作, 而后执行回绝策略, 也就是回滚从新排队
         if (! isRunning(recheck) && remove(command))
             reject(command);
         // 线程池处于 running 状态, 然而没有线程, 那就创立线程执行工作
         else if (workerCountOf(recheck) == 0)
             addWorker(null, false);
     }
     // 如果工作放入 workQueue 失败, 则尝试通过创立非核心线程来执行工作
     // 创立非核心线程失败, 则阐明线程池曾经敞开或者曾经饱和, 会执行回绝策略
     else if (!addWorker(command, false))
         reject(command);
 }

excute办法次要业务逻辑

  • 如果以后的线程池运行线程小于coreSize，则创立新线程来执行工作。
  • 如果以后运行的线程等于 coreSize 或多余coreSize(动静批改了 coreSize 才会呈现这种状况)，把工作放到阻塞队列中。
  • 如果队列已满无奈将新退出的工作放进去的话，则须要创立新的线程来执行工作。
  • 如果新创建线程曾经达到了最大线程数，工作将会被回绝。

addWorker 办法

上述办法的外围次要就是 addWorker 办法,

private boolean addWorker(Runnable firstTask, boolean core) {
       // 后面还有一部分就省略了。。。。boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {mainLock.unlock();
                }
                if (workerAdded) {t.start();
                    workerStarted = true;
                }
            }
        } finally {if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

这个办法咱们先看看这个 work 类吧

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        
        public void run() {runWorker(this);
        }

work类实现了 Runnable 接口，而后 run 办法外面调用了 runWorker 办法

final void runWorker(Worker w) {Thread wt = Thread.currentThread();
        // 新增创立
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
             // 判断 task 是否为空, 如果不为空间接执行
            // 如果 task 为空, 调用 getTask() 办法, 从 workQueue 中取出新的 task 执行
            while (task != null || (task = getTask()) != null) {w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {task.run();
                    } catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error(x);
                    } finally {afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();}
            }
            completedAbruptly = false;
        } finally {processWorkerExit(w, completedAbruptly);
        }
    }

这个 runwork 办法中会优先取 worker 绑定的工作，如果创立这个 worker 的时候没有给 worker 绑定工作，worker就会从队列外面获取工作来执行，执行完之后 worker 并不会销毁，而是通过 while 循环不停的执行 getTask 办法从阻塞队列中获取工作调用 task.run()来执行工作，这样的话就达到了线程复用的目标。while (task != null || (task = getTask()) != null) 这个循环条件只有 getTask 返回获取的值不为空这个循环就不会终止，这样线程也就会始终在运行。
那么工作执行完怎么保障外围线程不销毁？非核心线程销毁？
答案就在这个 getTask() 办法外面

private Runnable getTask() {// 超时标记，默认为 false，如果调用 workQueue.poll()办法超时了，会标记为 true
  // 这个标记十分之重要，上面会说到
  boolean timedOut = false;
  for (;;) {
    // 获取 ctl 变量值
    int c = ctl.get();
    int rs = runStateOf(c);

    // 如果以后状态大于等于 SHUTDOWN，并且 workQueue 中的工作为空或者状态大于等于 STOP
    // 则操作 AQS 缩小工作线程数量，并且返回 null，线程被回收
    // 也阐明假如状态为 SHUTDOWN 的状况下，如果 workQueue 不为空，那么线程池还是能够继续执行剩下的工作
    if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
      // 操作 AQS 将线程池中的线程数量减一
      decrementWorkerCount();
      return null;
    }

    // 获取线程池中的无效线程数量
    int wc = workerCountOf(c);

    // 如果被动开启 allowCoreThreadTimeOut, 或者获取当前工作线程大于 corePoolSize，那么该线程是能够被超时回收的
    // allowCoreThreadTimeOut 默认为 false，即默认不容许外围线程超时回收
    // 这里也阐明了在外围线程以外的线程都为“长期”线程，随时会被线程池回收
    boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
    // 这里阐明了两点销毁线程的条件：// 1. 原则上线程池数量不可能大于 maximumPoolSize，但可能会呈现并发时操作了 setMaximumPoolSize 办法，如果此时将最大线程数量调少了，很可能会呈现当前工作线程大于最大线程的状况，这时就须要线程超时回收，以维持线程池最大线程小于 maximumPoolSize，// 2.timed && timedOut 如果为 true，示意以后操作须要进行超时管制，这里的 timedOut 为 true，阐明该线程曾经从 workQueue.poll()办法超时了
    // 以上两点满足其一，都能够触发线程超时回收
    if ((wc > maximumPoolSize || (timed && timedOut))
        && (wc > 1 || workQueue.isEmpty())) {
      // 尝试用 AQS 将线程池线程数量减一
      if (compareAndDecrementWorkerCount(c))
        // 减一胜利后返回 null，线程被回收
        return null;
      // 否则循环重试
      continue;
    }

    try {
      // 如果 timed 为 true，阻塞超时获取工作，否则阻塞获取工作
      Runnable r = timed ?
        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
      workQueue.take();
      if (r != null)
        return r;
      // 如果 poll 超时获取工作超时了, 将 timeOut 设置为 true
      // 持续循环执行，如果碰巧开发者开启了 allowCoreThreadTimeOut，那么该线程就满足超时回收了
      timedOut = true;
    } catch (InterruptedException retry) {timedOut = false;}
  }
}

所以保障线程不被销毁的要害代码就是这一句代码

   Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();

只有 timed 为false这个 workQueue.take() 就会始终阻塞，也就保障了线程不会被销毁。timed的值又是通过 allowCoreThreadTimeOut 和正在运行的线程数量是否大于 coreSize 管制的。

  • 只有 getTask 办法返回 null 咱们的线程就会被回收(runWorker 办法会调用processWorkerExit)
  • 这个办法的源码也就解释了为什么咱们在创立线程池的时候设置了 allowCoreThreadTimeOut =true 的话，外围线程也会进行销毁。
  • 通过这个办法我也们能够答复下面那个问题线程池是不辨别外围线程和非核心线程的。

完结

  • 因为本人满腹经纶，难免会有纰漏，如果你发现了谬误的中央，还望留言给我指出来, 我会对其加以修改。
  • 如果你感觉文章还不错，你的转发、分享、赞叹、点赞、留言就是对我最大的激励。
  • 感谢您的浏览, 非常欢送并感谢您的关注。


伟人的肩膀摘苹果
http://objcoding.com/2019/04/…

退出移动版