最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

java 中ThreadPoolExecutor原理分析

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

java 中ThreadPoolExecutor原理分析

線(xiàn)程池簡(jiǎn)介

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

線(xiàn)程池使用

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

在線(xiàn)程池配置方面,Executors類(lèi)中提供了一些靜態(tài)方法能夠提供一些常用場(chǎng)景的線(xiàn)程池,如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è)置線(xiàn)程池的核心線(xiàn)程數(shù),當(dāng)添加新任務(wù)時(shí),如果線(xiàn)程池中的線(xiàn)程數(shù)小于corePoolSize,則不管當(dāng)前是否有線(xiàn)程閑置,都會(huì)創(chuàng)建一個(gè)新的線(xiàn)程來(lái)執(zhí)行任務(wù)。
  • maximunPoolSize是線(xiàn)程池中允許的最大的線(xiàn)程數(shù)
  • workQueue用于存放排隊(duì)的任務(wù)
  • keepAliveTime是大于corePoolSize的線(xiàn)程閑置的超時(shí)時(shí)間
  • handler用于在任務(wù)逸出、線(xiàn)程池關(guān)閉時(shí)的任務(wù)處理 ,線(xiàn)程池的線(xiàn)程增長(zhǎng)策略為,當(dāng)前線(xiàn)程數(shù)小于corePoolSize時(shí),新增線(xiàn)程,當(dāng)線(xiàn)程數(shù)=corePoolSize且corePoolSize時(shí),只有在workQueue不能存放新的任務(wù)時(shí)創(chuàng)建新線(xiàn)程,超出的線(xiàn)程在閑置keepAliveTime后銷(xiāo)毀。

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

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

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

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

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

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

通過(guò)HashSet來(lái)存儲(chǔ)工作者集合,訪(fǎng)問(wèn)該HashSet前必須先獲取保護(hù)狀態(tài)的mainLock:ReentrantLock

submit、execute

execute的執(zhí)行方式為,首先檢查當(dāng)前worker數(shù)量,如果小于corePoolSize,則嘗試add一個(gè)core Worker。線(xiàn)程池在維護(hù)線(xiàn)程數(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();
    }
    // 如果線(xiàn)程池正在運(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);
      // 如果工作線(xiàn)程都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)該線(xiàn)程,執(zhí)行Worker的run方法,Worker的run方法執(zhí)行外部的runWorker(Worker)
          t.start();
          workerStarted = true;
        }
      }
    } finally {
      if (! workerStarted)
        addWorkerFailed(w);
    }
    return workerStarted;
  }

Worker類(lèi)繼承了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的輪詢(xún)執(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),如果沒(méi)有結(jié)果,則將自己通過(guò)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)前線(xiàn)程設(shè)置為執(zhí)行線(xiàn)程,然后調(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)文章

最新評(píng)論

青阳县| 建始县| 潮安县| 高阳县| 淮南市| 肇源县| 祁门县| 蓝山县| 确山县| 舒兰市| 闵行区| 监利县| 安泽县| 盘锦市| 大同市| 壶关县| 资源县| 渝中区| 西藏| 青海省| 云梦县| 丰顺县| 张家界市| 泾源县| 崇左市| 获嘉县| 祥云县| 景德镇市| 读书| 重庆市| 营口市| 吉隆县| 攀枝花市| 白城市| 大关县| 辽源市| 三亚市| 望都县| 锡林郭勒盟| 孟州市| 葵青区|