Java 同時プログラミングループバリア CyclicBarrier の例の分析-＆＃＆チュートリアル-php.cn

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

WBOY

リリース： 2023-05-18 23:19:27

転載

705 人が閲覧しました

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

   }
}

ログイン後にコピー

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

上記のコードは 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();

    }
}

ログイン後にコピー

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