欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

java 中ThreadPoolExecutor原理分析

程序员文章站 2024-03-02 16:30:46
java 中threadpoolexecutor原理分析 线程池简介 java线程池是开发中常用的工具,当我们有异步、并行的任务要处理时,经常会用到线程池,或者在实...

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);
    }
  }

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