欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

java 中ThreadPoolExecutor原理分析

 更新時(shí)間:2017年03月29日 08:45:57   作者:劉正陽  
這篇文章主要介紹了java 中ThreadPoolExecutor原理分析的相關(guān)資料,需要的朋友可以參考下

java 中ThreadPoolExecutor原理分析

線程池簡介

Java線程池是開發(fā)中常用的工具,當(dāng)我們有異步、并行的任務(wù)要處理時(shí),經(jīng)常會(huì)用到線程池,或者在實(shí)現(xiàn)一個(gè)服務(wù)器時(shí),也需要使用線程池來接收連接處理請(qǐng)求。

線程池使用

JDK中提供的線程池實(shí)現(xiàn)位于java.util.concurrent.ThreadPoolExecutor。在使用時(shí),通常使用ExecutorService接口,它提供了submit,invokeAll,shutdown等通用的方法。

在線程池配置方面,Executors類中提供了一些靜態(tài)方法能夠提供一些常用場(chǎng)景的線程池,如newFixedThreadPool,newCachedThreadPool,newSingleThreadExecutor等,這些方法最終都是調(diào)用到了ThreadPoolExecutor的構(gòu)造函數(shù)。

ThreadPoolExecutor的包含所有參數(shù)的構(gòu)造函數(shù)是

/**
   * @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設(shè)置線程池的核心線程數(shù),當(dāng)添加新任務(wù)時(shí),如果線程池中的線程數(shù)小于corePoolSize,則不管當(dāng)前是否有線程閑置,都會(huì)創(chuàng)建一個(gè)新的線程來執(zhí)行任務(wù)。
  • maximunPoolSize是線程池中允許的最大的線程數(shù)
  • workQueue用于存放排隊(duì)的任務(wù)
  • keepAliveTime是大于corePoolSize的線程閑置的超時(shí)時(shí)間
  • handler用于在任務(wù)逸出、線程池關(guān)閉時(shí)的任務(wù)處理 ,線程池的線程增長策略為,當(dāng)前線程數(shù)小于corePoolSize時(shí),新增線程,當(dāng)線程數(shù)=corePoolSize且corePoolSize時(shí),只有在workQueue不能存放新的任務(wù)時(shí)創(chuàng)建新線程,超出的線程在閑置keepAliveTime后銷毀。

實(shí)現(xiàn)(基于JDK1.8)

ThreadPoolExecutor中保存的狀態(tài)有

當(dāng)前線程池狀態(tài), 包括RUNNING,SHUTDOWN,STOP,TIDYING,TERMINATED。

當(dāng)前有效的運(yùn)行線程的數(shù)量。

將這兩個(gè)狀態(tài)放到一個(gè)int變量中,前三位作為線程池狀態(tài),后29位作為線程數(shù)量。

例如0b11100000000000000000000000000001, 表示RUNNING, 一個(gè)線程。

通過HashSet來存儲(chǔ)工作者集合,訪問該HashSet前必須先獲取保護(hù)狀態(tài)的mainLock:ReentrantLock

submit、execute

execute的執(zhí)行方式為,首先檢查當(dāng)前worker數(shù)量,如果小于corePoolSize,則嘗試add一個(gè)core Worker。線程池在維護(hù)線程數(shù)量以及狀態(tài)檢查上做了大量檢測(cè)。

public void execute(Runnable command) {
    int c = ctl.get();
    // 如果當(dāng)期數(shù)量小于corePoolSize
    if (workerCountOf(c) < corePoolSize) {
      // 嘗試增加worker
      if (addWorker(command, true))
        return;
      c = ctl.get();
    }
    // 如果線程池正在運(yùn)行并且成功添加到工作隊(duì)列中
    if (isRunning(c) && workQueue.offer(command)) {
      // 再次檢查狀態(tài),如果已經(jīng)關(guān)閉則執(zhí)行拒絕處理
      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方法實(shí)現(xiàn)

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) {
          // 如果添加成功,則啟動(dòng)該線程,執(zhí)行Worker的run方法,Worker的run方法執(zhí)行外部的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的輪詢執(zhí)行邏輯,不斷地從工作隊(duì)列中獲取任務(wù)并執(zhí)行它們。Worker每次執(zhí)行任務(wù)前需要進(jìn)行l(wèi)ock,防止在執(zhí)行任務(wù)時(shí)被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,F(xiàn)utureTask定義的幾個(gè)狀態(tài)為
NEW, 尚未執(zhí)行
COMPLETING, 正在執(zhí)行
NORMAL, 正常執(zhí)行完成得到結(jié)果
EXCEPTIONAL, 執(zhí)行拋出異常
CANCELLED, 執(zhí)行被取消
INTERRUPTING,執(zhí)行正在被中斷
INTERRUPTED, 已經(jīng)中斷。

其中關(guān)鍵的get方法

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

先獲取當(dāng)前狀態(tài),如果還未執(zhí)行完成并且正常,則進(jìn)入等待結(jié)果流程。在awaitDone不斷循環(huán)獲取當(dāng)前狀態(tài),如果沒有結(jié)果,則將自己通過CAS的方式添加到等待鏈表的頭部,如果設(shè)置了超時(shí),則LockSupport.parkNanos到指定的時(shí)間。

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方法是執(zhí)行任務(wù)并設(shè)置結(jié)果的位置,首先判斷當(dāng)前狀態(tài)是否為NEW并且將當(dāng)前線程設(shè)置為執(zhí)行線程,然后調(diào)用Callable的call獲取結(jié)果后設(shè)置結(jié)果修改FutureTask狀態(tài)。

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

感謝閱讀,希望能幫助到大家,謝謝大家對(duì)本站的支持!

相關(guān)文章

  • java文件操作工具類實(shí)現(xiàn)復(fù)制文件和文件合并

    java文件操作工具類實(shí)現(xiàn)復(fù)制文件和文件合并

    這篇文章主要介紹了java文件操作工具類,類實(shí)現(xiàn)了復(fù)制文件和文件合并的功能,需要的朋友可以參考下
    2014-03-03
  • 解讀Spring事務(wù)是如何實(shí)現(xiàn)的

    解讀Spring事務(wù)是如何實(shí)現(xiàn)的

    這篇文章主要介紹了Spring事務(wù)是如何實(shí)現(xiàn)的問題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2023-03-03
  • MyBatis-Spring配置的講解

    MyBatis-Spring配置的講解

    今天小編就為大家分享一篇關(guān)于MyBatis-Spring配置的講解,小編覺得內(nèi)容挺不錯(cuò)的,現(xiàn)在分享給大家,具有很好的參考價(jià)值,需要的朋友一起跟隨小編來看看吧
    2018-12-12
  • Java基礎(chǔ)知識(shí)匯總

    Java基礎(chǔ)知識(shí)匯總

    這篇文章對(duì)Java編程語言的基礎(chǔ)知識(shí)作了一個(gè)較為全面的匯總,在這里給大家分享一下。需要的朋友可以參考。
    2017-09-09
  • SpringBoot項(xiàng)目鑒權(quán)的4種方式小結(jié)

    SpringBoot項(xiàng)目鑒權(quán)的4種方式小結(jié)

    本文主要介紹了SpringBoot項(xiàng)目鑒權(quán)的4種方式小結(jié),包括傳統(tǒng)AOP、攔截器、參數(shù)解析器和過濾器,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下
    2021-12-12
  • Spring外部化配置的幾種技巧分享

    Spring外部化配置的幾種技巧分享

    在油管上看了龍之春的一個(gè)Spring tips 視頻,講述Spring外部化配置的幾種技巧,收獲頗多,想拿出來給大家分享下。對(duì)spring感興趣的朋友可以了解下本文
    2021-06-06
  • SpringCloud Gateway網(wǎng)關(guān)功能介紹與使用

    SpringCloud Gateway網(wǎng)關(guān)功能介紹與使用

    SpringCloud Gateway 是 Spring Cloud 的一個(gè)全新項(xiàng)目,它旨在為微服務(wù)架構(gòu)提供一種簡單有效的統(tǒng)一的 API 路由管理方式。這篇文章主要介紹了SpringCloud Gateway網(wǎng)關(guān)作用,需要的朋友可以參考下
    2022-12-12
  • IDEA巧用Postfix Completion讓碼速起飛(小技巧)

    IDEA巧用Postfix Completion讓碼速起飛(小技巧)

    這篇文章主要介紹了IDEA巧用Postfix Completion讓碼速起飛,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2020-08-08
  • JavaSE基礎(chǔ)之反射機(jī)制(反射Class)詳解

    JavaSE基礎(chǔ)之反射機(jī)制(反射Class)詳解

    反射機(jī)制有什么用?通過java語言中的反射機(jī)制可以操作字節(jié)碼文件,可以讀和修改字節(jié)碼文件。所以本文將為大家講講反射機(jī)制的使用,需要的可以參考一下
    2022-09-09
  • SpringBoot+Vue+Element-ui實(shí)現(xiàn)前后端分離

    SpringBoot+Vue+Element-ui實(shí)現(xiàn)前后端分離

    使用前后端分離的方式,可以減少代碼耦合,本文主要介紹了SpringBoot+Vue+Element-ui實(shí)現(xiàn)前后端分離,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
    2023-06-06

最新評(píng)論