[笔记][Java7并发编程实战手册]3.5 在集合点的同步CyclicBarrier循环barrier

[笔记][Java7并发编程实战手册]系列目录

CyclicBarrier详细原理解说，可先查看别人的博客：/article/4708153.html

简介

CyclicBarrier是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。

注意比较CountDownLatch和CyclicBarrier：

1. CountDownLatch的作用是允许1或N个线程等待其他线程完成执行；而CyclicBarrier则是允许N个线程相互等待。

2. CountDownLatch的计数器无法被重置；CyclicBarrier的计数器可以被重置后使用，因此它被称为是循环的barrier。

下面两个伪代码来看懂 他们的区别和用法

class CountDownLatch{
CountDownLatch cd = new CountDownLatch(5);
main(){
启动5个线程。。。
cd.await(); //让当前线程休眠，
执行代码... (该代码不会立即执行，等待5个线程执行完成后，当前线程会被唤醒，继续执行后面的代码)
}

class Th extends Thread{
public void run() {
执行代码...
cd.countDown(); //报告CountDownLatch 当前线程已经到位
执行代码...（该代码不会因为countDown()而阻塞，countDown()没有阻塞效果）

}
}

}

// 没有指定 构造线程的barrier伪代码
class CyclicBarrier{
CyclicBarrier cd = new CyclicBarrier(5);
main(){
启动5个线程。。。
执行代码...; //该代码会被立即执行
}

class Th extends Thread{
public void run() {
执行代码...
cb.await(); //barrier参与量+1，并且当前线程休眠
执行代码...（该代码不会立即执行，只有等待指定的5个线程集合了后，该代码才会继续执行）
}
}

}

// 指定 构造线程的barrier伪代码
class CyclicBarrier{
CyclicBarrier cd = new CyclicBarrier(5，runable);
main(){
启动5个线程。。。
执行代码...; //该代码会被立即执行
}

class Th extends Thread{
public void run() {
执行代码...
cb.await(); //barrier参与量+1，并且当前线程休眠
执行代码...（该代码不会立即执行，只有等待指定的5个线程集合了后，会启动构造传入的runable线程代码，该线程结束后（所有等待的线程才能被唤醒），该代码才会继续执行）
}
}

}

CyclicBarrier使用心得

await():将CyclicBarrie参与量+1，并且让当前线程 休眠，直到指定的 参与量达到了数量在CyclicBarrie等待的所有线程将被唤醒。

new CyclicBarrier(5, new Runnable(）)： 构造指定的参与量，当参与量达到5的时候，将会启动传入的线程。

构造barrier传入的线程,将会由最后一个集合的线程启动运行（看效果看出来的，没有看源码）

构造barrier传入的线程，被启动后，只有该线程结束后，在该barrier上等待的线程才能被唤醒

reset()将屏障重置为其初始状态的时候，要注意，如果该barrier上有等待的线程，等待的线程将会抛出BrokenBarrierException异常

既然称为可循环的，下面的伪代码，注意到 cb.await()可以被执行多次，也就是说能循环的集合

// 指定 构造线程的barrier伪代码
class CyclicBarrier{
CyclicBarrier cd = new CyclicBarrier(5，runable);
main(){
启动5个线程。。。
执行代码...;
}

class Th extends Thread{
public void run() {
执行代码...
cb.await(); // 当参与者数量达到了指定的数量（也就是集合点），runable线程将被执行一次，
执行代码...
cb.await(); // 然而，这个await可以重复使用，也就是说，只要集合一次，runable线程将被执行一次
执行代码...
}
}

}

用一个简单的示例来介绍CyclicBarrier的用法

/**
* Created by zhuqiang on 2015/8/22 0022.
*/
public class Clinet2 {

private static int SIZE = 5;
private static CyclicBarrier cb;

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

Runnable barrierAction = new Runnable() {
public void run() {
//                cb.reset();  //重置为初始状态，如果当前屏障上有等待线程，将抛出BrokenBarrierException异常
System.out.println("barrierAction:必须同时到达barrier的线程个数:" + cb.getParties() + ";在该barrier上等待的线程数量： " + cb.getNumberWaiting());
}
};
cb = new CyclicBarrier(SIZE, barrierAction);

// 新建5个任务
for (int i = 0; i < SIZE; i++) {
InnerThread innerThread = new InnerThread();
innerThread.start();
}
Thread.sleep(5000);  //休眠5秒，等待以上5个线程会执行完毕后，再调用cb.reset()的话。 就不会抛出异常了

System.out.println("必须同时到达barrier的线程个数:" + cb.getParties() + ";在该barrier上等待的线程数量： " + cb.getNumberWaiting());
}

static class InnerThread extends Thread {
public void run() {
try {
System.out.println(Thread.currentThread().getName() + " wait for CyclicBarrier. 在该barrier上等待的线程数量：" + cb.getNumberWaiting());

// 将cb的参与者数量加1
cb.await();

// cb的参与者数量等于5时，才继续往后执行
System.out.println(Thread.currentThread().getName() + " continued.");
} catch (BrokenBarrierException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

某一次运行结果

Thread-0 wait for CyclicBarrier. 在该barrier上等待的线程数量：0
Thread-4 wait for CyclicBarrier. 在该barrier上等待的线程数量：0
Thread-3 wait for CyclicBarrier. 在该barrier上等待的线程数量：0
Thread-2 wait for CyclicBarrier. 在该barrier上等待的线程数量：0
Thread-1 wait for CyclicBarrier. 在该barrier上等待的线程数量：0
barrierAction:必须同时到达barrier的线程个数:5;在该barrier上等待的线程数量： 5
Thread-1 continued.
Thread-0 continued.
Thread-3 continued.
Thread-2 continued.
Thread-4 continued.
必须同时到达barrier的线程个数:5;在该barrier上等待的线程数量： 0

稍微复杂的示例：分治编程技术

该示例是书籍上的demo，以下示例将的是一个： 用5个线程在一个矩阵中，查找指定的数字，得出出现的次数是多少。

MatrixMock 生成矩阵对象，Searcher 线程负责查找自己线程指定范围内的结果，Client中的一个线程 用来等待统计5个Searcher运行之后的结果。

/**
* Created by zhuqiang on 2015/8/22 0022.
* 随机矩阵
*/
public class MatrixMock {
private int[][] data;

/**
*
* @param size 矩阵行数
* @param cols 矩阵列数
* @param number 要寻找的数字
*/
public MatrixMock(int size,int cols,int number) {
data = new int[size][cols];
Random random = new Random();

int counter = 0;
//  用随机数为矩阵赋值。每生成一个字，就用它跟要查找的数字比较，进行比较。如果一致，就用计数器加1
for (int i = 0; i < size; i++) {
for (int j = 0; j < cols; j++) {
data[i][j] = random.nextInt(10);
if(data[i][j] == number){
counter++;
}
}
}
//用来验证多线程查找的正确性
System.out.printf("在矩阵中找到了数字：%d,%d次\n",number,counter);
//        测试的时候，可以放开此代码，能打印出 矩阵分布图。当然需要矩阵10 * 10 比较小的收，控制台才能装得下
//        for (int i = 0; i < data.length; i++) {
//            for (int j = 0; j < data[i].length; j++) {
//                System.out.printf(data[i][j] + " | ");
//            }
//            System.out.println("");
//        }
}

/**
* 返回指定矩阵中的行数据
* @param row 行号
* @return
*/
public int[] getRow(int row){
if(row >= 0 && row < data.length){
return data[row];
}
return  null;
}
}

/**
* Created by zhuqiang on 2015/8/22 0022.
* 结果类
*/
class Results {
private int[] data;

public Results(int size) {
data = new int[size];
}
public void setRowResult(int index,int value){
data[index] = value;
}
public int[] getData(){
return data;
}
}

/**
* Created by zhuqiang on 2015/8/22 0022.
*/
public class Searcher implements Runnable {
// 查找子集范围的开始行数和结束行数
private int firstRow;
private int lastRow;
private MatrixMock mock ; //矩阵类
private Results results; //结果类
private int number;  //要查找的数字
private CyclicBarrier barrier;  //线程辅助类

public Searcher(int firstRow, int lastRow, MatrixMock mock, Results results, int number, CyclicBarrier barrier) {
this.firstRow = firstRow;
this.lastRow = lastRow;
this.mock = mock;
this.results = results;
this.number = number;
this.barrier = barrier;
}

@Override
public void run() {
System.out.printf("%s，查找范围是：%d 》 %d rows,查找开始..........................................\n",Thread.currentThread().getName(),firstRow,lastRow);
int num = 0;  //记录当前线程 总共在自己所查询的子集中匹配了多少次
for (int i = firstRow; i < lastRow ; i++) {
int counter = 0;   //记录每一行的查找次数
int[] row = mock.getRow(i);  //根据行号获取矩阵中的行数据，然后逐一对比
for (int j = 0; j < row.length; j++) {
if(row[j] == number){
counter++;
num++;
}
}
results.setRowResult(i,counter);  // 把行号，和匹配次数结果放到结果集里
counter = 0;  // 重置每行结果

}
System.out.printf("%s，查找结束，共查找了%d个匹配项.\n",Thread.currentThread().getName(),num);
try {
barrier.await(); //让当前线程睡眠，直到 barrier的参与者数量达到指定数量的时候，下面的代码才继续执行
System.out.println(Thread.currentThread().getName() + " 等待结束");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}

/**
* Created by zhuqiang on 2015/8/22 0022.
*/
public class Client implements Runnable {
private Results results;

public Client(Results results) {
this.results = results;
}

// 统计结果的线程
@Override
public void run() {
int count = 0;
int[] data = results.getData();
for (int i = 0; i < data.length; i++) {
count += data[i];
}
System.out.printf("%s统计线程，查找的的匹配次数是：%d\n", Thread.currentThread().getName(), count);
//        try {   //可以看到 只有当该线程执行完毕之后，等待线程才能被唤醒
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
}

// 测试方法
public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
final int rows = 1000;  // 矩阵行数
final int cols = 1000;  //矩阵列数
final int searcher = 5;  //查找线程个数
final int number = 5;  //查找的数字

MatrixMock mock = new MatrixMock(rows,cols,number);  //生成矩阵对象
Results results = new Results(rows);  //结果对象，有多少行，就生成多少结果，每一行的数据匹配项累加
Client client = new Client(results);  //统计结果运算线程

CyclicBarrier barrier = new CyclicBarrier(searcher,client);  //创建barrier， 等待5个线程执行执行到指定的屏障点，再唤醒所有在该屏障点睡眠的线程，同时启动统计结果线程，统计出每个线程得到的结果。

Searcher[] list = new Searcher[searcher];

int handerCols = rows/searcher;  //计算出每个线程执行的子集数量

for (int i = 0; i < list.length; i++) {
int firstRow = i * handerCols;
int lastRow = firstRow + handerCols;

list[i] =  new Searcher(firstRow,lastRow,mock,results,number,barrier);
new Thread(list[i]).start();
}
System.out.println("main线程执行完毕了");
}
}

某一次的运行示例：

在矩阵中找到了数字：5,100001次
Thread-0，查找范围是：0 》 200 rows,查找开始..........................................
Thread-4，查找范围是：800 》 1000 rows,查找开始..........................................
main线程执行完毕了
Thread-3，查找范围是：600 》 800 rows,查找开始..........................................
Thread-2，查找范围是：400 》 600 rows,查找开始..........................................
Thread-1，查找范围是：200 》 400 rows,查找开始..........................................
Thread-3，查找结束，共查找了20039个匹配项.
Thread-4，查找结束，共查找了20008个匹配项.
Thread-0，查找结束，共查找了20052个匹配项.
Thread-1，查找结束，共查找了19975个匹配项.
Thread-2，查找结束，共查找了19927个匹配项.
Thread-2统计线程，查找的的匹配次数是：100001
Thread-3 等待结束
Thread-1 等待结束
Thread-4 等待结束
Thread-2 等待结束
Thread-0 等待结束

为了方便看到矩阵中的结构，放开注释中的 打印矩阵结构代码，并且把矩阵的大小调成 10*10,放开以下代码

//        测试的时候，可以放开此代码，能打印出 矩阵分布图。当然需要矩阵10 * 10 比较小的收，控制台才能装得下
for (int i = 0; i < data.length; i++) {
for (int j = 0; j < data[i].length; j++) {
System.out.printf(data[i][j] + " | ");
}
System.out.println("");
}

修改矩阵大小：

final int rows = 10;  // 矩阵行数
final int cols = 10;  //矩阵列数

某一次的运行结果：

在矩阵中找到了数字：5,7次
1 | 1 | 2 | 8 | 1 | 7 | 2 | 2 | 8 | 7 |
1 | 8 | 7 | 6 | 6 | 1 | 0 | 1 | 7 | 6 |
7 | 7 | 7 | 8 | 3 | 9 | 5 | 0 | 7 | 8 |
4 | 7 | 1 | 7 | 9 | 0 | 3 | 1 | 2 | 3 |
8 | 0 | 3 | 2 | 5 | 5 | 6 | 1 | 1 | 5 |
8 | 0 | 9 | 8 | 1 | 8 | 8 | 5 | 8 | 8 |
2 | 2 | 9 | 5 | 4 | 1 | 6 | 8 | 2 | 8 |
6 | 2 | 0 | 4 | 1 | 7 | 2 | 2 | 7 | 2 |
9 | 5 | 6 | 2 | 0 | 3 | 0 | 1 | 1 | 8 |
6 | 0 | 1 | 8 | 4 | 1 | 4 | 2 | 2 | 6 |
Thread-0，查找范围是：0 》 2 rows,查找开始..........................................
Thread-4，查找范围是：8 》 10 rows,查找开始..........................................
Thread-4，查找结束，共查找了1个匹配项.
Thread-3，查找范围是：6 》 8 rows,查找开始..........................................
Thread-3，查找结束，共查找了1个匹配项.
Thread-2，查找范围是：4 》 6 rows,查找开始..........................................
Thread-2，查找结束，共查找了4个匹配项.
main线程执行完毕了
Thread-1，查找范围是：2 》 4 rows,查找开始..........................................
Thread-0，查找结束，共查找了0个匹配项.
Thread-1，查找结束，共查找了1个匹配项.
Thread-1统计线程，查找的的匹配次数是：7
Thread-1 等待结束
Thread-3 等待结束
Thread-0 等待结束
Thread-4 等待结束
Thread-2 等待结束