欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

线程同步辅助类CyclicBarrier笔记

程序员文章站 2022-04-20 16:18:03
...

    CyclicBarrier/关卡,多个线程在关卡处同步,是CountDownLatch的加强版,多了一些有用的新特性。

    Demo(Java7并发编程):在一个巨大的矩阵中寻找某个数字,任务可能相当耗时,于是多个线程分配好任务后一起上,人多力量大,大家都完成任务后,到组长那里汇报结果,组长再对结果进行统计(分治法)。

    1.模拟矩阵

package java7.Lesson5_CyclicBarrier;

import java.util.Random;

import util.Util;

public class MatrixMock {
	private int[][] data;
	
	public MatrixMock(int width,int length,int num){
		int counter = 0;
		this.data = new int[width][length];
		Random r = new Random();
		for (int i=0;i<width;i++){
			for (int j=0;j<length;j++){
				data[i][j]=r.nextInt(10);
				if(data[i][j]==num){
					counter++;
				}
			}
		}
		Util.sop("等于"+num+"的数字有"+counter+"个");//先知道答案,核对用
	}
	
	//返回矩阵某行数据
	public int[] getRow(int row){
		if ((row >= 0)&&(row < data.length)){
			return data[row];
		}else{
			return null;
		}
	}	
}

     2.存放中间结果,即寻找线程生产的数据

package java7.Lesson5_CyclicBarrier;

public class Results {
	private int[] data;
	
	public Results(int width){
		this.data = new int[width];
	}
	
	//此处不用实现同步,调用者保证不使用相同position
	public void setDate(int position,int val){
		data[position] = val;
	}
	
	public int[] getData(){
		return data;
	}
}

   3.搜索者

package java7.Lesson5_CyclicBarrier;

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

import util.Util;

public class Searcher implements Runnable {
	private int firstRow;
	private int lastRow;
	private MatrixMock mock;
	private Results results;
	private int num;
	private final CyclicBarrier barrier;

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

	@Override
	public void run() {
		int counter;
		System.out.printf("%s:查找范围是从%d到%d\n", Thread.currentThread().getName(),
				this.firstRow, this.lastRow);
		// 查找
		for (int i = this.firstRow; i < this.lastRow; i++) {
			int[] row = mock.getRow(i);
			counter = 0;
			for (int j=0;j<row.length;j++){
				if(num==row[j]){
					counter++;
				}
			}
			results.setDate(i, counter);
		}
		Util.sop(Thread.currentThread().getName()+":搜索结束");
		//等待
		try {
			barrier.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (BrokenBarrierException e) {
			e.printStackTrace();
		}
	}

}

    说明:

     (1).大家注意到与CountDownLatch的await()方法不同,这里的await()方法抛出2个异常,InterruptedException是中断异常, BrokenBarrierException是关卡破坏异常,顾名思义,出现这个异常时,关卡被破坏,不再有意义。

   (2).CyclicBarrier有个特殊状态(Broken),当有多个线程阻塞在await()处,某个线程突然中断,抛出InterruptedException,其它等待线程将抛出BrokenBarrierException。说好在一起你却突然不干了,大家只能散伙线程同步辅助类CyclicBarrier笔记
            
    
    博客分类: nio 线程同步CyclicBarrier ,这时Cyclicbarrier处于Broken状态,用isBroken()方法可以获取。

   (3).另外,Cyclicbarrier是可以用reset()方法重置的,此时await()处阻塞的线程也会抛出BrokenBarrierException异常,这个也是可以理解的,组长宣布本次无效,大家只能乖乖从头再来。

 

  4.负责统计的组长

package java7.Lesson5_CyclicBarrier;

import util.Util;

public class Grouper implements Runnable{

	private Results results;
	
	public Grouper(Results results){
		this.results = results;
	}
	
	@Override
	public void run() {
		int finalResults = 0;
		for(int d : results.getData()){
			finalResults += d;
		}
		Util.sop("最终结果是:"+finalResults);
	}
}

  5.测试

package java7.Lesson5_CyclicBarrier;

import java.util.concurrent.CyclicBarrier;

public class Client {

	public static void main(String[] args) {
		final int rows = 10000;
		final int numbers = 1000;
		final int search = 5;
		final int participant = 5;
		final int lines_participant = 2000;

		MatrixMock mock = new MatrixMock(rows, numbers, search);
		Results results = new Results(rows);
		Grouper grouper = new Grouper(results);
		CyclicBarrier barrier = new CyclicBarrier(participant, grouper);
		Searcher[] searchers = new Searcher[participant];
		for (int i = 0; i < participant; i++) {
			searchers[i] = new Searcher(i * lines_participant, i
					* lines_participant + lines_participant, mock, results, 5,
					barrier);
			new Thread(searchers[i]).start();
		}
	}

}

   说明:大家注意到:new CyclicBarrier(participant, grouper),当所有searcher线程通过关卡后grouper线程会被执行。员工干完活后,组长开始统计。

 

  测试结果

等于5的数字有998605个
Thread-2:查找范围是从4000到6000
Thread-4:查找范围是从8000到10000
Thread-3:查找范围是从6000到8000
Thread-4:搜索结束
Thread-3:搜索结束
Thread-0:查找范围是从0到2000
Thread-1:查找范围是从2000到4000
Thread-0:搜索结束
Thread-1:搜索结束
Thread-2:搜索结束
最终结果是:998605

  结果无误,符合预期。 

 

    CyclicBarrier是可以反复重置的,这个例子没有体现,另外CyclicBarrier的await()方法会给所有通过关卡的线程返回一个唯一的到达索引号,这对我们挑选出“领导者”线程有用(大家可以看一下领导者跟随者线程模型)