java 中ThreadPoolExecutor原理分析

java 中ThreadPoolExecutor原理分析

线程池简介

Java线程池是开发中常用的工具，当我们有异步、并行的任务要处理时，经常会用到线程池，或者在实现一个服务器时，也需要使用线程池来接收连接处理请求。

线程池使用

JDK中提供的线程池实现位于java.util.concurrent.ThreadPoolExecutor。在使用时，通常使用ExecutorService接口，它提供了submit,invokeAll,shutdown等通用的方法。

在线程池配置方面，Executors类中提供了一些静态方法能够提供一些常用场景的线程池，如newFixedThreadPool,newCachedThreadPool,newSingleThreadExecutor等，这些方法最终都是调用到了ThreadPoolExecutor的构造函数。

ThreadPoolExecutor的包含所有参数的构造函数是

/**
   * @param corePoolSize the number of threads to keep in the pool, even
   *    if they are idle, unless {@code allowCoreThreadTimeOut} is set
   * @param maximumPoolSize the maximum number of threads to allow in the
   *    pool
   * @param keepAliveTime when the number of threads is greater than
   *    the core, this is the maximum time that excess idle threads
   *    will wait for new tasks before terminating.
   * @param unit the time unit for the {@code keepAliveTime} argument
   * @param workQueue the queue to use for holding tasks before they are
   *    executed. This queue will hold only the {@code Runnable}
   *    tasks submitted by the {@code execute} method.
   * @param threadFactory the factory to use when the executor
   *    creates a new thread
   * @param handler the handler to use when execution is blocked
   *    because the thread bounds and queue capacities are reached
  public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue,
               ThreadFactory threadFactory,
               RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
      maximumPoolSize <= 0 ||
      maximumPoolSize < corePoolSize ||
      keepAliveTime < 0)
      throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
      throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
  }

• corePoolSize设置线程池的核心线程数，当添加新任务时，如果线程池中的线程数小于corePoolSize，则不管当前是否有线程闲置，都会创建一个新的线程来执行任务。
• maximunPoolSize是线程池中允许的最大的线程数
• workQueue用于存放排队的任务
• keepAliveTime是大于corePoolSize的线程闲置的超时时间
• handler用于在任务逸出、线程池关闭时的任务处理 ，线程池的线程增长策略为，当前线程数小于corePoolSize时，新增线程，当线程数=corePoolSize且corePoolSize时，只有在workQueue不能存放新的任务时创建新线程，超出的线程在闲置keepAliveTime后销毁。
  

实现(基于JDK1.8)

ThreadPoolExecutor中保存的状态有

当前线程池状态, 包括RUNNING,SHUTDOWN,STOP,TIDYING,TERMINATED。

当前有效的运行线程的数量。

将这两个状态放到一个int变量中，前三位作为线程池状态，后29位作为线程数量。

例如0b11100000000000000000000000000001， 表示RUNNING, 一个线程。

通过HashSet来存储工作者集合，访问该HashSet前必须先获取保护状态的mainLock:ReentrantLock

submit、execute

execute的执行方式为，首先检查当前worker数量，如果小于corePoolSize，则尝试add一个core Worker。线程池在维护线程数量以及状态检查上做了大量检测。

public void execute(Runnable command) {
    int c = ctl.get();
    // 如果当期数量小于corePoolSize
    if (workerCountOf(c) < corePoolSize) {
      // 尝试增加worker
      if (addWorker(command, true))
        return;
      c = ctl.get();
    }
    // 如果线程池正在运行并且成功添加到工作队列中
    if (isRunning(c) && workQueue.offer(command)) {
      // 再次检查状态，如果已经关闭则执行拒绝处理
      int recheck = ctl.get();
      if (! isRunning(recheck) && remove(command))
        reject(command);
      // 如果工作线程都down了
      else if (workerCountOf(recheck) == 0)
        addWorker(null, false);
    }
    else if (!addWorker(command, false))
      reject(command);
  }

addWorker方法实现

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
      int c = ctl.get();
      int rs = runStateOf(c);
      // Check if queue empty only if necessary.
      if (rs >= SHUTDOWN &&
        ! (rs == SHUTDOWN &&
          firstTask == null &&
          ! workQueue.isEmpty()))
        return false;
      for (;;) {
        int wc = workerCountOf(c);
        if (wc >= CAPACITY ||
          wc >= (core ? corePoolSize : maximumPoolSize))
          return false;
        if (compareAndIncrementWorkerCount(c))
          break retry;
        c = ctl.get(); // Re-read ctl
        if (runStateOf(c) != rs)
          continue retry;
        // else CAS failed due to workerCount change; retry inner loop
      }
    }
    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) {
          // 如果添加成功，则启动该线程，执行Worker的run方法，Worker的run方法执行外部的runWorker(Worker)
          t.start();
          workerStarted = true;
        }
      }
    } finally {
      if (! workerStarted)
        addWorkerFailed(w);
    }
    return workerStarted;
  }

Worker类继承了AbstractQueuedSynchronizer获得了同步等待这样的功能。

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
  {
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;
    /** Thread this worker is running in. Null if factory fails. */
    final Thread thread;
    /** Initial task to run. Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;
    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
      setState(-1); // inhibit interrupts until runWorker
      this.firstTask = firstTask;
      this.thread = getThreadFactory().newThread(this);
    }
    /** Delegates main run loop to outer runWorker */
    public void run() {
      runWorker(this);
    }
    // Lock methods
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.
    protected boolean isHeldExclusively() {
      return getState() != 0;
    }
    protected boolean tryAcquire(int unused) {
      if (compareAndSetState(0, 1)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
      }
      return false;
    }
    protected boolean tryRelease(int unused) {
      setExclusiveOwnerThread(null);
      setState(0);
      return true;
    }
    public void lock()    { acquire(1); }
    public boolean tryLock() { return tryAcquire(1); }
    public void unlock()   { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
    void interruptIfStarted() {
      Thread t;
      if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
          t.interrupt();
        } catch (SecurityException ignore) {
        }
      }
    }

runWorker(Worker)是Worker的轮询执行逻辑，不断地从工作队列中获取任务并执行它们。Worker每次执行任务前需要进行lock，防止在执行任务时被interrupt。

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
      while (task != null || (task = getTask()) != null) {
        w.lock();
        // If pool is stopping, ensure thread is interrupted;
        // if not, ensure thread is not interrupted. This
        // requires a recheck in second case to deal with
        // shutdownNow race while clearing interrupt
        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);
    }
  }

ThreadPoolExecutor的submit方法中将Callable包装成FutureTask后交给execute方法。

FutureTask

FutureTask继承于Runnable和Future，FutureTask定义的几个状态为
NEW, 尚未执行
COMPLETING, 正在执行
NORMAL, 正常执行完成得到结果
EXCEPTIONAL, 执行抛出异常
CANCELLED， 执行被取消
INTERRUPTING，执行正在被中断
INTERRUPTED, 已经中断。

其中关键的get方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
      s = awaitDone(false, 0L);
    return report(s);
  }

先获取当前状态，如果还未执行完成并且正常，则进入等待结果流程。在awaitDone不断循环获取当前状态，如果没有结果，则将自己通过CAS的方式添加到等待链表的头部，如果设置了超时，则LockSupport.parkNanos到指定的时间。

static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
  }
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
      if (Thread.interrupted()) {
        removeWaiter(q);
        throw new InterruptedException();
      }
      int s = state;
      if (s > COMPLETING) {
        if (q != null)
          q.thread = null;
        return s;
      }
      else if (s == COMPLETING) // cannot time out yet
        Thread.yield();
      else if (q == null)
        q = new WaitNode();
      else if (!queued)
        queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                           q.next = waiters, q);
      else if (timed) {
        nanos = deadline - System.nanoTime();
        if (nanos <= 0L) {
          removeWaiter(q);
          return state;
        }
        LockSupport.parkNanos(this, nanos);
      }
      else
        LockSupport.park(this);
    }
  }

FutureTask的run方法是执行任务并设置结果的位置,首先判断当前状态是否为NEW并且将当前线程设置为执行线程，然后调用Callable的call获取结果后设置结果修改FutureTask状态。

public void run() {
    if (state != NEW ||
      !UNSAFE.compareAndSwapObject(this, runnerOffset,
                     null, Thread.currentThread()))
      return;
    try {
      Callable<V> c = callable;
      if (c != null && state == NEW) {
        V result;
        boolean ran;
        try {
          result = c.call();
          ran = true;
        } catch (Throwable ex) {
          result = null;
          ran = false;
          setException(ex);
        }
        if (ran)
          set(result);
      }
    } finally {
      // runner must be non-null until state is settled to
      // prevent concurrent calls to run()
      runner = null;
      // state must be re-read after nulling runner to prevent
      // leaked interrupts
      int s = state;
      if (s >= INTERRUPTING)
        handlePossibleCancellationInterrupt(s);
    }
  }

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

# java  # ThreadPoolExecutor  # 原理  # ThreadPoolExecutor深入理解  # ThreadPoolExecutor线程池原理及其execute方法(详解)  # java中Executor  # ExecutorService  # ThreadPoolExecutor详解  # 简单谈谈ThreadPoolExecutor线程池之submit方法  # java ThreadPoolExecutor 并发调用实例详解  # java ThreadPoolExecutor使用方法简单介绍  # Java ThreadPoolExecutor的参数深入理解  # java线程池对象ThreadPoolExecutor的深入讲解  # 池中  # 都是  # 几个  # 这两个  # 希望能  # 如果没有  # 还未  # 当我们  # 谢谢大家  # 设置为  # 经常会  # 必须先  # 创建一个  # 抛出  # 上做  # 则将  # 类中  # 新任务  # 逸出  # 正在运行 

相关文章： 高端建站如何打造兼具美学与转化的品牌官网？  免费公司网站制作软件,如何申请免费主页空间做自己的网站？  合肥做个网站多少钱,合肥本地有没有比较靠谱的交友平台？  山东网站制作公司有哪些,山东大源集团官网？  网站制作服务平台,有什么网站可以发布本地服务信息？  建站主机是什么？如何选择适合的建站主机？  深圳网站制作案例,网页的相关名词有哪些？  c# 在ASP.NET Core中管理和取消后台任务  建站之星导航配置指南：自助建站与SEO优化全解析  如何基于云服务器快速搭建网站及云盘系统？  广东企业建站网站优化与SEO营销核心策略指南  北京企业网站设计制作公司,北京铁路集团官方网站？  建站主机CVM配置优化、SEO策略与性能提升指南  如何在阿里云香港服务器快速搭建网站？  建站之星云端配置指南：模板选择与SEO优化一键生成  如何选购建站域名与空间？自助平台全解析  如何在万网主机上快速搭建网站？  建站VPS配置与SEO优化指南：关键词排名提升策略  如何自己制作一个网站链接,如何制作一个企业网站，建设网站的基本步骤有哪些？  如何在Windows 2008云服务器安全搭建网站？  浅谈Javascript中的Label语句  唐山网站制作公司有哪些,唐山找工作哪个网站最靠谱？  深圳网站制作设计招聘,关于服装设计的流行趋势，哪里的资料比较全面？  内网网站制作软件,内网的网站如何发布到外网？  个人摄影网站制作流程,摄影爱好者都去什么网站？  ,巨量百应是干嘛的？  ,制作一个手机app网站要多少钱？  如何在阿里云域名上完成建站全流程？  常州自助建站工具推荐：低成本搭建与模板选择技巧  黑客如何利用漏洞与弱口令入侵网站服务器？  宁波自助建站系统如何快速打造专业企业网站？  如何通过免费商城建站系统源码自定义网站主题与功能？  免费网站制作模板下载,除了易企秀之外还有什么H5平台可以制作H5长页面，最好是免费的？  手机网站制作与建设方案,手机网站如何建设？  三星网站视频制作教程下载,三星w23网页如何全屏？  如何打造高效商业网站？建站目的决定转化率  如何快速搭建高效WAP手机网站？  创业网站制作流程,创业网站可靠吗？  定制建站方案优化指南：企业官网开发与建站费用解析  如何通过多用户协作模板快速搭建高效企业网站？  青岛网站建设如何选择本地服务器？  武汉网站制作费用多少,在武汉武昌，建面100平方左右的房子，想装暖气片，费用大概是多少啊？  阿里云网站制作公司,阿里云快速搭建网站好用吗？  c# 服务器GC和工作站GC的区别和设置  制作网页的网站有哪些,电脑上怎么做网页？  如何挑选优质建站一级代理提升网站排名？  小米网站链接制作教程,请问miui新增网页链接调用服务有什么用啊？  如何在腾讯云服务器快速搭建个人网站？  制作表格网站有哪些,线上表格怎么弄？  专业公司网站制作公司,用什么语言做企业网站比较好？