Java のスレッド プールとエグゼキュータの原理の詳細な説明

この記事では、Java スレッド プールと Executor の原則の分析を詳しく説明する関連情報を主に紹介します。この部分の知識を必要とする人は参考にしてください。 Java スレッド プールと Executor の原理 分析

スレッド プールの機能と基礎知識

始める前に、まず「スレッド プール」の概念について説明します。 「スレッドプール」はその名の通り、スレッドキャッシュのことです。これは 1 つまたは複数のスレッドのコレクションであり、ユーザーは実行の詳細にあまり関与することなく、実行する必要があるタスクをスレッド プールにスローするだけで済みます。では、スレッド プールにはどのような機能があるのでしょうか?または、Thread を直接使用する場合と比較した利点は何ですか?以下の点を簡単にまとめました:

スレッドの作成と破棄による消費を削減する

Java Threadの実装については、以前のブログで分析しました。 Java Thread とカーネルスレッドは 1:1 (Linux) また、Thread は Java 層と C++ 層に多くのメンバーデータを持っているため、実際には Java Thread は比較的重いです。 Java スレッドの作成と破棄には、OS と JVM の両方で多くの作業が必要となるため、Java スレッドがキャッシュされていれば、ある程度の効率向上が達成できます。

コンピューティングリソース制御のより便利で透過的な実装

この項目について説明するには、いくつかの例を挙げる必要があるかもしれません。非常に有名な Web サーバー Nginx を例に挙げます。Nginx は、強力な同時実行機能と低いリソース消費で知られています。これらの厳しい要件を達成するために、Nginx はワーカー スレッドの数を厳しく制限しています (通常、ワーカー スレッドは CPU の数と同じです)。この設計の焦点は、スレッドの切り替えによって生じるパフォーマンスの損失を軽減することです。この最適化方法は Java にも適用できます。タスクごとに新しいスレッドを作成すると、最終的にはプログラム リソースの制御が難しくなり (特定の関数が CPU を占有する)、全体の実行速度が相対的に遅くなります。 Java スレッド プールには、スレッドの最大数を制御するために使用できる FixedThreadPool が用意されています。

上記では「ナンセンス」が多かったので、Java スレッド プールの実装に基づいて分析してみましょう。 Java のスレッド プールにはいくつかの実装があります:

キャッシュされたスレッド プール

キャッシュされたスレッド プールの特徴は、以前のスレッドをキャッシュし、新しく送信されたタスクはキャッシュされたスレッドで実行できることです。これにより、上記のことが達成されます。 。

固定 ThreadPool

cachedThreadPool の機能の 1 つは、新しく送信されたタスクを実行するためのアイドル スレッドがない場合、新しいスレッドが作成されることです。 FixedThreadPool はこれを行いません。タスクを保存し、アイドル状態のスレッドが存在するまで待機してから実行します。すなわち、上記第２の効果が得られる。

スケジュールされたThreadPool

スケジュールされたThreadPoolの特徴は、タスクの遅延実行や定期実行などのタスクのスケジューリングを実現できることです。

上記の 3 つに加えて、Java は Fork/Join フレームワークに基づいた newWorkStealingPool も実装します。これについてはまだ調べていないので、そのままにしておきます。 Java の同時実行サポートでは、Executor はさまざまなスレッド プールをパッケージ化するために使用されます。「Executor」という名前は実際には非常に適切です。

1.キャッシュされた ThreadPool と固定 ThreadPool の実装

前の説明からわかるように、これら 2 つのスレッド プールは非常に似ています。実際、これは実際に当てはまります。そうでない場合は、実際の例を見てみましょう:

ThreadPoolExecutor executor1 = (ThreadPoolExecutor)Executors.newCachedThreadPool();

これは、2 つのスレッド プールを作成する新しい方法です。とても似ている！そう思わないなら、私は真実を見せることしかできません。

ThreadPoolExecutor executor2 = (ThreadPoolExecutor)Executors.newFixedThreadPool(4);

はい、同じコンストラクターを呼び出しますが、パラメーターがわずかに異なるだけです。それでは、これらのパラメーターの意味と、2 つのパラメーター セットの違いを見てみましょう。まず、ThreadPoolExecutor のコンストラクターをポストする必要があります。

public static ExecutorService newCachedThreadPool() {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                 60L, TimeUnit.SECONDS,
                 new SynchronousQueue<Runnable>());
}

public static ExecutorService newFixedThreadPool(int nThreads) {
  return new ThreadPoolExecutor(nThreads, nThreads,
                 0L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<Runnable>());
}

新鮮に見せるために、別のレイヤーのコンストラクターは投稿しません。そのコンストラクターは単なる単純な割り当てです。ここの関数プロトタイプは、コメントと同様に、JDK コードの命名が非常に優れていると言わざるを得ません。

maximumPoolSize はスレッド プール内のスレッドの最大数です。キャッシュされた ThreadPool の場合、この値は Integer.MAX_VALUE であり、これは基本的に何十億ものスレッドを実行できるマシンの種類に相当します。 ！固定 ThreadPool の場合、この値はユーザーが設定したスレッド プールの数です。

keepAliveTime とユニットは、スレッドのキャッシュ有効期限を決定します。キャッシュされた ThreadPool の場合、スレッドのキャッシュ有効期限は 1 分です。つまり、ワーカー スレッドが 1 分間何もしない場合、ワーカー スレッドは保存のために取り消されます。リソース。固定 ThreadPool に渡される時間は 0 です。これは、固定 ThreadPool のワーカー スレッドが期限切れにならないことを意味します。

corePoolSize是线程池的最小线程数；对于cached ThreadPool，这个值为0，因为在完全没有任务的情况下，cached ThreadPool的确会成为“光杆司令”。至于fixed ThreadPool，这个fixed已经表明corePoolSize是等于线程总数的。
接下来，我们根据一个简单的使用例子，来看看一下cached ThreadPool的流程。

public class Task implements Callable<String> {

private String name;
public Task(String name) {
  this.name = name;
}
@Override
public String call() throws Exception {
  System.out.printf("%s: Starting at : %s\n", this.name, new Date());
  return "hello, world";
}
public static void main(String[] args) {
  ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors.newCachedThreadPool();
  Task task = new Task("test");
  Future<String> result = executor.submit(task);
  try {
    System.out.printf("%s\n", result.get());
  } catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
  }
  executor.shutdown();
  System.out.printf("Main ends at : %s\n", new Date());
}
}

首先，来看看executor.submit(task)，这其实调用了ThreadPoolExecutor.execute(Runnable command)方法，这个方法的代码如下，整段代码的逻辑是这样的。首先检查线程池的线程数是否不够corePoolSize，如果不够就直接新建线程并把command添加进去；如果线程数已经够了或者添加失败（多个线程增加添加的情况），就尝试把command添加到队列中（workQueue.offer(command)），如果添加失败了，就reject掉cmd。大体的逻辑是这样的，这段代码有很多基于线程安全的设计，这里为了不跑题，就先忽略细节了。

public void execute(Runnable command) {
  if (command == null)
    throw new NullPointerException();
  int c = ctl.get();
  if (workerCountOf(c) < corePoolSize) {
    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);
    else if (workerCountOf(recheck) == 0)
      addWorker(null, false);
  }
  else if (!addWorker(command, false))
    reject(command);
}

到这里，看起来线程池实现的整体思路其实也没多么复杂。但是还有一个问题——一个普通的Thread在执行完自己的run方法后会自动退出。那么线程池是如何实现Worker线程不断的干活，甚至在没有任务的时候。其实答案很简单，就是Worker其实在跑大循环，Worker实际运行方法如下：

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();
  /***/
      try {
        beforeExecute(wt, task);
        Throwable thrown = null;
        try {
          task.run();
        /***/
        } finally {
          afterExecute(task, thrown);
        }
      } finally {
        task = null;
        w.completedTasks++;
        w.unlock();
      }
    }
    completedAbruptly = false;
  } finally {
    processWorkerExit(w, completedAbruptly);
  }
}

关键就在这个while的判断条件，对于需要cached线程的情况下，getTask()会阻塞起来，如果缓存的时间过期，就会返回一个null，然后Worker就退出了，也就结束了它的服役周期。而在有任务的情况下，Woker会把task拿出来，然后调用task.run()执行任务，并通过Future通知客户线程（即future.get()返回）。这样一个简单的线程池使用过程就完了。。。

当然，线程池的很多精髓知识——基于线程安全的设计，我都没有分析。有兴趣可以自己分析一下，也可以和我讨论。此外Scheduled ThreadPool这里也没有分析，它的要点其实是调度，主要是根据时间最小堆来驱动的。

以上がJava のスレッド プールとエグゼキュータの原理の詳細な説明の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。