Java并发控制的几个工具类CountDownLatch,CyclicBarrier,Semaphore

今天就简单总结一下这几个工具类怎么在多线程并发的时候体现他们的优点的。下面我们进行一一介绍

CountDownLatch

一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。
主要方法

 public CountDownLatch(int count); //构造方法参数指定了计数的次数
 public void countDown();//当前线程调用此方法,则计数减一
 public void await() throws InterruptedException//调用此方法会一直阻塞当前线程,直到计时器的值为0

这里举例模拟一个场景说明,运动员在听到枪声响,一同跑步,最后全部跑完结束比赛

private static void startCountdLatch() {
        // 这里当执行countdown时开始比赛
        CountDownLatch begin = new CountDownLatch(1);
        // 这里计数等于零时,结束比赛
        CountDownLatch end = new CountDownLatch(PLAYER_COUNT);
        Player[] players = new Player[PLAYER_COUNT];
        for (int i = 0; i < PLAYER_COUNT; i++) {
            players[i] = new Player(i, begin, end);
        }

        ExecutorService exe = Executors.newFixedThreadPool(PLAYER_COUNT);
        for (Player player : players) {
            exe.execute(player);
            // new Thread(player).start();
        }

        System.out.println("Race begin.");
        begin.countDown(); // 执行一次计数减一

        try {
            end.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("Race end.");
        }
        exe.shutdown(); //
    }

接着是运动员

public class Player implements Runnable {
    private int id;
    private CountDownLatch begin;
    private CountDownLatch end;

    public Player(int id, CountDownLatch begin, CountDownLatch end) {
        super();
        this.id = id;
        this.begin = begin;
        this.end = end;
    }

    @Override
    public void run() {
        try {
            begin.await(); // 当计数器大于0时阻塞线程
            System.out.println("Player " + id + " is running.");
            long time = (long) (Math.random() * 3000);
            Thread.sleep(time);
            System.out.println("Player " + id + " is arrvied. his time is " + time + "ms");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            end.countDown(); // 计数器减1
        }
    }
}

输出结果是

Race begin.
Player 0 is running.
Player 1 is running.
Player 3 is running.
Player 4 is running.
Player 5 is running.
Player 2 is running.
Player 0 is arrvied. his time is 154ms
Player 1 is arrvied. his time is 601ms
Player 3 is arrvied. his time is 1860ms
Player 2 is arrvied. his time is 2316ms
Player 4 is arrvied. his time is 2782ms
Player 5 is arrvied. his time is 2872ms
Race end.

CyclicBarrier

CyclicBarrier是一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待,此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用,所以称它为循环 的 barrier。
  CyclicBarrier类似于CountDownLatch也是个计数器, 不同的是CyclicBarrier数的是调用了CyclicBarrier.await()进入等待的线程数, 当线程数达到了CyclicBarrier初始时规定的数目时,所有进入等待状态的线程被唤醒并继续。 CyclicBarrier就象它名字的意思一样,可看成是个障碍, 所有的线程必须到齐后才能一起通过这个障碍。 CyclicBarrier初始时还可带一个Runnable的参数,此Runnable任务在CyclicBarrier的数目达到后,所有其它线程被唤醒前被执行。

相关方法

await()
在所有参与者都已经在此 barrier 上调用 await方法之前,将一直等待。
await(long timeout, TimeUnit unit)
在所有参与者都已经在此屏障上调用 await方法之前,将一直等待。

这里模拟一下,所有人做一项工作,当所有人都做完了,才可以进行下一项工作。

public class People implements Runnable {
    private CyclicBarrier barrier;
    private int id;

    public People(int id, CyclicBarrier barrier) {
        this.id = id;
        this.barrier = barrier;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+":People " + id + " is working.");
        long time = (long) (Math.random() * 2000);
        try {
            Thread.sleep(time);
            System.out.println("People " + id + " complete Task with " + time + "ms");
            barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println("All people complete work.People " + id + " begin other task.");
    }
}

测试

private static void startCyclicBarrier() {
        CyclicBarrier barrier = new CyclicBarrier(PLAYER_COUNT, new Runnable() {

            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + " excute other task.");
            }
        });
        People[] peoples = new People[PLAYER_COUNT];
        for (int i = 0; i < PLAYER_COUNT; i++) {
            peoples[i] = new People(i, barrier);
        }
        ExecutorService exe = Executors.newFixedThreadPool(PLAYER_COUNT);
        
        //这里如果人数小于PLAYER_COUNT,将会一直等待
        for (People people : peoples) {
            exe.execute(people);
        }
        exe.shutdown();
    }

输出结果:

pool-1-thread-1:People 0 is working.
pool-1-thread-3:People 2 is working.
pool-1-thread-2:People 1 is working.
pool-1-thread-4:People 3 is working.
pool-1-thread-6:People 5 is working.
pool-1-thread-5:People 4 is working.
People 5 complete Task with 380ms
People 4 complete Task with 585ms
People 3 complete Task with 617ms
People 1 complete Task with 955ms
People 0 complete Task with 1324ms
People 2 complete Task with 1754ms
pool-1-thread-3 excute other task.
All people complete work.People 2 begin other task.
All people complete work.People 4 begin other task.
All people complete work.People 3 begin other task.
All people complete work.People 1 begin other task.
All people complete work.People 5 begin other task.
All people complete work.People 0 begin other task.

Semaphore

一个计数信号量。从概念上讲,信号量维护了一个许可集。如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个 release() 添加一个许可,从而可能释放一个正在阻塞的获取者。但是,不使用实际的许可对象,Semaphore 只对可用许可的号码进行计数,并采取相应的行动。拿到信号量的线程可以进入代码,否则就等待。通过acquire()和release()获取和释放访问许可。

方法介绍

acquire
public void acquire() throws InterruptedException
从此信号量获取一个许可,在提供一个许可前一直将线程阻塞,否则线程被中断。获取一个许可(如果提供了一个)并立即返回,将可用的许可数减 1。
如果没有可用的许可,则在发生以下两种情况之一前,禁止将当前线程用于线程安排目的并使其处于休眠状态:
某些其他线程调用此信号量的 release()方法,并且当前线程是下一个要被分配许可的线程;或者其他某些线程中断当前线程。
如果当前线程:
被此方法将其已中断状态设置为 on ;或者在等待许可时被中断。
则抛出 InterruptedException,并且清除当前线程的已中断状态。
抛出:
InterruptedException

  • 如果当前线程被中断
    release
    public void release()
    释放一个许可,将其返回给信号量。释放一个许可,将可用的许可数增加 1。如果任意线程试图获取许可,则选中一个线程并将刚刚释放的许可给予它。然后针对线程安排目的启用(或再启用)该线程。
    不要求释放许可的线程必须通过调用 acquire()
    来获取许可。通过应用程序中的编程约定来建立信号量的正确用法。

下面做一个场景,许多工人,只有很少的机器,一个人用完了才能下一个人去用。

private static void startSemaphore() {
        Worker[] workers = new Worker[PLAYER_COUNT];
        Semaphore semaphore = new Semaphore(4);//允许最大并发数,加上boolean就是公平值,等的越久的首先获得许可
        for (int i = 0; i < PLAYER_COUNT; i++) {
            workers[i] = new Worker(i, semaphore);
        }

        for (Worker worker : workers) {
            new Thread(worker).start();
        }
    }

继续看代码

public class Worker implements Runnable {
    private Semaphore semaphore;
    private int id;

    public Worker(int id, Semaphore semaphore) {
        this.id = id;
        this.semaphore = semaphore;
    }

    @Override
    public void run() {
        try {
            semaphore.acquire();
            long time = (long) (Math.random() * 3000);
            System.out.println("工人 " + id + " 正在占用一台机器。");
            Thread.sleep(time);
            System.out.println("工人 " + id + " 释放一台机器。用了"+time+"ms");
            semaphore.release();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

输出结果:

工人 0 正在占用一台机器。
工人 1 正在占用一台机器。
工人 2 正在占用一台机器。
工人 4 正在占用一台机器。
工人 4 释放一台机器。用了1093ms
工人 3 正在占用一台机器。
工人 1 释放一台机器。用了1357ms
工人 5 正在占用一台机器。
工人 0 释放一台机器。用了1388ms
工人 2 释放一台机器。用了2633ms
工人 5 释放一台机器。用了1300ms
工人 3 释放一台机器。用了2735ms

补充总结:

CyclicBarrier和CountDownLatch的区别
CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让线程们重新执行一次。
CyclicBarrier还提供其他有用的方法,比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true。

好了,就这些了,又学习了新东西了。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,189评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,577评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,857评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,703评论 1 276
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,705评论 5 366
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,620评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,995评论 3 396
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,656评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,898评论 1 298
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,639评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,720评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,395评论 4 319
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,982评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,953评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,195评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 44,907评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,472评论 2 342

推荐阅读更多精彩内容