Java 同時プログラミング ループ バリア CyclicBarrier の例の分析

CyclicBarrier

以前に紹介した CountDownLatch は、スレッド呼び出しの join メソッドと比較して、複数のスレッドの同期を解決するために大幅に最適化されました。ただし、CountDownLatch のカウンタは 1 回限りです。つまり、カウンタ値が 0 に達した後、CountDownLatch の await メソッドと countdown メソッドを呼び出すとすぐに戻り、スレッド同期の効果は得られません。したがって、カウンタをリセットしたいというニーズを満たすために、JDK 開発チームは CyclicBarrier クラスを提供しており、CyclicBarrier クラスの機能は CountDownLatch の機能に限定されません。文字通り理解すると、CyclicBarrier はループ バリアを意味し、スレッドのグループがすべてある状態に到達し、それらをすべて同時に実行できるようにします。これがループバックと呼ばれる理由は、待機中のすべてのスレッドの実行が終了し、CyclicBarrier の状態がリセットされた後に再利用できるためです。 await メソッドを呼び出した後にスレッドがブロックされるため、これはバリアと呼ばれます。このブロック ポイントはバリア ポイントと呼ばれます。すべてのスレッドが await メソッドを呼び出した後、スレッドはバリアを突破して下方向に実行され続けます。原則を紹介する前に、理解を深めるためにいくつかの例を紹介します。次の例で実現したいのは、2 つのスレッドを使用して分解されたタスク A を実行し、2 つのスレッドがタスクを完了した後にその結果を集約することです。

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest {

   
   //创建一个线程数固定为2的线程池
   private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
       @Override
       public void run() {
           System.out.println(Thread.currentThread() + " task1 merge result");
       }
   });

   public static void main(String[] args) throws InterruptedException{
       ExecutorService executorService = Executors.newFixedThreadPool(2);

       
       //添加线程A到线程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task1");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //添加线程B到线程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task2");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //关闭线程池
       executorService.shutdown();

   }
}

上記のコードは CyclicBarrier オブジェクトを作成します。その最初のパラメーターはカウンターの初期値で、2 番目の数値 Runable は、カウント値は 0 です。 main 関数では、最初にサイズ 2 のスレッド プールが作成されます。 2 つのサブタスクをスレッド プールに追加すると、各サブタスクは独自のロジックを完了した後にこのメソッドを呼び出します。最初のカウンタ値は 2 です。最初のスレッドが await メソッドを呼び出すと、カウンタ値は 1 にデクリメントされます。この時点ではカウンタ値が 0 ではないため、現在のスレッドはバリア ポイントに到達し、ブロックされます。次に、2 番目のスレッドが await を呼び出すと、バリアに入り、カウンター値もデクリメントされます。現在、カウンター値は 0 です。この時点で、CyclicBarrier コンストラクターでタスクを実行します。実行が完了すると、バリアポイントを出て、ブロックされている 2 番目のスレッドを起動します。このとき、最初のスレッドもバリア ポイントを出て下に向かって走り続けます。

上記の例は、複数のスレッドが互いに待機していることを示しています。カウンター値が N の場合、その後 await メソッドを呼び出す N1 個のスレッドは、バリア ポイントに到達したためブロックされます。N 番目のスレッドが到達すると、await メソッドを呼び出します。 await を呼び出した後、カウンター値は 0 になります。この時点で、N 番目のスレッドは、前の N1 スレッドをウェイクアップする通知を送信します。つまり、すべてのスレッドがバリア ポイントに到達すると、一緒に下方向に実行を続けることができます。この例では、CountDownLatch を使用して同様の出力を実現できます。 CyclicBarrier の再利用性を説明するために別の例を示します。

タスクがフェーズ 1、フェーズ 2、フェーズ 3 で構成されているとします。各スレッドはフェーズ 1、フェーズ 2、フェーズ 3 を順番に実行する必要があります。複数のスレッドがタスクを実行する場合、すべてのフェーズ 2 の実行に入ることができるだけですすべてのスレッドのフェーズ 1 が完了した後、フェーズ 3 の実行はすべてのスレッドのフェーズ 2 が完了した後にのみ開始できます。 CyclicBarrier は、この要件を完了するために以下で使用されます。

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest1 {

    //创建一个线程数固定为2的线程池
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);

    public static void main(String[] args) throws InterruptedException{
        ExecutorService executorService = Executors.newFixedThreadPool(2);


        //添加线程A到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //添加线程B到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //关闭线程池
        executorService.shutdown();

    }
}

上記のコードでは、各サブスレッドはフェーズ 1 の実行後に await メソッドを呼び出します。すべてのスレッドがバリア ポイントに到達するまで待機してから、一緒に実行されます。すべてのスレッドがフェーズ 2 の実行を開始する前にフェーズ 1 を完了していること。

以上がJava 同時プログラミング ループ バリア CyclicBarrier の例の分析の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。