[Java并发包学习五]CountDownLatch和CyclicBarrier介绍

1

public CountDownLatch(int count)

12

public void await() throws InterruptedException
public boolean await(long timeout, TimeUnit unit)

1

public void countDown()

1

public void countDown()

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38
39
40
4142
43
44
45
46
47
48
49
50
5152
53
54
55
56
57
58
59
60
61

package com.winwill.test;

import java.util.Random;
import java.util.concurrent.CountDownLatch;

/**
* @author qifuguang
* @date 15/8/24 23:35
*/
public class TestCountDownLatch {
private static final int RUNNER_NUMBER = 5; // 运动员个数
private static final Random RANDOM = new Random();

public static void main(String[] args) {
// 用于判断发令之前运动员是否已经完全进入准备状态，需要等待5个运动员，所以参数为5
CountDownLatch readyLatch = new CountDownLatch(RUNNER_NUMBER);
// 用于判断裁判是否已经发令，只需要等待一个裁判，所以参数为1CountDownLatch startLatch = new CountDownLatch(1);
for (int i = 0; i < RUNNER_NUMBER; i++) {
Thread t = new Thread(new Runner((i + 1) + "号运动员", readyLatch, startLatch));
t.start();
}
try {
readyLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
startLatch.countDown();
System.out.println("裁判：所有运动员准备完毕，开始...");
}

static class Runner implements Runnable {
private CountDownLatch readyLatch;
private CountDownLatch startLatch;
private String name;

public Runner(String name, CountDownLatch readyLatch, CountDownLatch startLatch) {
this.name = name;
this.readyLatch = readyLatch;
this.startLatch = startLatch;
}

public void run() {
int readyTime = RANDOM.nextInt(1000);
System.out.println(name + "：我需要" + readyTime + "秒时间准备.");
try {
Thread.sleep(readyTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "：我已经准备完毕.");
readyLatch.countDown();
try {
startLatch.await();  // 等待裁判发开始命令
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "：开跑...");
}
}
}

12

public CyclicBarrier(int parties, Runnable barrierAction)
public CyclicBarrier(int parties)

1

public int getParties()

12
3
4
5

public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException

1

public boolean isBroken()

1

public void reset()

1

public int getNumberWaiting()

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38
39
40
4142
43
44
45
46
47
48
49
50
5152
53

package com.winwill.test;

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
* @author qifuguang
* @date 15/8/25 00:34
*/
public class TestCyclicBarrier {
private static final int THREAD_NUMBER = 5;
private static final Random RANDOM = new Random();

public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(THREAD_NUMBER, new Runnable() {
public void run() {
System.out.println(Thread.currentThread().getId() + "：我宣布，所有小伙伴写入数据完毕");
}
});
for (int i = 0; i < THREAD_NUMBER; i++) {
Thread t = new Thread(new Worker(barrier));
t.start();
}
}

static class Worker implements Runnable {
private CyclicBarrier barrier;

public Worker(CyclicBarrier barrier) {
this.barrier = barrier;
}

public void run() {
int time = RANDOM.nextInt(1000);
System.out.println(Thread.currentThread().getId() + "：我需要" + time + "毫秒时间写入数据.");
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：写入数据完毕，等待其他小伙伴...");
try {
barrier.await(); // 等待所有线程都调用过此函数才能进行后续动作
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：所有线程都写入数据完毕，继续干活...");
}
}
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38
39
40
4142
43
44
45
46
47
48
49
50
5152
53
54
55
56
57
58
59

package com.winwill.test;

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
* @author qifuguang
* @date 15/8/25 00:34
*/
public class TestCyclicBarrier {
private static final int THREAD_NUMBER = 5;
private static final Random RANDOM = new Random();

public static void main(String[] args) throws Exception {
CyclicBarrier barrier = new CyclicBarrier(THREAD_NUMBER, new Runnable() {
public void run() {
System.out.println(Thread.currentThread().getId() + "：我宣布，所有小伙伴写入数据完毕");
}
});
for (int i = 0; i < THREAD_NUMBER; i++) {
Thread t = new Thread(new Worker(barrier));
t.start();
}
Thread.sleep(10000);
System.out.println("================barrier重用==========================");
for (int i = 0; i < THREAD_NUMBER; i++) {
Thread t = new Thread(new Worker(barrier));
t.start();
}
}

static class Worker implements Runnable {
private CyclicBarrier barrier;

public Worker(CyclicBarrier barrier) {
this.barrier = barrier;
}

public void run() {
int time = RANDOM.nextInt(1000);
System.out.println(Thread.currentThread().getId() + "：我需要" + time + "毫秒时间写入数据.");
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：写入数据完毕，等待其他小伙伴...");
try {
barrier.await(); // 等待所有线程都调用过此函数才能进行后续动作
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：所有线程都写入数据完毕，继续干活...");
}
}
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38
39
40
4142
43
44
45
46
47
48
49
50
5152
53
54
55
56
57
58
59
60
6162
63

package com.winwill.test;

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* @author qifuguang
* @date 15/8/25 00:34
*/
public class TestCyclicBarrier {
private static final int THREAD_NUMBER = 5;
private static final Random RANDOM = new Random();

public static void main(String[] args) throws Exception {
CyclicBarrier barrier = new CyclicBarrier(THREAD_NUMBER, new Runnable() {
public void run() {
System.out.println(Thread.currentThread().getId() + "：我宣布，所有小伙伴写入数据完毕");
}
});
for (int i = 0; i < THREAD_NUMBER; i++) {
if (i < THREAD_NUMBER - 1) {
Thread t = new Thread(new Worker(barrier));
t.start();
} else {  //最后一个线程故意延迟3s创建。
Thread.sleep(3000);
Thread t = new Thread(new Worker(barrier));
t.start();
}
}
}

static class Worker implements Runnable {
private CyclicBarrier barrier;

public Worker(CyclicBarrier barrier) {
this.barrier = barrier;
}

public void run() {
int time = RANDOM.nextInt(1000);
System.out.println(Thread.currentThread().getId() + "：我需要" + time + "毫秒时间写入数据.");
try {
Thread.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：写入数据完毕，等待其他小伙伴...");
try {
barrier.await(2000, TimeUnit.MILLISECONDS); // 只等待2s，必然会等待最后一个线程超时
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getId() + "：所有线程都写入数据完毕，继续干活...");
}
}
}