Detaillierte Analyse des Prinzips von ThreadPoolExecutor in Java (mit Code)

Dieser Artikel stellt hauptsächlich relevante Informationen zur Prinzipanalyse von ThreadPoolExecutor in Java vor. Freunde, die sie benötigen, können sich auf

Prinzipielle Analyse von ThreadPoolExecutor in Java

Einführung in den Thread-Pool

Der Java-Thread-Pool ist ein häufig verwendetes Tool in der Entwicklung. Wenn wir asynchrone und parallele Aufgaben verarbeiten müssen, wird der Thread-Pool häufig verwendet . Oder wenn Sie einen Server implementieren, müssen Sie auch einen Thread-Pool verwenden, um

Verbindungsverarbeitungsanfragen zu empfangen.

Thread-Pool verwendet

Die im JDK bereitgestellte Thread-Pool-Implementierung befindet sich unter

java.util.concurrent.ThreadPoolExecutor. Bei Verwendung wird normalerweise die ExecutorServiceSchnittstelle verwendet, die gängige Methoden wie „submit“, „invokeAll“ und „shutdown“ bereitstellt.

In Bezug auf die Thread-Pool-Konfiguration stellt die Executors-Klasse einige

statische Methoden bereit, die Thread-Pools für einige gängige Szenarien bereitstellen können, wie z. B. newFixedThreadPool, newCachedThreadPool, newSingleThreadExecutor usw., diese Methoden rufen schließlich den Konstruktor von ThreadPoolExecutor auf.

Der Konstruktor von ThreadPoolExecutor, der alle Parameter enthält, ist

/**
   * @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 legt die Anzahl der Kernthreads im Thread-Pool fest, wenn die Anzahl der threads kleiner als corePoolSize ist, wird ein neuer Thread erstellt, um die Aufgabe auszuführen, unabhängig davon, ob derzeit inaktive Threads vorhanden sind.

• maximunPoolSize ist die maximal zulässige Anzahl von Threads im Thread-Pool

• workQueue wird zum Speichern von Aufgaben in der Warteschlange verwendet

• keepAliveTime ist die Leerlaufzeit für Threads, die größer als corePoolSize ist

• Der Handler wird für die Aufgabenverarbeitung verwendet, wenn die Aufgabe entkommt und der Thread-Pool geschlossen wird. Die Thread-Wachstumsstrategie von Der Thread-Pool lautet: Wenn die aktuelle Anzahl der Threads kleiner als corePoolSize ist, fügt

  einen neuen Thread hinzu. Wenn die Anzahl der Threads = corePoolSize und corePoolSize ist, werden neue Threads nur erstellt, wenn die workQueue keine neuen Aufgaben speichern kann . Überschüssige Threads werden nach der Leerlaufzeit von keepAliveTime zerstört.
  

Implementierung (basierend auf JDK1.8)

Der in ThreadPoolExecutor gespeicherte Status ist

Aktueller Thread-Pool-Status, einschließlich RUNNING, SHUTDOWN, STOP, TIDYING, TERMINATED.

Die aktuelle effektive Anzahl laufender Threads.

Fügen Sie diese beiden Zustände in eine int-Variable ein. Die ersten drei Ziffern sind der Thread-Pool-Status und die letzten 29 Ziffern sind die Anzahl der Threads.

Zum Beispiel steht 0b11100000000000000000000000000001 für RUNNING, einen Thread.

Speichern Sie den Worker-Satz über HashSet, müssen Sie zunächst den geschützten mainLock:ReentrantLock erhalten

Senden, ausführen

Die Ausführungsmethode der Ausführung besteht darin, zunächst die aktuelle Anzahl der Worker zu überprüfen. Wenn sie kleiner als corePoolSize ist, versuchen Sie, einen Core-Worker hinzuzufügen. Der Thread-Pool führt viele Tests durch, um die Anzahl der Threads aufrechtzuerhalten und den Status zu überprüfen.

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-Methodenimplementierung

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

Die Worker-Klasse erbt AbstractQueuedSynchronizer, um die Funktion des synchronen Wartens zu erhalten.

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) ist die Polling-Ausführungslogik von Worker, die kontinuierlich Aufgaben aus der Arbeitswarteschlange abruft und diese ausführt. Der Worker muss vor der Ausführung jeder Aufgabe gesperrt werden, um zu verhindern, dass er während der Ausführung der Aufgabe unterbrochen wird.

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

In der Submit-Methode von ThreadPoolExecutor wird das Callable in eine FutureTask gepackt und dann an die Execute-Methode übergeben.

FutureTask

FutureTask erbt von Runnable und Future. Die verschiedenen von FutureTask definierten Zustände sind

NEU, noch nicht ausgeführt,
ABGESCHLOSSEN, und Ausführen
NORMAL, normale Ausführung wird abgeschlossen und das Ergebnis wird erhalten
AUßERGEWÖHNLICH, Ausführung
wirft eine Ausnahme ausABGEBROCHEN, Ausführung wird abgebrochen
UNTERBRECHT, Ausführung wird unterbrochen
INTERRUPTED, wurde unterbrochen.

Die Schlüsselabrufmethode

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

ruft zunächst den aktuellen Status ab. Wenn die Ausführung nicht abgeschlossen ist und normal ist, tritt sie in den Warteergebnisprozess ein. Kontinuierliche

-Schleifen in waitingDone, um den aktuellen Status zu erhalten. Wenn kein Ergebnis vorliegt, wird es über CAS zum Anfang der Warteliste hinzugefügt. Wenn ein Timeout festgelegt ist, wird LockSupport.parkNanos auf die angegebene Zeit gesetzt.

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

Die Ausführungsmethode von FutureTask besteht darin, die Aufgabe auszuführen und die Position des Ergebnisses festzulegen. Bestimmen Sie zunächst, ob der aktuelle Status NEU ist, und legen Sie den aktuellen Thread als Ausführungsthread fest Rufen Sie auf, um das Ergebnis zu erhalten und das Ergebnis festzulegen, um den FutureTask-Status zu ändern.

Das obige ist der detaillierte Inhalt vonDetaillierte Analyse des Prinzips von ThreadPoolExecutor in Java (mit Code). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!