Java并发工具类

JDK并发包中常用并发工具类:

CountDownLatch、CyclicBarrier和Semaphore工具类提供了一种并发流程控制的手段;

Exchanger工具类则提供了在线程间交换数据的一种手段。

等待多线程完成的CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作。

需求:解析一个Excel里多个sheet的数据,可以考虑使用多线程,每个线程解析一个sheet里的数据,等到所有的sheet都解析完之后,程序需要提示解析完成。

实现主线程等待所有线程完成sheet的解析操作,最简单的做法是使用join()方法

public class JoinCountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        Thread parser1 = new Thread(() -> {
        });
        Thread parser2 = new Thread(() -> System.out.println("parser2 finish"));
        parser1.start();
        parser2.start();
        parser1.join();
        parser2.join();
        System.out.println("all parser finish");
    }
}

join用于让当前执行线程等待join线程执行结束实现原理是不停检查join线程是否存活,如果join线程存活则让当前线程永远等待。其中,wait(0)表示永远等待下去。代码片段如下:

            while (isAlive()) {
                wait(0);
            }

直到join线程中止后,线程的this.notifyAll()方法会被调用,调用notifyAll()方法是在JVM里实现的,在JDK里看不到。

CountDownLatch也可以实现join的功能,并且比join的功能更多

public class CountDownLatchTest {
    static CountDownLatch c = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            System.out.println(1);
            c.countDown();
            System.out.println(2);
            c.countDown();
        }).start();
        c.await();
        System.out.println("3");
    }
}

CountDownLatch的构造函数接收一个int类型的参数作为计数器,等待N个点完成,这里就传入N。

调用CountDownLatch的countDown方法时,N就会减1,CountDownLatch的await方法会阻塞当前线程,直到N变成零

countDown方法可以用在任何地方,N个点,可以是N个线程,也可以是1个线程里的N个执行步骤。用在多个线程时,只需要把这个CountDownLatch的引用传递到线程里即可。

如果有某个解析sheet的线程处理得比较慢,不可能让主线程一直等待,可以使用另外一个带指定时间的await方法——await(long time,TimeUnit unit),这个方法等待特定时间后,就会不再阻塞当前线程。join也有类似的方法。

同步屏障CyclicBarrier

让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行。

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

public class CyclicBarrierTest {
    static CyclicBarrier c = new CyclicBarrier(2);

    public static void main(String[] args) {
        new Thread(() -> {
            try {
                c.await();
            } catch (Exception e) {
            }
            System.out.println(1);
        }).start();

        try {
            c.await();
        } catch (Exception e) {
        }
        System.out.println(2);
    }
}

主线程和子线程的调度是由CPU决定的,两个线程都有可能先执行。

输出结果可能有两种:

一种是:

image-20230824230526154

另一种:

image-20230824230510366

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

更高级的构造函数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(() -> {
            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);
        }
    }
}

初始值设为2,等代码中的第一个线程和线程A都执行完之后,才会继续执行主线程,然后输出2。结果一定是:

image-20230824231118034

CyclicBarrier的应用场景

CyclicBarrier可以用于多线程计算数据,最后合并计算结果的场景。

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

    private void count() {
        for (int i = 0; i < 4; i++) {
            executor.execute(() -> {
                // 计算当前sheet的银流数据,计算代码省略
                sheetBankWaterCount
                        .put(Thread.currentThread().getName(), 1);
                // 银流计算完成,插入一个屏障
                try {
                    c.await();
                } catch (InterruptedException |
                         BrokenBarrierException 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();
    }
}

计算银行流水,一个sheet开启一个线程,所有线程执行完毕,将所有计算结果相加得银行总流水。

最后输出结果为4。

CyclicBarrier和CountDownLatch的区别

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

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

控制并发线程数的Semaphore

Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。

应用场景

Semaphore可以用于做流量控制,特别是公用资源有限的应用场景,比如数据库连接。

需求:要读取几万个文件的数据,因为都是IO密集型任务,可以启动几十个线程并发地读取,但是如果读到内存后,还需要存储到数据库中,而数据库的连接数只有10个,这时必须控制只有10个线程同时获取数据库连接保存数据,否则会报错无法获取数据库连接。此时就使用Semaphore来做流量控制。

public class SemaphoreTest {
    private static final int THREAD_COUNT = 30;
    private static final ExecutorService threadPool = Executors
            .newFixedThreadPool(THREAD_COUNT);
    private static final Semaphore s = new Semaphore(10);

    public static void main(String[] args) {
        for (int i = 0; i < THREAD_COUNT; i++) {
            threadPool.execute(() -> {
                try {
                    s.acquire();
                    System.out.println("save data");
                    s.release();
                } catch (InterruptedException e) {
                }
            });
        }
        threadPool.shutdown();
    }
}

有30个线程在执行,但是只允许10个并发执行。

构造方法Semaphore(int permits)接受一个整型的数字,表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证,也就是最大并发数是10。Semaphore的用法也很简单,首先线程使用Semaphore的acquire()方法获取一个许可证,使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证。

一些其他方法:

image-20230824233007777

线程间交换数据的Exchanger

Exchanger(交换者)是一个用于线程间协作的工具类。

用于进行线程间的数据交换。它提供一个同步点,在这个同步点,两个线程可以交换彼此的数据。

两个线程通过exchange方法交换数据,如果第一个线程先执行exchange()方法,它会一直等待第二个线程也执行exchange方法,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。

Exchanger可以用于遗传算法。遗传算法里需要选出两个人作为交配对象,这时候会交换两人的数据,并使用交叉规则得出2个交配结果。

Exchanger也可以用于校对工作

需求:将纸制银行流水通过人工的方式录入成电子银行流水,为了避免错误,采用AB岗两人进行录入,录入到Excel之后,系统需要加载这两个Excel,并对两个Excel数据进行校对,看看是否录入一致。

两个线程间数据传递。

public class ExchangerTest {
    private static final Exchanger<String> exgr = new Exchanger<>();
    private static final ExecutorService threadPool = Executors.newFixedThreadPool(2);

    public static void main(String[] args) {
        threadPool.execute(() -> {
            try {
                // A录入银行流水数据
                String A = "银行流水A";
                String B = exgr.exchange(A);
                System.out.println("B----- " + B);
            } catch (InterruptedException e) {
            }
        });
        threadPool.execute(() -> {
            try {
                // B录入银行流水数据
                String B = "银行流水B";
                String A = exgr.exchange("B");
                System.out.println("A和B数据是否一致:" + A.equals(B) + ",A录入的是:"
                        + A + ",B录入是:" + B);
            } catch (InterruptedException e) {
            }
        });
        threadPool.shutdown();
    }
}

输出结果:

image-20230824233759768

如果两个线程有一个没有执行exchange()方法,则会一直等待,避免一直等待,可以使用exchange(V x,longtimeout,TimeUnit unit)设置最大等待时长

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/89691.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

爬虫异常处理:异常捕获与容错机制设计

作为一名专业的爬虫程序员&#xff0c;每天使用爬虫IP面对各种异常情况是我们每天都会遇到的事情。 在爬取数据的过程中&#xff0c;我们经常会遇到网络错误、页面结构变化、被反爬虫机制拦截等问题。在这篇文章中&#xff0c;我将和大家分享一些关于如何处理爬虫异常情况的经…

创建和分析二维桁架和梁结构研究(Matlab代码实现)

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…

Jsp 解决out.print()输出多出空行

一、原因 在 JSP 中&#xff0c;HTML 标签和 JSP 指令之外的内容会被当作文本处理&#xff0c;包括空行、空格和制表符等。当 JSP 引擎解析 JSP 页面时&#xff0c;会将这些文本内容原封不动地输出到响应中。 http响应 二、解决方法 在Jsp页面最前端添加 <% page trimDir…

【Linux】动态库和静态库

动态库和静态库 软链接硬链接硬链接要注意 自定义实现一个静态库(.a)解决、使用方法静态库的内部加载过程 自定义实现一个动态库&#xff08;.so&#xff09;动态库加载过程 静态库和动态库的特点 软链接 命令:ln -s 源文件名 目标文件名 软链接是独立连接文件的&#xff0c;他…

【Winform学习笔记(八)】通过委托实现跨窗体传值

通过委托实现跨窗体传值 前言正文1、委托及事件2、通过委托实现跨窗体传值的步骤1.在子窗体中定义委托2.在子窗体中声明一个委托类型的事件3.调用委托类型事件4.在实例化子窗体后&#xff0c;子窗体订阅事件接受方法5.实现具体的事件 3、具体示例4、完整代码5、实现效果 前言 …

论文阅读_条件控制_ControlNet

name_en: Adding Conditional Control to Text-to-Image Diffusion Models name_ch: 向文本到图像的扩散模型添加条件控制 paper_addr: http://arxiv.org/abs/2302.05543 date_read: 2023-08-17 date_publish: 2023-02-10 tags: [‘图形图像’,‘大模型’,‘多模态’] author: …

解决政务审计大数据传输难题!镭速传输为政务行业提供解决方案

政务行业是国家治理的重要组成部分&#xff0c;涉及到国家安全、社会稳定、民生福祉等方面。随着信息技术的快速发展和革新&#xff0c;政务信息化也迎来了新一轮的升级浪潮。国家相继出台了《国家信息化发展战略纲要》《“十三五”国家信息化规划》《“十四五”推进国家政务信…

华为OD机试 - 最佳植树距离 - 二分查找(Java 2023 B卷 100分)

目录 一、题目描述二、输入描述三、输出描述四、备注说明五、二分查找六、解题思路七、Java算法源码八、效果展示1、输入2、输出3、说明 一、题目描述 按照环保公司要求&#xff0c;小明需要在沙化严重的地区进行植树防沙工作&#xff0c;初步目标是种植一条直线的树带。 由于…

算法笔记(一):时间复杂度

省略的技巧 已知 f ( n ) f(n) f(n) 来说&#xff0c;求 g ( n ) g(n) g(n) 表达式中相乘的常量&#xff0c;可以省略&#xff0c;如 f ( n ) 100 ∗ n 2 f(n) 100*n^2 f(n)100∗n2 中的 100 100 100 多项式中数量规模更小&#xff08;低次项&#xff09;的表达式&#…

基于JSP+Servlet+mysql养老院管理系统

基于JSPServletmysql养老院管理系统 一、系统介绍二、功能展示四、其他系统实现五、获取源码 一、系统介绍 项目类型&#xff1a;Java web项目 项目名称&#xff1a;基于JSPServlet的养老院管理系统 [yanglaoyuan] 当前版本&#xff1a;V1.0.0版本 难度等级&#xff1a;✩✩点…

使用Dockker创建vwas容器时报错的解决方法

执行命令 docker run -it -d -p 13443:3443 --cap-add LINUX_IMMUTABLE secfa/docker-awvs没有详细看报错之前找了各种各样的解决办法&#xff0c;都无法解决。因此以后在看报错提示的时候耐心一点看关键词Error 后来才发现启动vwas时docker报了这个错&#xff1a; OSError: …

人力资源小程序的设计与开发步骤

在当前信息化时代&#xff0c;小程序成为了各行各业提升用户体验和服务效率的重要渠道。人力资源部门也可以通过定制开发人力资源小程序来提升招聘、培训、员工福利等方面的工作效率。接下来&#xff0c;我们将介绍人力资源小程序定制系统开发的具体流程。 首先&#xff0c;我们…

Facebook HiPlot “让理解高维数据变得容易”

在这个全球信息化的时代&#xff0c;数据量呈爆炸式增长&#xff0c;数据的复杂性也是如此。如何有效地处理高维数据并找到隐藏在其中的相关性和模式是一个严峻的挑战。近年来&#xff0c;可视化和可视化分析已被应用于该任务&#xff0c;并取得了一些积极成果。Facebook的新Hi…

1782. 统计点对的数目

给你一个无向图&#xff0c;无向图由整数 n &#xff0c;表示图中节点的数目&#xff0c;和 edges 组成&#xff0c;其中 edges[i] [ui, vi] 表示 ui 和 vi 之间有一条无向边。同时给你一个代表查询的整数数组 queries 。 第 j 个查询的答案是满足如下条件的点对 (a, b) 的数…

Agile Iteration Velocity

【agile iteration velocity】敏捷速度指的平均速度 第四次迭代结束速度&#xff1a; 76 / 4 19 第五次迭代结束速度&#xff1a; &#xff08;76 24 &#xff09; / 5 100 / 5 20

Qt 自定义提示框 右下角冒泡

网页右下角上经常会出现一些提示性的信息&#xff0c;B/S有的东西&#xff0c;C/S当然也可以有&#xff0c;就像QQ的消息提示一样&#xff01; 实现一个类似的东西并不困难&#xff0c;只要想明白原理实现起来就很简单了&#xff01; 实现原理&#xff1a; &#xff08;1&#…

20230822 Windows上使用find_package引入OpenCV报错

报错信息 打开Cmake项目时&#xff0c;find_package 报错&#xff1a; Found OpenCV Windows Pack but it has no binaries compatible with yourconfiguration.You should manually point CMake variable OpenCV_DIR to your build of OpenCVlibrary.原因 大概率原项目是在 …

Systick滴答定时器

今天&#xff0c;对Systick滴答定时器进行资料的整理&#xff0c;这个定时器在程序中的作用就是提供延时函数。参考&#xff08;【STM32】Systick滴答定时器_一只大喵咪1201的博客-CSDN博客&#xff09; Systick滴答定时器的介绍 相关寄存器 寄存器CTRL 补充HCLK 寄存器LOAD…

Python项目开发案例————学生信息管理系统(附源码)

一、学生信息管理系统 本文使用Python语言开发了一个学生信息管理系统&#xff0c;该系统可以帮助教师快速录入学生的信息&#xff0c;并且对学生的信息进行基本的增、删、改、查操作&#xff1b;还可以实时地将学生的信息保存到磁盘文件中。 1.1 需求分析 为了顺应互联网时代…

2023年高教社杯数学建模思路 - 复盘:校园消费行为分析

文章目录 0 赛题思路1 赛题背景2 分析目标3 数据说明4 数据预处理5 数据分析5.1 食堂就餐行为分析5.2 学生消费行为分析 建模资料 0 赛题思路 &#xff08;赛题出来以后第一时间在CSDN分享&#xff09; https://blog.csdn.net/dc_sinor?typeblog 1 赛题背景 校园一卡通是集…