Java多线程六脉神剑-少商剑(CountDownLatch)、商阳剑(CyclicBarrier)

原创

BLACK595

修改于 2024-09-26 20:43:26

3130

修改于 2024-09-26 20:43:26

CountDownLatch(少商剑)

少商剑：剑路雄劲，石破天惊，CountDownLatch以其强大而直接的方式控制线程等待和同步，具有明确而有力的作用。

CountDownLatch是一个同步工具类，它是根据计数器实现的，构造函数初始时会指定总的计数数量，每调用一次countDown数量会减一，当数量为0时，闸门将会放开，await等待的线程进而继续执行。

例如，小红，小兰，小明一起去野餐，他们约定先一起到公园门口再开始活动，这时计数器初始值就是3，当小红到达目的地，计数器减1，小红继续等待小兰和小明；小兰到达计数器再减1，再一起等待小明的到达；小明到达，数量就减为0了，他们三人再一起进行活动。

一个简单的动画流程：

CountDownLath方法详解

构造函数 CountDownLatch(int count)：count为计数器的初始值（一般count就是线程数）
countDown()：每调用一次count数减1（一般当线程完成任务后调用）
getCount()：获取当前计数器count的值。
await()：一直等待，直到计数器count值为0。
boolean await(long timeout, TimeUnit unit)：一直等待，直到count值为0，或者过了timeout时间后停止等待。如果是count值为0停止的等待，返回的boolean就为true；如果是过了timeout时间后停止的等待，返回的boolean就为false。

举个栗子🌰

组装加工一台电脑，我们需要加工CPU、主板、内存、显卡、电源等部件，如果进行串行加工，那需要等待CPU加工完再加工主板，那耗时就是所有部件的总和，但如果像工厂那样，流水线作业并行处理，加工CPU的同时，其他所有部件也一起加工，等待所有的部件加工完，我们再组装成电脑，这样效率快了很多。

public class CountDownLatchCase {
    public static void produce() throws InterruptedException {
        List<Task> taskList = new ArrayList<>();
        taskList.add(new Task("CPU",2000L));
        taskList.add(new Task("主板",800L));
        taskList.add(new Task("内存",500L));
        taskList.add(new Task("显卡",1000L));
        taskList.add(new Task("电源",200L));
        CountDownLatch latch = new CountDownLatch(taskList.size());
        ConcurrentHashMap<String,String> results = new ConcurrentHashMap<>();
        for (int i = 0; i < taskList.size(); i++) {
            Task task = taskList.get(i);
            new Thread(() -> {
                try {
                    Thread.sleep(task.getTime());
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                latch.countDown();
                String threadName = Thread.currentThread().getName();
                results.put(threadName,task.getName());
                System.out.println(threadName+"完成加工:"+task.getName()+",耗时:"+task.getTime()+"毫秒");
            },"线程"+(i+1)).start();
        }
        latch.await();
        for (Map.Entry<String, String> entry : results.entrySet()) {
            System.out.println("总部收到了来自"+entry.getKey()+"加工的"+entry.getValue());
        }
        if(results.size() == taskList.size()){
            System.out.println("所有部件加工完成,可以组装电脑");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        produce();
    }
}

运行结果

线程5完成加工:电源,耗时:200毫秒
线程3完成加工:内存,耗时:500毫秒
线程2完成加工:主板,耗时:800毫秒
线程4完成加工:显卡,耗时:1000毫秒
线程1完成加工:CPU,耗时:2000毫秒
总部收到了来自线程4加工的显卡
总部收到了来自线程5加工的电源
总部收到了来自线程2加工的主板
总部收到了来自线程3加工的内存
总部收到了来自线程1加工的CPU
所有部件加工完成,可以组装电脑

其他使用场景

统计大屏页面数据时，把各个模块的数据使用多线程统计出来后，封装之后再一起返给前端。undefinedtheme: smartblueCyclicBarrier(商阳剑)商阳剑：商阳剑巧妙灵活，CyclicBarrier 可以灵活地设置线程同步的条件和重复使用，如同商阳剑般变化多端。
文件处理时，同时启动多个线程分别处理不同的文件，多线程把所有文件处理完毕后，再进行汇总和分析。
系统启动时多线程加载配置文件、初始化数据库连接等操作，当这些操作完成后，业务处理线程才能开始工作。

CyclicBarrier(商阳剑)

CyclicBarrier直译过来就是“循环栅栏”，在Java并发编程中，用于控制一组线程，先到达栅栏的等待其他未到达的线程，都到达后再一起继续执行，跑到下一个栅栏，也是同样等待所有线程都到达栅栏再继续到下一个栅栏，以此往复。

CyclicBarrier内有一个计数器，在构造函数初始化时会初始计数器的值，当调用await方法时，计数器会减一并将当前线程进行阻塞，表示次线程已到达栅栏，等待其他线程全部执行完毕，也就是计数器的值为0时。

CyclicBarrier方法详解

CyclicBarrier(int parties)：构造方法，其中parties表示计数器的数量（一般parties为线程数）
CyclicBarrier(int parties, Runnable barrierAction)：构造方法，barrierAction表示当线程到达屏障后先执行一个预定义的动作，再进行阻塞等待。
int await()：计数器减一，阻塞等待其他线程到达栅栏。返回的值表示当前线程在本次屏障等待中到达的序号，序号从0开始。
int await(long timeout, TimeUnit unit)：添加了一个超时时间，如果未在给定的timeout时间内被栅栏放行，将会抛出InterruptedException的异常。
reset()：重置计数器，重置后，如果有其他线程在await栅栏放行，在await的线程将会抛出BrokenBarrierException的异常。
boolean isBroken()：检查CyclicBarrier是否处于破损状态。当以下情况发生时CyclicBarrier会进入破损状态：
1. 某个等待的线程被中断。
2. 某个等待的线程超时。
3. 屏障动作（如果有指定）在执行过程中抛出异常。
int getNumberWaiting()：获取正在await的数量。
int getParties()：获取总数，也就是在构造方法中我们指定的parties值。

举个栗子🌰

在我们玩一些PVP游戏中，我们一般在开始游戏的时候都会有匹配玩家->选择角色->加载进入的过程，而每一个流程都需要所有玩家全部确认完成后才能继续，每一个过程相当于就是一个栅栏，拦住全部的玩家，直到所有的玩家完成才进行下一步。

比较经典的一张图：

代码实现：

public class CyclicBarrierCase {
    //匹配游戏->选择角色->加载进入
    public static void produce(){
        List<String> player = new ArrayList<>();
        player.add("玩家一");
        player.add("玩家二");
        player.add("玩家三");
        player.add("玩家四");
        CyclicBarrier barrier = new CyclicBarrier(player.size());
        for (int i = 0; i < player.size(); i++) {
            new Thread(() -> {
                String threadName = Thread.currentThread().getName();
                sleep(); //准备游戏时间
                System.out.println(threadName+"准备完成");
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
                sleep(); //选择角色时间
                System.out.println(threadName + "选择完成");
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
                sleep(); //加载进入时间
                System.out.println(threadName + "加载完成");
                try {
                    barrier.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
            },player.get(i)+"线程").start();
        }
    }
    public static void sleep(){
        //睡1到3秒的随机数
        int random = RandomUtil.randomInt(1000, 3000);
        try {
            Thread.sleep(random);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        produce();
    }
}

执行结果：

await时的异常

调用await时，会让我们捕获两个异常，一个是BrokenBarrierException另一个是InterruptedException。

出现BrokenBarrierException这种情况可能是CyclicBarrier的某个线程在等待期间被中断，或者CyclicBarrier被重置。这通常意味着CyclicBarrier无法正常工作，处理的方案可能有捕获异常并记录日志，通过某种共享状态或消息机制通知其他线程CyclicBarrier已损坏，执行一些恢复操作来尽量弥补或减轻由于栅栏损坏导致的影响。

出现InterruptedException是因为在调用await(long timeout, TimeUnit unit)我们指定了超时时间，当指定时间过后，线程还没有被放行，抛出超时异常，处理超时的策略可能包括重试机制或者回退逻辑。但重要的是要确保所有的线程在超时后都能正确地处理这种情况，避免资源泄漏或者线程阻塞。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯技术创作特训营S9

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯技术创作特训营S9

#CountDownLatch

#CyclicBarrier

登录后参与评论

0 条评论

热度