线程的并发工具类（3）---CyclicBarrier-蒲公英云

线程的并发工具类（3）---CyclicBarrier

朱雀 2023-05-28 15:45 342阅读 0赞

CyclicBarrier的用法

- - - 1、CyclicBarrier介绍
    - - 1.1、用法场景
        1.2、原理
    - 2、CyclicBarrier实例
    - 3、CyclicBarrier和CountDownLatch比较

1、CyclicBarrier介绍

CyclicBarrier,线程的同步工具类。

作用：让一组线程达到某个屏障，被阻塞，一直到组内最后一个线程达到屏障时，屏障开放，所有被阻塞的线程会继续运行。由自身(该组线程本身)去控制。

1.1、用法场景

实现多个线程的最大并行性。就像生活中我们会约朋友们到某个餐厅一起吃饭，有些朋友可能会早到，有些朋友可能会晚到，但是这个餐厅规定必须等到所有人到齐之后才会让我们进去。(由这组线程本身控制)

1.2、原理

在初始化CyclicBarrier实例的时候，使用new CyclicBarrier(N)初始化计数器为N。工作线程调用await()方法进行等待，等待N个线程都到达await()这里时，屏障开放,向下执行。如果初始化计数器使用new CyclicBarrier(N,new ThreadAfterWork())，则当N个线程都到达await()这里，屏障开放，会先依次执行ThreadAfterWork()这个线程实例。当所有的线程执行完ThreadAfterWork子线程后再继续向下执行。

2、CyclicBarrier实例

CyclicBarrier中主要使用了await()来实现它的功能，代码如下：

package com.ld.task;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
/** * CyclicBarrier:放行条件由一组线程本身决定 */
public class UseCyclicBarrier { 
    // static CyclicBarrier cyclicBarrier = new CyclicBarrier(6);
    //6个线程都执行到await()方法时，会先执行AfterWork()线程，AfterWork()线程执行完成后在继续向下执行。
    static CyclicBarrier cyclicBarrier = new CyclicBarrier(6, new AfterWork());
    // List<String> threadNames = new ArrayList<>();
    static ConcurrentHashMap<String, Object> concurrentHashMap = new ConcurrentHashMap<>(); //线程安全的集合
    static class WorkTask implements Runnable { 
        @Override
        public void run() { 
            try { 
                concurrentHashMap.put(Thread.currentThread().getName(), Thread.currentThread().getId());
                Random r = new Random();
                if (r.nextBoolean()) { 
                    System.out.println("线程" + Thread.currentThread().getName() + "做点别的事情...");
                    Thread.sleep(2000);
                }
                System.out.println("线程" + Thread.currentThread().getName() + "进入等待状态...");
                cyclicBarrier.await();//线程运行到此处等待的数量达到设定的初始值后，就会自动向下进行。
                Thread.sleep(1000);
                System.out.println("线程" + Thread.currentThread().getName() + "开始向下执行...");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (BrokenBarrierException e) { 
                e.printStackTrace();
            }
        }
    }
    static class AfterWork implements Runnable { 
        @Override
        public void run() { 
            if (!concurrentHashMap.isEmpty()) { 
                for (String name : concurrentHashMap.keySet()) { 
                    System.out.println("AfterWork:" + name);
                }
            }
        }
    }
    public static void main(String[] args) throws InterruptedException { 
        for (int i = 0; i < 6; i++) { 
// Thread.sleep(1000);
            new Thread(new WorkTask()).start();
        }
    }
}

上面的CyclicBarrier实例中计数器初始化为6，并且设定了一个子线程去处理别的业务。首先将循环启动6次线程实例，当6个线程都执行到await()方式的时候，屏障开放，6个线程分别会首先执行AfterWork()子线程，等待6个AfterWork()子线程都执行完成后，才会继续向下执行。
运行结果：

线程Thread-1进入等待状态...
线程Thread-0进入等待状态...
AfterWork:Thread-3
AfterWork:Thread-4
AfterWork:Thread-5
AfterWork:Thread-0
AfterWork:Thread-1
AfterWork:Thread-2
线程Thread-0开始向下执行...
线程Thread-3开始向下执行...
线程Thread-4开始向下执行...
线程Thread-5开始向下执行...
线程Thread-2开始向下执行...
线程Thread-1开始向下执行...