How to use Java multi-thread synchronization tool class CyclicBarrier

What is CyclicBarrier

CyclicBarrier is a synchronization tool class provided in the Java concurrency package, which allows multiple threads to wait at a certain barrier , execution will not continue until all threads reach the barrier. The implementation principle of CyclicBarrier is based on ReentrantLock and Condition, and the waiting and waking up of threads are realized by calling the await() method multiple times.

Basic usage of CyclicBarrier

The basic usage of CyclicBarrier is very simple. You only need to create a CyclicBarrier object and set the counter value to the number of waiting threads. After each thread finishes executing, call the await() method of CyclicBarrier to wait for other threads to finish executing. When all threads reach the barrier, the barrier will be opened and all threads will continue to execute.

Source code implementation of CyclicBarrier

The implementation principle of CyclicBarrier is based on ReentrantLock and Condition, and the waiting and waking up of threads are realized by calling the await() method multiple times. The source code implementation of CyclicBarrier mainly includes two parts: initialization of the barrier and waiting and waking up of the barrier.

(1) Initialization of CyclicBarrier

When creating a CyclicBarrier object, you need to specify the number of waiting threads and the execution operation of the barrier. The construction method of the CyclicBarrier object is as follows:

public CyclicBarrier(int parties, Runnable barrierAction)

Among them, parties represents the number of waiting threads, and barrierAction represents the operation performed by the barrier.

In the construction method, a parties-sized ReentrantLock array and a Condition object will be created based on the number of waiting threads. The ReentrantLock array is used to ensure that multiple threads can reach the barrier and wait at the same time, and the Condition object is used to wait and wake up threads.

(2) Waiting and waking up of CyclicBarrier

When the thread executes the await() method, it will first try to acquire the lock of the ReentrantLock object. If the acquisition fails, the thread will be added to the waiting queue. Waiting for the lock to be released. After acquiring the lock, the thread will determine whether the current counter has reached the number of waiting threads. If so, it will perform barrier operations and reset the counter to parties, waking up all threads in the waiting queue. If the counter does not reach the number of waiting threads, the thread will be added to the waiting queue to wait for the arrival of other threads.

The source code of CyclicBarrier's await() method is as follows:

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        // 获取锁
        lock.lock();
        // 计数器减1
        int index = --count;
        if (index == 0) {
            // 如果计数器为0，执行屏障操作并唤醒等待队列中的所有线程
            final Runnable command = barrierCommand;
            if (command != null) {
                command.run();
            }
            next
        trip.signalAll();
    } else {
        try {
            // 等待其他线程到达屏障处
            int phase = generation;
            trip.await();
            // 如果是最后一个到达屏障的线程，执行屏障操作并唤醒等待队列中的所有线程
            if (phase == generation) {
                command = barrierCommand;
                if (command != null) {
                    command.run();
                }
            }
            // 计数器重置
            nextGeneration();
        } catch (InterruptedException ie) {
            // 如果线程在等待时被中断，抛出InterruptedException异常
            cancel();
            throw ie;
        } catch (BrokenBarrierException bbe) {
            // 如果屏障被破坏，抛出BrokenBarrierException异常
            broken = true;
            trip = new Condition[parties];
            throw bbe;
        }
    }
    return index;
} finally {
    // 释放锁
    lock.unlock();
}
}

In the await() method of CyclicBarrier, first acquire the lock of the ReentrantLock object and decrement the counter by 1. If the counter is 0, perform the barrier operation and wake up all threads in the waiting queue. If the counter is not 0, wait for other threads to arrive at the barrier.

During the waiting process, if the thread is interrupted, an InterruptedException will be thrown. If the barrier is broken, a BrokenBarrierException will be thrown. If it is the last thread to reach the barrier, the barrier operation will be performed and all threads in the waiting queue will be woken up and the counter will be reset to parties.

Usage scenarios of CyclicBarrier

CyclicBarrier is suitable for scenarios where multiple threads need to wait for each other to reach a certain barrier point before continuing to execute. For example, if multiple threads need to execute a certain task at the same time, but a certain task needs to wait for other tasks to complete before it can continue to execute, then you can use CyclicBarrier to achieve thread synchronization and collaboration.

In addition, CyclicBarrier can also be used to implement pipeline processing. For example, in the producer-consumer model, multiple producers can add data to the queue at the same time. When the queue is full, all producers need to wait for consumption. After the user has finished processing the data, continue adding data.

The above is the detailed content of How to use Java multi-thread synchronization tool class CyclicBarrier. For more information, please follow other related articles on the PHP Chinese website!