Java並發包中CyclicBarrier的工作原理、使用示例

1. CyclicBarrier的介紹與源碼分析

CyclicBarrier 的字面意思是可循環（Cyclic）使用的屏障（Barrier）。它要做的事情是，讓一組線程到達一個屏障（也可以叫同步點）時被阻塞，直到最后一個線程到達屏障時，屏障才會開門，所有被屏障攔截的線程才會繼續干活。線程進入屏障通過CyclicBarrier的await()方法。

CyclicBarrier默認的構造方法是CyclicBarrier(int parties)，其參數表示屏障攔截的線程數量，每個線程調用await方法告訴CyclicBarrier我已經到達了屏障，然后當前線程被阻塞。

CyclicBarrier還提供一個更高級的構造函數CyclicBarrier(int parties, Runnable barrierAction)，用於在線程到達屏障時，優先執行barrierAction這個Runnable對象，方便處理更復雜的業務場景。

構造函數

 

public CyclicBarrier(int parties) {
    this(parties, null);
}
public int getParties() {
    return parties;
}

實現原理：在CyclicBarrier的內部定義了一個Lock對象，每當一個線程調用CyclicBarrier的await方法時，將剩余攔截的線程數減1，然后判斷剩余攔截數是否為0，如果不是，進入Lock對象的條件隊列等待。如果是，執行barrierAction對象的Runnable方法，然后將鎖的條件隊列中的所有線程放入鎖等待隊列中，這些線程會依次的獲取鎖、釋放鎖，接着先從await方法返回，再從CyclicBarrier的await方法中返回。

await源碼

 

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

 

dowait源碼

 

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        final Generation g = generation;

        if (g.broken)
            throw new BrokenBarrierException();

        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        int index = --count;
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    command.run();
                ranAction = true;
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // loop until tripped, broken, interrupted, or timed out
        for (;;) {
            try {
                if (!timed)
                    trip.await();
                else if (nanos > 0L)
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    // We're about to finish waiting even if we had not
                    // been interrupted, so this interrupt is deemed to
                    // "belong" to subsequent execution.
                    Thread.currentThread().interrupt();
                }
            }

            if (g.broken)
                throw new BrokenBarrierException();

            if (g != generation)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

當最后一個線程到達屏障點，也就是執行dowait方法時，會在return 0 返回之前調用finally塊中的breakBarrier方法。

breakBarrier源代碼

private void breakBarrier() {
    generation.broken = true;
    count = parties;
    trip.signalAll();
}

 

CyclicBarrier主要用於一組線程之間的相互等待，而CountDownLatch一般用於一組線程等待另一組些線程。實際上可以通過CountDownLatch的countDown()和await()來實現CyclicBarrier的功能。即 CountDownLatch中的countDown()+await() = CyclicBarrier中的await()。注意：在一個線程中先調用countDown()，然后調用await()。

其它方法：CycliBarrier對象可以重復使用，重用之前應當調用CyclicBarrier對象的reset方法。

reset源碼

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

 

2. 使用示例

package javalearning;

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

public class CyclicBarrierDemo {
	private CyclicBarrier cb = new CyclicBarrier(4);
	private Random rnd = new Random();
	
	class TaskDemo implements Runnable{
		private String id;
		TaskDemo(String id){
			this.id = id;
		}
		@Override
		public void run(){
			try {
				Thread.sleep(rnd.nextInt(1000));
				System.out.println("Thread " + id + " will wait");
				cb.await();
				System.out.println("-------Thread " + id + " is over");
			} catch (InterruptedException e) {
			} catch (BrokenBarrierException e) {
			}
		}
	}
	
	public static void main(String[] args){
		CyclicBarrierDemo cbd = new CyclicBarrierDemo();
		ExecutorService es = Executors.newCachedThreadPool();
		es.submit(cbd.new TaskDemo("a"));
		es.submit(cbd.new TaskDemo("b"));
		es.submit(cbd.new TaskDemo("c"));
		es.submit(cbd.new TaskDemo("d"));
		es.shutdown();
	}
}

 

在這個示例中，我們創建了四個線程a、b、c、d，這四個線程提交給了線程池。四個線程不同時間到達cb.await()語句，當四個線程都輸出“Thread x will wait”以后才會輸出“Thread x is over”。

運行結果

Thread d will wait

Thread a will wait

Thread c will wait

Thread b will wait

-------Thread b is over

-------Thread d is over

-------Thread a is over

-------Thread c is over

 

3. 參考內容

[1] http://ifeve.com/concurrency-cyclicbarrier/