java併發之同步輔助類CyclicBarrier

CyclicBarrier含義：

柵欄容許兩個或者多個線程在某個集合點同步。當一個線程到達集合點時，它將調用await()方法等待其它的線程。線程調用await()方法後，CyclicBarrier將阻塞這個線程並將它置入休眠狀態等待其它線程的到來。等最後一個線程調用await()方法時，CyclicBarrier將喚醒全部等待的線程而後這些線程將繼續執行。CyclicBarrier能夠傳入另外一個Runnable對象做爲初始化參數。當全部的線程都到達集合點後，CyclicBarrier類將Runnable對象做爲線程執行。

 

方法：

await()：使線程置入休眠直到最後一個線程的到來以後喚醒全部休眠的線程

 

例子

在矩陣（二維數組）中查找一個指定的數字。矩陣將被分爲多個子集，每一個子集交給一個線程去查找。當全部線程查找完畢後交給最後的線程彙總結果。

查找類：在一個子集中查找指定數字，找到以後把結果存儲後調用await()方法置入休眠等待最後一個線程的到來喚醒

import java.util.List;

import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;

 

public class Searcher implements Runnable {

 

private  CyclicBarrier barrier;

 

private  int[] submock;

 

private  List<Result> result;

 

private int row;

 

private int searchNmu;

 

public Searcher(int[] submock, List<Result> result,  CyclicBarrier barrier, int row, int searchNmu) {

this.barrier = barrier;

this.submock = submock;

this.result = result;

this.row = row;

this.searchNmu = searchNmu;

}

 

 

@Override

public void run() {

System.out.printf("%s: Processing lines from %d .\n", Thread.currentThread().getName(), row);

for(int i=0; i<submock.length; i++){

if(submock[i] == searchNmu){

Result r = new Result();

r.setRow(row);

r.setCol(i);

result.add(r);

}

}

System.out.printf("%s: Lines processed.\n", Thread.currentThread().getName());

try {

barrier.await();

} catch (InterruptedException e) {

e.printStackTrace();

} catch (BrokenBarrierException e) {

e.printStackTrace();

}

}

}

 

結果類：

 

public class Result {

 

//行

int row;

//列

int col;

 

public int getRow() {

return row;

}

 

public void setRow(int row) {

this.row = row;

}

 

public int getCol() {

return col;

}

 

public void setCol(int col) {

this.col = col;

}

 

}

 

彙總類：彙總每一個Searcher找到的結果:

 

import java.util.List;

 

public class Grouper implements Runnable {

 

private List<Result> result;

 

int[][] mock;

 

public Grouper(List<Result> result, int[][] mock) {

this.result = result;

this.mock = mock;

}

 

@Override

public void run() {

System.out.printf("Grouper: Processing results...\n");

for (int i = 0; i < result.size(); i++) {

Result r = result.get(i);

if(r!=null)

System.out.println("mock[" + r.row + "][" + r.col + "]" + mock[r.row][r.col]);

}

System.out.printf("Grouper proccessing end...\n");

}

}

 

主函數,如何把Searcher和Grouper類配合起來呢？？

 

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.CyclicBarrier;

 

public class CyclicBarrierMain {

 

public static void main(String[] args) {

// 要找的數據

final int SEARCH = 5;

 

// 矩陣的聲明

int[][] mock = { { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 },

{ 1, 2, 3, 5, 5, 6, 7, 8, 9, 10 },

{ 5, 2, 3, 4, 5, 6, 7, 8, 9, 10 }, 

{ 1, 2, 3, 4, 6, 6, 7, 8, 5, 10 }, 

{ 1, 5, 3, 4, 5, 6, 7, 8, 5, 10 },

{ 1, 5, 3, 4, 12, 6, 7, 8, 0, 5 } };

// 查找的線程數

int PARTICIPANTS = mock.length;

List<Result> result = new ArrayList<Result>();

// 彙總線程

Grouper grouper = new Grouper(result, mock);

// 柵欄，傳入參數含義：線程同步個數，彙總線程

CyclicBarrier barrier = new CyclicBarrier(PARTICIPANTS, grouper);

 

Searcher searchers[] = new Searcher[PARTICIPANTS];

 

for (int i = 0; i < PARTICIPANTS; i++) {

searchers[i] = new Searcher(mock[i], result, barrier, i, SEARCH);

Thread thread = new Thread(searchers[i]);

thread.start();

}

System.out.printf("Main: The main thread has finished.\n");

}

 

}

 

須要注意的地方

線程完成任務後調用CyclicBarrier的await(）方法休眠等待。在全部線程在集合點均到達時，柵欄調用傳入的Runnable對象進行最後的執行。

與CountDownLatch的區別：

• 在全部線程到達集合點後接受一個Runnable類型的對象做爲後續的執行

• 沒有顯示調用CountDown()方法

• CountDownLatch通常只能使用一次，CyclicBarrier能夠屢次使用

應用場景

多個線程作任務，等到達集合點同步後交給後面的線程作彙總

 

海量it視頻獲取