Analyse détaillée du principe de ThreadPoolExecutor en java (avec code)

Cet article présente principalement des informations pertinentes sur l'analyse des principes de ThreadPoolExecutor en Java. Les amis qui en ont besoin peuvent se référer à

Analyse des principes de ThreadPoolExecutor en Java

<.> Introduction au pool de threads

Le pool de threads Java est un outil couramment utilisé en développement Lorsque nous avons des tâches asynchrones et parallèles à traiter, le pool de threads est souvent utilisé. . Ou lors de l'implémentation d'un serveur, vous devez également utiliser un pool de threads pour recevoir les

traitements de connexion requêtes.

Le pool de threads utilise

L'implémentation du pool de threads fournie dans le JDK se trouve à

java.util.concurrent.ThreadPoolExecutor. Lorsqu'elle est utilisée, l'ExecutorServiceinterface est généralement utilisée, qui fournit des méthodes courantes telles que submit, EnsureAll et shutdown.

En termes de configuration du pool de threads, la classe Executors fournit des méthodes

static qui peuvent fournir des pools de threads pour certains scénarios courants, tels que newFixedThreadPool, newCachedThreadPool, newSingleThreadExecutor etc., ces méthodes finissent par appeler le constructeur de ThreadPoolExecutor.

Le constructeur de ThreadPoolExecutor contenant tous les paramètres est

/**
   * @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 définit le nombre de threads principaux dans le pool de threads lors de l'ajout d'une nouvelle tâche, si si le nombre de. threads est inférieur à corePoolSize, un nouveau thread sera créé pour effectuer la tâche, qu'il existe ou non des threads actuellement inactifs.

• maximunPoolSize est le nombre maximum de threads autorisés dans le pool de threads

• workQueue est utilisé pour stocker les tâches en file d'attente

• keepAliveTime est le délai d'inactivité pour les threads supérieur à corePoolSize

• le gestionnaire est utilisé pour le traitement des tâches lorsque la tâche s'échappe et que le pool de threads est fermé. le pool de threads est, Lorsque le nombre actuel de threads est inférieur à corePoolSize,

  ajoute un nouveau thread Lorsque le nombre de threads = corePoolSize et corePoolSize, de nouveaux threads ne seront créés que lorsque workQueue ne peut pas stocker de nouvelles tâches. . Les threads en excès seront détruits après un keepAliveTime inactif.
  

Implémentation (basée sur JDK1.8)

L'état enregistré dans ThreadPoolExecutor est

État actuel du pool de threads, y compris RUNNING, SHUTDOWN, STOP, TIDYING, TERMINATED.

Le nombre effectif actuel de threads en cours d'exécution.

Mettez ces deux états dans une variable int, les trois premiers chiffres sont l'état du pool de threads et les 29 derniers chiffres sont le nombre de threads.

Par exemple, 0b11100000000000000000000000000001, représente RUNNING, un thread.

Stockez l'ensemble de travailleurs via HashSet. Avant d'accéder au HashSet, vous devez d'abord obtenir le mainLock:ReentrantLock protégé

soumettre, exécuter
<🎜. >La méthode d'exécution de l'exécution consiste à vérifier d'abord le nombre actuel de travailleurs. S'il est inférieur à corePoolSize, essayez d'ajouter un travailleur principal. Le pool de threads effectue de nombreux tests sur le maintien du nombre de threads et la vérification de l'état.

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

Implémentation de la méthode addWorker

La classe Worker hérite de AbstractQueuedSynchronizer pour obtenir la fonction d'attente synchrone.

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

runWorker (Worker) est la logique d'exécution d'interrogation de Worker, qui obtient en continu les tâches de la file d'attente de travail et les exécute. Le Worker doit être verrouillé avant l'exécution de chaque tâche pour éviter qu'il ne soit interrompu lors de l'exécution de la tâche.

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

Dans la méthode submit de ThreadPoolExecutor, le Callable est empaqueté dans une FutureTask puis transmis à la méthode d'exécution.

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

FutureTask

FutureTask hérite de Runnable et Future Les différents états définis par FutureTask sont

NEW, pas encore exécutés,

COMPLETING, et exécution de
NORMAL, l'exécution normale est terminée et le résultat est obtenu
EXCEPTIONNEL, l'exécution
lève une exception
ANNULÉE, l'exécution est annuléeINTERRUPTION, l'exécution est interrompue
INTERRUPTED, a été interrompu.


La méthode key get

obtient d'abord l'état actuel si l'exécution n'est pas terminée et est normale, elle entre dans le processus de résultat en attente.

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

boucle

en continu dans waitDone pour obtenir l'état actuel, s'il n'y a pas de résultat, s'ajoute en tête de la liste d'attente via CAS. Si un délai d'attente est défini, LockSupport.parkNanos à l'heure spécifiée.

La méthode d'exécution de FutureTask consiste à exécuter la tâche et à définir la position du résultat. Tout d'abord, déterminez si l'état actuel est NOUVEAU et définissez le thread actuel comme thread d'exécution. appelez pour obtenir le résultat et définissez le résultat pour modifier l’état FutureTask.

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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!