Java中的并发工具类

`CountDownLatch`等待多线程完成

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

譬如：解析一个excel，一个线程解析一个sheet页，当所有线程解析完成之后，提示解析完成。可以使用join来实现，也可以用CountDownLatch。

使用join
join让当前执行线程等待join线程执行结束。

package com.thread;

public class JoinCountDownLatchTest {
    public static void main(String[] args) {
        Thread parser1 = new Thread(new Runnable(){
            @Override
            public void run() {
                
            }
        });
        Thread parser2 = new Thread(new Runnable(){
            @Override
            public void run() {
                try {
                    Thread.sleep(1000*2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("parser2 finished!");
            }
        });
        parser1.start();
        parser2.start();
        try {
            parser1.join();
            parser2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("All parser finished!");
    }
}

使用CountDownLatch
CountDownLatch传入一个N当做计数器，每次执行countDown的时候N就会减1，CountDownLatch的await方法就会阻塞当前线程，直到N变成零。countDown可以是一个线程中的N个步骤或者是N个线程。

一个线程调用countDown方法，一个线程调用await方法。

package com.thread;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest {
    public static CountDownLatch cdl = new CountDownLatch(2);
    
    public static void main(String[] args) {
        new Thread(new Runnable(){
            @Override
            public void run() {
                System.out.println(1);
                cdl.countDown();
                System.out.println(2);
                cdl.countDown();
            }
        }).start();
        try {
            cdl.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(3);
    }
}

`CyclicBarrier`同步屏障

让一组线程到达一个屏障（或者是同步点）的时候被阻塞，直到最后一个线程到达屏障，屏障才会打开，所有的线程继续往下执行。

package com.thread;

import java.util.concurrent.CyclicBarrier;

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

** CyclicBarrier升级版

高级的构造方法CyclicBarrier(int parties, Runnable barrierAction)：当所有线程到达同步点之后，优先执行barrierAction，等待该线程执行完之后，再继续执行await后面的方法。

package com.thread;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest {
    static CyclicBarrier c = new CyclicBarrier(2, new A());
    
    public static void main(String[] args) throws Exception {
        new Thread(new Runnable(){
            @Override
            public void run() {
                try {
                    Thread.sleep(3000);
                    System.out.println(2);
                    c.await();
                    System.out.println(2.1);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        System.out.println(1);
        c.await();
        System.out.println(1.1);
    }
    
    static class A implements Runnable{
        @Override
        public void run() {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(3);
        }
    }
}
/*
1
2
3
2.1
1.1
*/

`CyclicBarrier`应用场景

`CyclicBarrier`和`CountDownLatch`的区别

CountDownBatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法重置。因此CyclicBarrier可以实现更加复杂的功能。例如：处理计算错误，可以重置计数器，让线程重新执行一次。

CyclicBarrier的其他方法：

getNumberWaiting：获取阻塞的线程数量。
isBroken()用来了解阻塞的线程是否被中断。

package com.thread;

import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest2 {
    static CyclicBarrier c = new CyclicBarrier(2);
    
    public static void main(String[] args) {
        Thread t1 = new Thread(new Runnable(){
            @Override
            public void run() {
                try {
                    c.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }       
        });
        t1.start();
        t1.interrupt();
        try {
            c.await();
        } catch (Exception e) {//这里为什么会抛出异常呢？？？不明白
            System.out.println(c.isBroken());
            e.printStackTrace();
        }
    }
}

控制并发线程数的`Semaphore`

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

int availablePermits：返回此信号量中当前可用的许可证数。
int getQueueLength()：返回正在等待获取许可证的线程数。
boolean hasQueueThreads：是否有线程正在等待获取许可证
void reducePermits(int reduction)：减少reduction个许可证。
Collection getQueuedThreads()：返回所有等待获取许可证的线程集合

下面同时开启了30个线程，都进入了run方法内，但是同时运行在s.acquire();***s.release();之间的只能有10个线程。

package com.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class SemaphoreTest {
    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
    private static Semaphore s = new Semaphore(10);
    
    public static void main(String[] args) {
        for(int i=0; i<THREAD_COUNT; i++){
            threadPool.execute(new MyThread(i, s));
        }
        threadPool.shutdown();
    }
    
}
class MyThread implements Runnable{
    int c = 0;
    Semaphore s;
    public MyThread(int c, Semaphore s) {
        this.c = c;
        this.s = s;
    }
    @Override
    public void run() {
        try {
            System.out.println(c + " begin:");
            s.acquire();
            System.out.println("saveDate=" + c);
            Thread.sleep(3000);
            s.release();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
0 begin:
1 begin:
saveDate=0
saveDate=1
2 begin:
saveDate=2
8 begin:
6 begin:
7 begin:
saveDate=7
5 begin:
saveDate=5
3 begin:
4 begin:
saveDate=3
11 begin:
saveDate=11
9 begin:
saveDate=8
saveDate=6
12 begin:
10 begin:
saveDate=4
16 begin:
20 begin:
14 begin:
19 begin:
15 begin:
13 begin:
25 begin:
29 begin:
27 begin:
23 begin:
17 begin:
21 begin:
28 begin:
24 begin:
22 begin:
18 begin:
26 begin:
//下面是3s之后
saveDate=9
saveDate=12
saveDate=10
saveDate=16
saveDate=20
saveDate=14
saveDate=19
saveDate=15
saveDate=13
saveDate=25
//下面是6s之后
saveDate=29
saveDate=27
saveDate=23
saveDate=17
saveDate=21
saveDate=28
saveDate=24
saveDate=22
saveDate=18
saveDate=26
*/

线程之间交换数据`Exchanger`

Exchanger是个用于线程间协作的工具类，用于线程之间的数据交换。
它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。第一个线程先执行exchange()方法，第二个线程也执行exchange()方法，当两个线程同时到达同步点，这两个线程就可以交换数据。

如果一个线程一直没有执行exchange()方法，那么会一直等下去，如果担心特殊情况，可以使用exchange(V v,longtimeout, TimeUnit unit)设置最大等待时间。

package com.thread;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExchangerTest {
    private static final Exchanger<String> exgr = new Exchanger<String>();
    private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
    public static void main(String[] args) {
        threadPool.execute(new Runnable(){
            @Override
            public void run() {
                String a = "银行流水A";
                try {
                    String b = exgr.exchange(a);
                    System.out.println("a中数据交换完毕.a=" + a+";b="+b);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        threadPool.execute(new Runnable(){
            @Override
            public void run() {
                String b = "银行流水B";
                try {
                    Thread.sleep(3000);
                    String a = exgr.exchange(b);//传递b数据并获得a的数据
                    System.out.println("b中数据交换完毕.a=" + a+";b="+b);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
/*
b中数据交换完毕.a=银行流水A;b=银行流水B
a中数据交换完毕.a=银行流水A;b=银行流水B
*/

最后编辑于：2017.12.04 01:38:33

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 194,390评论 5赞 459
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 81,821评论 2赞 371
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 141,632评论 0赞 319
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 52,170评论 1赞 263
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 61,033评论 4赞 355
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 46,098评论 1赞 272
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 36,511评论 3赞 381
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 35,204评论 0赞 253
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 39,479评论 1赞 290
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 34,572评论 2赞 309
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 36,341评论 1赞 326
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 32,213评论 3赞 312
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 37,576评论 3赞 298
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 28,893评论 0赞 17
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 30,171评论 1赞 250
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 41,486评论 2赞 341
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 40,676评论 2赞 335

Java中的并发工具类

CountDownLatch等待多线程完成

CyclicBarrier同步屏障

CyclicBarrier应用场景

CyclicBarrier和CountDownLatch的区别

控制并发线程数的Semaphore

线程之间交换数据Exchanger

推荐阅读更多精彩内容