Java 동시 프로그래밍 루프 장벽 CyclicBarrier 예제 분석-java지도 시간-php.cn

Java 동시 프로그래밍 루프 장벽 CyclicBarrier 예제 분석

WBOY

풀어 주다： 2023-05-18 23:19:27

앞으로

761명이 탐색했습니다.

CyclicBarrier

앞서 소개한 CountDownLatch는 스레드를 호출하는 조인 방식에 비해 여러 스레드의 동기화 문제를 해결하는데 많이 최적화되었습니다. 그러나 CountDownLatch의 카운터는 일회성입니다. 즉, 카운터 값이 0에 도달한 후 CountDownLatch의 wait 및 countdown 메서드를 호출하면 즉시 반환되므로 스레드 동기화 효과를 얻을 수 없습니다. 따라서 카운터를 재설정해야 하는 필요성을 충족하기 위해 JDK 개발팀에서는 CyclicBarrier 클래스를 제공하며 CyclicBarrier 클래스의 기능은 CountDownLatch의 기능에 국한되지 않습니다. 문자 그대로 이해하면 CyclicBarrier는 루프백 장벽을 의미합니다. 이를 통해 스레드 그룹이 모두 특정 상태에 도달한 다음 동시에 모두 실행할 수 있습니다. 여기서 루프백이라고 부르는 이유는 대기 중인 모든 스레드의 실행이 완료되고 CyclicBarrier의 상태가 재설정된 후에 재사용할 수 있기 때문입니다. Wait 메서드를 호출한 후 스레드가 차단되기 때문에 이를 장벽이라고 합니다. 이 차단 지점을 장벽 지점이라고 합니다. 모든 스레드가 Wait 메서드를 호출한 후 스레드는 장벽을 뚫고 계속해서 아래쪽으로 실행됩니다. 원리를 소개하기 전에 이해를 돕기 위해 몇 가지 예를 소개하겠습니다. 다음 예에서 우리가 달성하고자 하는 것은 두 개의 스레드를 사용하여 분해된 작업 A를 실행한 다음 두 스레드가 작업을 완료한 후 결과를 요약하는 것입니다.

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 객체를 생성하는데, 첫 번째 매개변수는 카운터의 초기값이고 두 번째 숫자인 Runable은 카운트 값이 0일 때 수행해야 하는 작업입니다. 기본 함수에서는 크기 2의 스레드 풀이 먼저 생성됩니다. 스레드 풀에 두 개의 하위 작업을 추가하면 각 하위 작업은 자체 논리를 완료한 후 이 메서드를 호출합니다. 처음에는 카운터 값이 2입니다. 첫 번째 스레드가 Wait 메서드를 호출하면 카운터 값이 1로 감소합니다. 이때 카운터 값이 0이 아니므로 현재 스레드는 장벽 지점에 도달하여 차단됩니다. 그런 다음 두 번째 스레드가 대기를 호출하면 장벽에 들어가고 카운터 값도 감소합니다. 이제 카운터 값은 0입니다. 이때 실행이 완료된 후 CyclicBarrier 생성자에서 작업을 실행합니다. 장벽 지점을 종료하고 차단된 두 번째 스레드를 깨울 것입니다. 이때 첫 번째 스레드도 장벽 지점을 벗어나 계속 아래쪽으로 실행됩니다.

위의 예는 여러 스레드가 서로를 기다리고 있음을 보여줍니다. 카운터 값이 N이면 이후에 Wait 메서드를 호출하는 N1 스레드는 N번째 스레드 호출이 대기할 때 차단됩니다. 카운터 값이 0에 도달하면 N번째 스레드는 이전 N1 스레드를 깨우라는 알림을 발행합니다. 즉, 모든 스레드가 장벽 지점에 도달하면 함께 아래쪽으로 계속 실행할 수 있습니다. 이 예에서는 CountDownLatch를 사용하여 유사한 출력을 얻을 수 있습니다. CyclicBarrier의 재사용성을 설명하기 위해 또 다른 예를 들어보겠습니다.

작업이 1단계, 2단계, 3단계로 구성되어 있다고 가정합니다. 각 스레드는 1단계, 2단계 및 3단계를 순차적으로 실행해야 합니다. 여러 스레드가 작업을 실행할 때 모든 스레드의 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 예제 분석