Java并發(fā)之CyclicBarrier（在集合點(diǎn)同步）

  • CyclicBarrier引入
  • 創(chuàng)建CyclicBarrier
  • 遇到CyclicBarrier之后休眠
  • CyclicBarrier的回調(diào)線程
  • CyclicBarrier的簡(jiǎn)單例子
  • CyclicBarrier進(jìn)行分治編程的例子

CyclicBarrier引入

CyclicBarrier類(lèi)是一個(gè)同步輔助類(lèi)，類(lèi)似于CountDownLatch，但遠(yuǎn)比CountDownLatch要強(qiáng)大。CyclicBarrier 的字面意思是可循環(huán)使用（Cyclic）的屏障（Barrier）。它要做的事情是，讓一組線程到達(dá)一個(gè)屏障（也可以叫同步點(diǎn)）時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障時(shí)，屏障才會(huì)開(kāi)門(mén)，所有被屏障攔截的線程才會(huì)繼續(xù)干活。就如下面這個(gè)圖所示

image.png

CyclicBarrier相當(dāng)于一個(gè)屏障插在線程執(zhí)行的過(guò)程中，取決于線程調(diào)用await方法的位置，直到指定線程數(shù)量的到達(dá)之后，這個(gè)屏障才可以取走。

創(chuàng)建CyclicBarrier

當(dāng)你創(chuàng)建一個(gè)CyclicBarrier類(lèi)的時(shí)候，需要指定需要等待的線程數(shù)

CyclicBarrier barrier = new CyclicBarrier(2);

遇到CyclicBarrier之后休眠

當(dāng)在線程指定位置設(shè)置屏障的時(shí)候，只需要調(diào)用CyclicBarrier的await方法.

barrier.await();

await方法還可以指定等待的時(shí)間。當(dāng)達(dá)到這個(gè)等待的時(shí)間，即使沒(méi)有足夠的線程到達(dá)，這個(gè)屏障也會(huì)被解除

barrier.await(10, TimeUnit.SECONDS);

終止線程遇到屏障之后的等待條件有下面這些：

  • 足夠的線程到達(dá)屏障處，自動(dòng)解除屏障
  • 線程等待屏幕指定的等待時(shí)間之后，超時(shí)，解除屏障
  • 線程被中斷，其他線程被中斷，屏障會(huì)解除
  • 外部線程調(diào)用了CyclicBarrier.reset()方法，屏障解除。

CyclicBarrier的回調(diào)線程

CyclicBarrier初始化的時(shí)候，可以傳入一個(gè)runnable對(duì)象作為初始化參數(shù)，當(dāng)所有線程都到達(dá)屏障點(diǎn)后，屏障會(huì)先把這個(gè)指定的runnable對(duì)象作為線程來(lái)執(zhí)行，執(zhí)行完之后，就會(huì)移除屏障喚醒所有線程，這個(gè)特性很有作用，可以達(dá)到分治操作，fork/join。想象一下，我們讓線程在屏障前計(jì)算好各自的結(jié)果，然后當(dāng)所有線程都算完之后，我們?cè)诨卣{(diào)線程中執(zhí)行統(tǒng)計(jì)所有計(jì)算結(jié)果，這樣就相當(dāng)于分治技術(shù)了，將一個(gè)大任務(wù)切分給其他線程分成小任務(wù)各自執(zhí)行，執(zhí)行完之后就將他們匯總。

Runnable      barrierAction = ... ;
CyclicBarrier barrier       = new CyclicBarrier(2, barrierAction);

CyclicBarrier的簡(jiǎn)單例子

Runnable barrier1Action = new Runnable() {
    public void run() {
        System.out.println("BarrierAction 1 executed ");
    }
};
Runnable barrier2Action = new Runnable() {
    public void run() {
        System.out.println("BarrierAction 2 executed ");
    }
};

CyclicBarrier barrier1 = new CyclicBarrier(2, barrier1Action);
CyclicBarrier barrier2 = new CyclicBarrier(2, barrier2Action);

CyclicBarrierRunnable barrierRunnable1 =
        new CyclicBarrierRunnable(barrier1, barrier2);

CyclicBarrierRunnable barrierRunnable2 =
        new CyclicBarrierRunnable(barrier1, barrier2);

new Thread(barrierRunnable1).start();
new Thread(barrierRunnable2).start();

public class CyclicBarrierRunnable implements Runnable{

    CyclicBarrier barrier1 = null;
    CyclicBarrier barrier2 = null;

    public CyclicBarrierRunnable(
            CyclicBarrier barrier1,
            CyclicBarrier barrier2) {

        this.barrier1 = barrier1;
        this.barrier2 = barrier2;
    }

    public void run() {
        try {
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() +
                                " waiting at barrier 1");
            this.barrier1.await();

            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() +
                                " waiting at barrier 2");
            this.barrier2.await();

            System.out.println(Thread.currentThread().getName() +
                                " done!");

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

運(yùn)行結(jié)果

image.png

CyclicBarrier進(jìn)行分治編程的例子

我們實(shí)現(xiàn)一個(gè)CyclicBarrier分治編程的例子
我們假設(shè)現(xiàn)在一個(gè)數(shù)組中一個(gè)元素出現(xiàn)的次數(shù)，我們分出幾個(gè)線程分別計(jì)算不同的行，讓他們算完之后在屏障那里wait，然后等所有線程都算完了，我們就可以調(diào)用回調(diào)線程來(lái)計(jì)算總的結(jié)果

大數(shù)組類(lèi)

package CyclicBarrier;

import java.util.Random;

public class MatrixMock {
    private int[][] data;
    
    public MatrixMock(int size, int length, int number) {
        int counter = 0;
        data = new int[size][length];
        Random random = new Random();
        
        for(int i=0;i<size;i++) {
            for(int j=0;j<length;j++) {
                data[i][j] = random.nextInt(10);
                if(data[i][j] == number)
                    counter++;
            }
        }
        
        System.out.println("矩陣中有 " + counter + " 個(gè)要查找的數(shù) " + number);
    }
    
    public int[] getRow(int row) {
        if((row >= 0) && (row < data.length))
            return data[row];
        return null;
    }
}

結(jié)果類(lèi)：

package CyclicBarrier;

public class Results {
    private int[] data;
    
    public Results(int size) {
        data = new int[size];
    }
    
    public void setData(int position, int value) {
        data[position] = value;
    }
    
    public int[] getData() {
        return data;
    }
}

搜索線程

package CyclicBarrier;

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

public class Searcher implements Runnable {

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



    @Override
    public void run() {
        int counter;
        
        System.out.println(Thread.currentThread().getName() + "正在搜索數(shù)據(jù)" + firstRow + " " + lastRow);
        
        for(int i=firstRow;i<lastRow;i++) {
            int[] row = mock.getRow(i);
            counter = 0;
            for(int j=0;j<row.length;j++) {
                if(row[j] == number)
                    counter++;
            }
            results.setData(i, counter);
        }
        
        System.out.println(Thread.currentThread().getName() + "查完了");
        
        try {
            barrier.await();
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
        
        System.out.println(Thread.currentThread().getName() + "終于等到了");
    }

}

回調(diào)線程統(tǒng)計(jì)結(jié)果

package CyclicBarrier;

public class Grouper implements Runnable {

    private Results results;
    
    public Grouper(Results results) {
        this.results = results;
    }
    
    
    @Override
    public void run() {
        int finalResult = 0;
        
        System.out.println("正在統(tǒng)計(jì)結(jié)果。。。");
        
        int[] data = results.getData();
        
        for(int number : data) {
            finalResult += number;
        }
        
        System.out.println(finalResult);
    }

}

main類(lèi)測(cè)試

package CyclicBarrier;

import java.util.concurrent.CyclicBarrier;

public class Main {

    public static void main(String[] args) {
        
        final int ROWS = 10000;
        final int NUMBERS = 10000;
        final int SEACHER = 5;
        final int PARTICIPANTS = 5;
        final int LINES_PARTICIPANTS = 2000;
        
        MatrixMock mock = new MatrixMock(ROWS, NUMBERS, SEACHER);
        
        Results results = new Results(ROWS);
        
        Grouper grouper = new Grouper(results);
        
        CyclicBarrier barrier = new CyclicBarrier(PARTICIPANTS, grouper);
        
        Searcher[] searchers = new Searcher[PARTICIPANTS];
        
        for(int i=0;i<PARTICIPANTS;i++) {
            searchers[i] = new Searcher(i*LINES_PARTICIPANTS, i*LINES_PARTICIPANTS + LINES_PARTICIPANTS, mock, results, 5, barrier);
            new Thread(searchers[i]).start();
        }
    }

}

運(yùn)行結(jié)果

image.png
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容