Java concurrent programming loop barrier CyclicBarrier example analysis

CyclicBarrier

The CountDownLatch introduced earlier has been optimized a lot in solving the synchronization of multiple threads compared to the join method of calling threads. However, the counter of CountDownLatch is one-time, that is, after the counter value reaches 0, calling the await and countdown methods of CountDownLatch will return immediately, which will not achieve the effect of thread synchronization. Therefore, in order to meet the need for the counter to be reset, the JDK development team provides the CyclicBarrier class, and the functions of the CyclicBarrier class are not limited to the functions of CountDownLatch. Literally understood, CyclicBarrier means a loop barrier, which allows a group of threads to all reach a state and then execute them all at the same time. The reason why it is called a loopback is that it can be reused after all waiting threads have finished executing and the state of the CyclicBarrier is reset. It is called a barrier because the thread will be blocked after calling the await method. This blocking point is called the barrier point. After all threads have called the await method, the threads will break through the barrier and continue running downward. Before introducing the principles, we will introduce a few examples to deepen our understanding. In the following example, what we want to achieve is to use two threads to execute a decomposed task A, and then aggregate their results after the two threads have completed their tasks.

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

   }
}

The above code creates a CyclicBarrier object, its first parameter is the initial value of the counter, and the second number Runable is the task that needs to be performed when the count value is 0 . In the main function, a thread pool of size 2 is first created. Add two subtasks to the thread pool, and each subtask will call this method after completing its own logic. At the beginning, the counter value is 2. When the first thread calls the await method, the counter value will be decremented to 1. Since the counter value is not 0 at this time, the current thread reaches the barrier point and is blocked. Then when the second thread calls await, it will enter the barrier and the counter value will also be decremented. Now the counter value is 0. At this time, it will execute the task in the CyclicBarrier constructor. After the execution is completed, it will exit the barrier point and wake up the blocked Second thread. At this time, the first thread will also exit the barrier point and continue running downward.

The above example illustrates that multiple threads are waiting for each other. If the counter value is N, then the N1 threads that subsequently call the await method will be blocked because they reach the barrier point. When the Nth thread After calling await, the counter value is 0. At this time, the Nth thread will send a notification to wake up the previous N1 threads. That is to say, when all threads reach the barrier point, they can continue to execute downward together. This example can achieve similar output by using CountDownLatch. Let's give another example to illustrate the reusability of CyclicBarrier.

Assume that a task consists of Phase 1, Phase 2 and Phase 3. Each thread must execute Phase 1, Phase 2 and Phase 3 serially. When multiple threads execute the task, all Phase 2 execution can only be entered after the phase 1 of all threads is completed, and phase 3 execution can be entered only after the phase 2 of all threads is completed. CyclicBarrier is used below to complete this requirement.

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

    }
}

In the above code, each sub-thread calls the await method after executing phase 1. It will wait until all threads reach the barrier point before executing together. This ensures that all threads have completed phase 1 before they start executing phase 2.

The above is the detailed content of Java concurrent programming loop barrier CyclicBarrier example analysis. For more information, please follow other related articles on the PHP Chinese website!