同步屏障CyclicBarrier

简介: 同步屏障CyclicBarrier

同步屏障CyclicBarrier

CyclicBarrier的字面意思是可循环使用(Cyclic)的屏障(Barrier)。它要做的事情是,让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行。

CyclicBarrier默认的构造方法是CyclicBarrier(int parties),其参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。

public class CyclicBarrierTest {
    static CyclicBarrier c = new CyclicBarrier(2);
    public static void main(String[] args) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
                System.out.println(1);
            }
        }).start();
        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }
}

因为主线程和子线程的调度是由CPU决定的,两个线程都有可能先执行,所以会产生两种输出 12 和21

如果把new CyclicBarrier(2)修改成new CyclicBarrier(3),则主线程和子线程会永远等待,因为没有第三个线程执行await方法,即没有第三个线程到达屏障,所以之前到达屏障的两个线程都不会继续执行。

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties,Runnable barrierAction),用于在线程到达屏障时,优先执行barrierAction,方便处理更复杂的业务场景

public class CyclicBarrierTest2 {
    static CyclicBarrier c = new CyclicBarrier(2, new A());

    public static void main(String[] args) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
                System.out.println(1);
            }
        }).start();
        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }

    static class A implements Runnable {
        @Override
        public void run() {
            System.out.println(3);
        }
    }
}

因为CyclicBarrier设置了拦截线程的数量是2,所以必须等代码中的第一个线程和线程A都执行完之后,才会继续执行主线程,然后输出312

public class BankWaterService implements Runnable {
    /**
     * 创建4个屏障,处理完之后执行当前类的run方法
     */
    private CyclicBarrier c = new CyclicBarrier(4, this);
    /**
     * 假设只有4个sheet,所以只启动4个线程
     */
    private Executor executor = Executors.newFixedThreadPool(4);
    /**
     * 保存每个sheet计算出的银流结果
     */
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new
            ConcurrentHashMap<String, Integer>();

    private void count() {
        for (int i = 0; i < 4; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                // 计算当前sheet的银流数据,计算代码省略
                    sheetBankWaterCount
                            .put(Thread.currentThread().getName(), 1);
                // 银流计算完成,插入一个屏障
                    try {
                        c.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    @Override
    public void run() {
        int result = 0;
        // 汇总每个sheet计算出的结果
        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
            result += sheet.getValue();
        }
// 将结果输出
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService bankWaterCount = new BankWaterService();
        bankWaterCount.count();
    }
}

使用线程池创建4个线程,分别计算每个sheet里的数据,每个sheet计算结果是1,再由BankWaterService线程汇总4个sheet计算出的结果,输出结果4

CountDownLatch的计数器只能使用一次,而CyclicBarrier的计数器可以使用reset()方法重置。所以CyclicBarrier能处理更为复杂的业务场景。例如,如果计算发生错误,可以重置计数器,并让线程重新执行一次。

CyclicBarrier还提供其他有用的方法,比如getNumberWaiting方法可以获得Cyclic-Barrier阻塞的线程数量。isBroken()方法用来了解阻塞的线程是否被中断

public class CyclicBarrierTest3 {
    static CyclicBarrier c = new CyclicBarrier(2);
    public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                }
            }
        });
        thread.start();
        thread.interrupt();
        try {
            c.await();
        } catch (Exception e) {
            System.out.println(c.isBroken());
        }
    }
}
相关文章
|
运维 安全 Devops
蚂蚁集团TRaaS技术风险防控平台入选中国信通院《信息系统稳定性保障能力建设指南(1.0)》最佳实践案例
近日,中国信息通信研究院分布式系统稳定性实验室正式发布了《信息系统稳定性保障能力建设指南》(以下简称《指南》)。蚂蚁集团应邀深度参与了《指南》的研讨编制,该指南收录了包括蚂蚁集团在内的多家知名机构在系统稳定性保障服务方面的优秀案例,旨在为各行业提升系统稳定性能力提供参考。
1029 0
蚂蚁集团TRaaS技术风险防控平台入选中国信通院《信息系统稳定性保障能力建设指南(1.0)》最佳实践案例
|
API
Navigator和导航栏之间的区别以及用法场景的分析
区别:    1.navigator是属于小程序组件中的,导航栏是属于小程序API中的    2.navigator组件是用在axml页面中跳转的导航,它有4种类型(见下表);导航栏API是用在js中实现页面跳转的    3.
1074 12
|
数据采集 消息中间件 存储
【译】Databricks使用Spark Streaming和Delta Lake对流式数据进行数据质量监控介绍
本文主要对Databricks如何使用Spark Streaming和Delta Lake对流式数据进行数据质量监控的方法和架构进行了介绍,本文探讨了一种数据管理架构,该架构可以在数据到达时,通过主动监控和分析来检测流式数据中损坏或不良的数据,并且不会造成瓶颈。
【译】Databricks使用Spark Streaming和Delta Lake对流式数据进行数据质量监控介绍
|
机器人
阿里云RPA(机器人流程自动化)干货系列之五:业务流程梳理方法
本文是阿里云RPA(机器人流程自动化)干货系列之五,详细介绍了在开发阿里云RPA机器人应用之前对客户的业务流程进行详细、全面地梳理,并识别出流程中的关键点和换位思考点,为后续应用开发打好基础。
11228 1
|
域名解析 网络协议 CDN
一文看懂云解析DNS添加CDN解析
CDN解析是通过将域名指向CDN厂商提供的CNAME地址,实现网站的加速服务。
3017 0
|
Kubernetes Perl 容器
K8S集群优化之修复ServiceEndpoint更新的延迟
几个月前,我在更新 Kubernetes 集群中的 Deployment 时发现了一个很奇怪的连接超时现象,在更新 Deployment 之后的 30 秒到两分钟左右,所有与以该 Deployment作为服务后端的 Service 的连接都会超时或失败。
2289 0
|
算法
C#4.0新特性-&quot;协变&quot;“.NET研究”与&quot;逆变&quot;以及背后的编程思想
  在《上篇》中我们揭示了“缺省参数”的本质,现在我们接着来谈谈C#4.0中另一个重要的新特性:协变(Covariance)与逆变(Contravariance)。对于协变与逆变,大家肯定不会感到陌生,但是我相信有很多人不能很清晰地说出他们之间的区别。
1175 0
|
Linux Shell 数据安全/隐私保护
chmod命令
chmod是权限管理命令change the permissions mode of a file的缩写.给文件添加可执行权限u代表所有者,x代表执行权限。 + 表示增加权限。chmod u+x file.sh 就表示对当前目录下的file.sh文件的所有者增加可执行权限。
23716 0

热门文章

最新文章