Java并發(fā)系列之CyclicBarrier源碼分析

//同步操作鎖
private final ReentrantLock lock = new ReentrantLock();
//線程攔截器
private final Condition trip = lock.newCondition();
//每次攔截的線程數(shù)
private final int parties;
//換代前執(zhí)行的任務
private final Runnable barrierCommand;
//表示柵欄的當前代
private Generation generation = new Generation();
//計數(shù)器
private int count;

//靜態(tài)內(nèi)部類Generation
private static class Generation {
  boolean broken = false;
}

//構造器1
public CyclicBarrier(int parties, Runnable barrierAction) {
  if (parties <= 0) throw new IllegalArgumentException();
  this.parties = parties;
  this.count = parties;
  this.barrierCommand = barrierAction;
}

//構造器2
public CyclicBarrier(int parties) {
  this(parties, null);
}

//非定時等待
public int await() throws InterruptedException, BrokenBarrierException {
  try {
    return dowait(false, 0L);
  } catch (TimeoutException toe) {
    throw new Error(toe);
  }
}

//定時等待
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
  return dowait(true, unit.toNanos(timeout));
}

//核心等待方法
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()) {
      //如果當前線程被中斷會做以下三件事
      //1.打翻當前柵欄
      //2.喚醒攔截的所有線程
      //3.拋出中斷異常
      breakBarrier();
      throw new InterruptedException();
    }
    //每次都將計數(shù)器的值減1
    int index = --count;
    //計數(shù)器的值減為0則需喚醒所有線程并轉(zhuǎn)換到下一代
    if (index == 0) {
      boolean ranAction = false;
      try {
        //喚醒所有線程前先執(zhí)行指定的任務
        final Runnable command = barrierCommand;
        if (command != null) {
          command.run();
        }
        ranAction = true;
        //喚醒所有線程并轉(zhuǎn)到下一代
        nextGeneration();
        return 0;
      } finally {
        //確保在任務未成功執(zhí)行時能將所有線程喚醒
        if (!ranAction) {
          breakBarrier();
        }
      }
    }

    //如果計數(shù)器不為0則執(zhí)行此循環(huán)
    for (;;) {
      try {
        //根據(jù)傳入的參數(shù)來決定是定時等待還是非定時等待
        if (!timed) {
          trip.await();
        }else if (nanos > 0L) {
          nanos = trip.awaitNanos(nanos);
        }
      } catch (InterruptedException ie) {
        //若當前線程在等待期間被中斷則打翻柵欄喚醒其他線程
        if (g == generation && ! g.broken) {
          breakBarrier();
          throw ie;
        } else {
          //若在捕獲中斷異常前已經(jīng)完成在柵欄上的等待, 則直接調(diào)用中斷操作
          Thread.currentThread().interrupt();
        }
      }
      //如果線程因為打翻柵欄操作而被喚醒則拋出異常
      if (g.broken) {
        throw new BrokenBarrierException();
      }
      //如果線程因為換代操作而被喚醒則返回計數(shù)器的值
      if (g != generation) {
        return index;
      }
      //如果線程因為時間到了而被喚醒則打翻柵欄并拋出異常
      if (timed && nanos <= 0L) {
        breakBarrier();
        throw new TimeoutException();
      }
    }
  } finally {
    lock.unlock();
  }
}

//切換柵欄到下一代
private void nextGeneration() {
  //喚醒條件隊列所有線程
  trip.signalAll();
  //設置計數(shù)器的值為需要攔截的線程數(shù)
  count = parties;
  //重新設置柵欄代次
  generation = new Generation();
}

//打翻當前柵欄
private void breakBarrier() {
  //將當前柵欄狀態(tài)設置為打翻
  generation.broken = true;
  //設置計數(shù)器的值為需要攔截的線程數(shù)
  count = parties;
  //喚醒所有線程
  trip.signalAll();
}

class Horse implements Runnable {
  
  private static int counter = 0;
  private final int id = counter++;
  private int strides = 0;
  private static Random rand = new Random(47);
  private static CyclicBarrier barrier;
  
  public Horse(CyclicBarrier b) { barrier = b; }
  
  @Override
  public void run() {
    try {
      while(!Thread.interrupted()) {
        synchronized(this) {
          //賽馬每次隨機跑幾步
          strides += rand.nextInt(3);
        }
        barrier.await();
      }
    } catch(Exception e) {
      e.printStackTrace();
    }
  }
  
  public String tracks() {
    StringBuilder s = new StringBuilder();
    for(int i = 0; i < getStrides(); i++) {
      s.append("*");
    }
    s.append(id);
    return s.toString();
  }
  
  public synchronized int getStrides() { return strides; }
  public String toString() { return "Horse " + id + " "; }
  
}

public class HorseRace implements Runnable {
  
  private static final int FINISH_LINE = 75;
  private static List<Horse> horses = new ArrayList<Horse>();
  private static ExecutorService exec = Executors.newCachedThreadPool();
  
  @Override
  public void run() {
    StringBuilder s = new StringBuilder();
    //打印賽道邊界
    for(int i = 0; i < FINISH_LINE; i++) {
      s.append("=");
    }
    System.out.println(s);
    //打印賽馬軌跡
    for(Horse horse : horses) {
      System.out.println(horse.tracks());
    }
    //判斷是否結束
    for(Horse horse : horses) {
      if(horse.getStrides() >= FINISH_LINE) {
        System.out.println(horse + "won!");
        exec.shutdownNow();
        return;
      }
    }
    //休息指定時間再到下一輪
    try {
      TimeUnit.MILLISECONDS.sleep(200);
    } catch(InterruptedException e) {
      System.out.println("barrier-action sleep interrupted");
    }
  }
  
  public static void main(String[] args) {
    CyclicBarrier barrier = new CyclicBarrier(7, new HorseRace());
    for(int i = 0; i < 7; i++) {
      Horse horse = new Horse(barrier);
      horses.add(horse);
      exec.execute(horse);
    }
  }
  
}