Java多线程之同步工具类CyclicBarrier

public static void main(String[] args) {

        int N = 4 ;

        CyclicBarrier barrier  = new CyclicBarrier(N);

        for ( int i= 0 ;i<N;i++)

            new Writer(barrier).start();

    }

    static class Writer extends Thread{

        private CyclicBarrier cyclicBarrier;

        public Writer(CyclicBarrier cyclicBarrier) {

            this .cyclicBarrier = cyclicBarrier;

        }

        @Override

        public void run() {

            System.out.println( "线程" +Thread.currentThread().getName()+ "正在写入数据..." );

            try {

                Thread.sleep( 5000 );      //以睡眠来模拟写入数据操作

                System.out.println( "线程" +Thread.currentThread().getName()+ "写入数据完毕，等待其他线程写入完毕" );

                cyclicBarrier.await();

            } catch (InterruptedException e) {

                e.printStackTrace();

            } catch (BrokenBarrierException e){

                e.printStackTrace();

            }

            System.out.println( "所有线程写入完毕，继续处理其他任务..." );

        }

    }

public class CyclicBarrierTest {

     public static void main(String[] args) {

         int N = 4 ;

         CyclicBarrier barrier  = new CyclicBarrier(N, new Runnable() {

             @Override

             public void run() {

                 System.out.println( "当前线程" +Thread.currentThread().getName());

             }

         });

         for ( int i= 0 ;i<N;i++)

             new Writer(barrier).start();

     }

     static class Writer extends Thread{

         private CyclicBarrier cyclicBarrier;

         public Writer(CyclicBarrier cyclicBarrier) {

             this .cyclicBarrier = cyclicBarrier;

         }

         @Override

         public void run() {

             System.out.println( "线程" +Thread.currentThread().getName()+ "正在写入数据..." );

             try {

                 Thread.sleep( 3000 );      //以睡眠来模拟写入数据操作

                 System.out.println( "线程" +Thread.currentThread().getName()+ "写入数据完毕，等待其他线程写入完毕" );

                 cyclicBarrier.await();

             } catch (InterruptedException e) {

                 e.printStackTrace();

             } catch (BrokenBarrierException e){

                 e.printStackTrace();

             }

             System.out.println( "所有线程写入完毕，继续处理其他任务..." );

         }

     }

}

public class CyclicBarrierTest {

     public static void main(String[] args) {

         int N = 4 ;

         CyclicBarrier barrier = new CyclicBarrier(N);

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

             if (i < N - 1 )

                 new Writer(barrier).start();

             else {

                 try {

                     //运行时间远小于2000(cyclicBarrier.await 指定时间) 就不会抛出TimeoutException

                     Thread.sleep( 3000 );

                 } catch (InterruptedException e) {

                     e.printStackTrace();

                 }

                 new Writer(barrier).start();

             }

         }

     }

     static class Writer extends Thread {

         private CyclicBarrier cyclicBarrier;

         public Writer(CyclicBarrier cyclicBarrier) {

             this .cyclicBarrier = cyclicBarrier;

         }

         @Override

         public void run() {

             System.out.println( "线程" + Thread.currentThread().getName() + "正在写入数据..." );

             try {

                 Thread.sleep( 3000 );      //以睡眠来模拟写入数据操作

                 System.out.println( "线程" + Thread.currentThread().getName() + "写入数据完毕，等待其他线程写入完毕" );

                 try {

                     cyclicBarrier.await( 2000 , TimeUnit.MILLISECONDS);

                 } catch (TimeoutException e) {

                     e.printStackTrace();

                 }

             } catch (InterruptedException e) {

                 e.printStackTrace();

             } catch (BrokenBarrierException e) {

                 e.printStackTrace();

             }

             System.out.println(Thread.currentThread().getName() + "所有线程写入完毕，继续处理其他任务..." );

         }

     }

}

public class CyclicBarrierTest {

     public static void main(String[] args) {

         int N = 4 ;

         CyclicBarrier barrier  = new CyclicBarrier(N);

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

             new Writer(barrier).start();

         }

         try {

             Thread.sleep( 5000 );

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         System.out.println( "CyclicBarrier重用" );

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

             new Writer(barrier).start();

         }

     }

     static class Writer extends Thread{

         private CyclicBarrier cyclicBarrier;

         public Writer(CyclicBarrier cyclicBarrier) {

             this .cyclicBarrier = cyclicBarrier;

         }

         @Override

         public void run() {

             System.out.println( "线程" +Thread.currentThread().getName()+ "正在写入数据..." );

             try {

                 Thread.sleep( 3000 );      //以睡眠来模拟写入数据操作

                 System.out.println( "线程" +Thread.currentThread().getName()+ "写入数据完毕，等待其他线程写入完毕" );

                 cyclicBarrier.await();

             } catch (InterruptedException e) {

                 e.printStackTrace();

             } catch (BrokenBarrierException e){

                 e.printStackTrace();

             }

             System.out.println(Thread.currentThread().getName()+ "所有线程写入完毕，继续处理其他任务..." );

         }

     }

}

/** The lock for guarding barrier entry */

private final ReentrantLock lock = new ReentrantLock();

/** Condition to wait on until tripped */

private final Condition trip = lock.newCondition();

/** The number of parties */

private final int parties; //拦截的线程数量

/* The command to run when tripped */

private final Runnable barrierCommand; //当屏障撤销时，需要执行的屏障操作

//当前的Generation。每当屏障失效或者开闸之后都会自动替换掉。从而实现重置的功能。

private Generation generation = new Generation();

/**

  * Number of parties still waiting. Counts down from parties to 0

  * on each generation.  It is reset to parties on each new

  * generation or when broken.

  */

private int count;

//带Runnable参数的函数

  public CyclicBarrier( int parties, Runnable barrierAction) {

         if (parties <= 0 ) throw new IllegalArgumentException();

         this .parties = parties; //有几个运动员要参赛

         this .count = parties; //目前还需要几个运动员准备好

         //你要在所有线程都继续执行下去之前要执行什么操作，可以为空

         this .barrierCommand = barrierAction;

     }

//不带Runnable参数的函数

  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); // cannot happen

     }

private int dowait( boolean timed, long nanos)

      throws InterruptedException, BrokenBarrierException,

             TimeoutException {

      final ReentrantLock lock = this .lock;

      lock.lock(); //获取ReentrantLock互斥锁

      try {

          final Generation g = generation; //获取generation对象

          if (g.broken) //如果generation损坏，抛出异常

              throw new BrokenBarrierException();

          if (Thread.interrupted()) {

              //如果当前线程被中断，则调用breakBarrier方法，停止CyclicBarrier，并唤醒所有线程

              breakBarrier();

              throw new InterruptedException();

          }

          int index = --count; // 看到这里了吧，count减1

          //index=0，也就是说，有0个线程未满足CyclicBarrier条件，也就是条件满足，

          //可以唤醒所有的线程了

          if (index == 0 ) {  // tripped

              boolean ranAction = false ;

              try {

                 //这就是构造器的第二个参数，如果不为空的话，就执行这个Runnable的run方法，

                 //你看，这里是执行的是run方法，也就是说，并没有新起一个另外的线程，

                 //而是最后一个执行await操作的线程执行的这个run方法。

                  final Runnable command = barrierCommand;

                  if (command != null )

                      command.run(); //同步执行barrierCommand

                  ranAction = true ;

                  nextGeneration(); //执行成功设置下一个nextGeneration

                  return 0 ;

              } finally {

                  if (!ranAction) . //如果barrierCommand执行失败，进行屏障破坏处理

                      breakBarrier();

              }

          }

          //如果当前线程不是最后一个到达的线程

          // loop until tripped, broken, interrupted, or timed out

          for (;;) {

              try {

                  if (!timed)

                      trip.await(); //调用Condition的await()方法阻塞

                  else if (nanos > 0L)

                      nanos = trip.awaitNanos(nanos); //调用Condition的awaitNanos()方法阻塞

              } catch (InterruptedException ie) {

              //如果当前线程被中断，则判断是否有其他线程已经使屏障破坏。若没有则进行屏障破坏处理，并抛出异常；否则再次中断当前线程

                  if (g == generation && ! g.broken) {

                      breakBarrier(); //执行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) //如果当前generation已经损坏，抛出异常

                  throw new BrokenBarrierException();

              if (g != generation) //如果generation已经更新换代，则返回index

                  return index;

              //如果是参数是超时等待，并且已经超时，则执行breakBarrier()方法

              //唤醒所有等待线程。

              if (timed && nanos <= 0L) {

                  breakBarrier();

                  throw new TimeoutException();

              }

          }

      } finally {

          lock.unlock();

      }

  }

private Generation generation = new Generation();

private static class Generation {

     boolean broken = false ;

}

private void nextGeneration() {

     // signal completion of last generation

     trip.signalAll(); //调用Condition的signalAll方法，唤醒所有await的线程

     // set up next generation

     count = parties; //重置count值

     //生成新的Generation，表示上一代的所有线程已经唤醒，进行更新换代

     generation = new Generation();

}

private void breakBarrier() {

     generation.broken = true ; //表示当代因为线程被中断，已经发成损坏了

     count = parties; //重置count值

     trip.signalAll(); //调用Condition的signalAll方法，唤醒所有await的线程

}

public boolean isBroken() {

     final ReentrantLock lock = this .lock;

     lock.lock();

     try {

         return generation.broken;

     } finally {

         lock.unlock();

     }

}

//将屏障重置为其初始状态。

public void reset() {

     final ReentrantLock lock = this .lock;

     lock.lock();

     try {

         //唤醒所有等待的线程继续执行，并设置屏障中断状态为true

         breakBarrier();   // break the current generation

         //唤醒所有等待的线程继续执行，并设置屏障中断状态为false

         nextGeneration(); // start a new generation

     } finally {

         lock.unlock();

     }

}

//返回当前在屏障处等待的参与者数目，此方法主要用于调试和断言。

public int getNumberWaiting() {

     final ReentrantLock lock = this .lock;

     lock.lock();

     try {

         return parties - count;

     } finally {

         lock.unlock();

     }

}