华为拿下的这个5G编码到底是什么 华为拿下的这个5G编码到底是什么
乐视危机虚实：贾跃亭，一个梦想 乐视危机虚实：贾跃亭，一个梦想
半个美国网络瘫痪让中国企业负责 半个美国网络瘫痪让中国企业负责

软件测试 > 测试开发技术 > 软件测试开发语言 > Java >

关于多线程同步的初步教程－－Barrier的设计及使用

发表于：2007-05-25来源：作者：点击数： 标签：线程--Barrier初步同步教程
Barrier是一个多线程编程中经常要用到的同步工具，尤其多用于大数据量的计算过程中的同步。本文以广为流程的DougLea的concurrent工具包的Barrier实现为例，进行一点探讨。在DougLea的concurrent工具包中，Barrier是一个接口，在concurrent包中提供了两个Barr

Barrier是一个多线程编程中经常要用到的同步工具，尤其多用于大数据量的计算过程中的同步。本文以广为流程的Doug Lea的concurrent工具包的Barrier实现为例，进行一点探讨。在Doug Lea的concurrent工具包中，Barrier是一个接口，在concurrent包中提供了两个Barrier的实现：CyclicBarrier和Rendezvous。下面是Barrier接口的定义：

  1. public interface Barrier {
  4.   /** 
  5.    * Return the number of parties that must meet per barrier
  6.    * point. The number of parties is always at least 1.
  7.    **/
  9.   public int parties();
  11.   /**
  12.    * Returns true if the barrier has been compromised
  13.    * by threads leaving the barrier before a synchronization
  14.    * point (normally due to interruption or timeout). 
  15.    * Barrier methods in implementation classes throw
  16.    * throw BrokenBarrierException upon detection of breakage.
  17.    * Implementations may also support some means
  18.    * to clear this status.
  19.    **/
  21.   public boolean broken();
  22. }


    Barrier接口中的方法非常简单，parties()返回所有需要在屏障处同步的线程数；broken()返回一个标志，指示释放是否已被破坏。Barrier接口中并没有提供加入屏障的方法，而是在c和Rendezvous的Barrier实现中提供的。你可以会疑问，为什么不在Barrier接口中提供这些方法呢？因为这些实现的差异迥异，以至于很难在这些实现中提炼出一个共用的方法签名。比如，对于CyclicBarrier加入屏障的方法是：barrier()， 

  1. // CyclicBarrier.java
  2.   public int barrier() throws InterruptedException, BrokenBarrierException {
  3.     return doBarrier(false, 0);
  4.   }
  6.   protected synchronized int doBarrier(boolean timed, long msecs) 
  7.     throws InterruptedException, TimeoutException, BrokenBarrierException  { 
  8.     
  9.     int index = --count_;
  11.     if (broken_) {
  12.       throw new BrokenBarrierException(index);
  13.     }
  14.     else if (Thread.interrupted()) {
  15.       broken_ = true;
  16.       notifyAll();
  17.       throw new InterruptedException();
  18.     }
  19.     else if (index == 0) {  // tripped
  20.       count_ = parties_;
  21.       ++resets_;
  22.       notifyAll();
  23.       try {
  24.         if (barrierCommand_ != null)
  25.           barrierCommand_.run();
  26.         return 0;
  27.       }
  28.       catch (RuntimeException ex) {
  29.         broken_ = true;
  30.         return 0;
  31.       }
  32.     }
  33.     else if (timed && msecs <= 0) {
  34.       broken_ = true;
  35.       notifyAll();
  36.       throw new TimeoutException(msecs);
  37.     }
  38.     else {                   // wait until next reset
  39.       int r = resets_;      
  40.       long startTime = (timed)? System.currentTimeMillis() : 0;
  41.       long waitTime = msecs;
  42.       for (;;) {
  43.         try {
  44.           wait(waitTime);
  45.         }
  46.         catch (InterruptedException ex) {
  47.           // Only claim that broken if interrupted before reset
  48.           if (resets_ == r) { 
  49.             broken_ = true;
  50.             notifyAll();
  51.             throw ex;
  52.           }
  53.           else {
  54.             Thread.currentThread().interrupt(); // propagate
  55.           }
  56.         }
  58.         if (broken_) 
  59.           throw new BrokenBarrierException(index);
  61.         else if (r != resets_)
  62.           return index;
  64.         else if (timed) {
  65.           waitTime = msecs - (System.currentTimeMillis() - startTime);
  66.           if  (waitTime <= 0) {
  67.             broken_ = true;
  68.             notifyAll();
  69.             throw new TimeoutException(msecs);
  70.           }
  71.         }
  72.       }
  73.     }
  74.   }


而Rendezvous中则是：rendezvous(Object x)：

  1. // Rendezvous.java
  3.   public Object rendezvous(Object x) throws InterruptedException, BrokenBarrierException {
  4.     return doRendezvous(x, false, 0);
  5.   }
  7.   protected Object doRendezvous(Object x, boolean timed, long msecs) 
  8.     throws InterruptedException, TimeoutException, BrokenBarrierException {
  10.     // rely on semaphore to throw interrupt on entry
  12.     long startTime;
  14.     if (timed) {
  15.       startTime = System.currentTimeMillis();
  16.       if (!entryGate_.attempt(msecs)) {
  17.         throw new TimeoutException(msecs);
  18.       }
  19.     }
  20.     else {
  21.       startTime = 0;
  22.       entryGate_.acquire();
  23.     }
  25.     synchronized(this) {
  27.       Object y = null;
  29.       int index =  entries_++;
  30.       slots_[index] = x;
  32.       try { 
  33.         // last one in runs function and releases
  34.         if (entries_ == parties_) {
  36.           departures_ = entries_;
  37.           notifyAll();
  39.           try {
  40.             if (!broken_ && rendezvousFunction_ != null)
  41.             rendezvousFunction_.rendezvousFunction(slots_);
  42.           }
  43.           catch (RuntimeException ex) {
  44.             broken_ = true;
  45.           }
  47.         }
  49.         else {
  51.           while (!broken_ && departures_ < 1) {
  52.             long timeLeft = 0;
  53.             if (timed) {
  54.               timeLeft = msecs - (System.currentTimeMillis() - startTime);
  55.               if (timeLeft <= 0) {
  56.                 broken_ = true;
  57.                 departures_ = entries_;
  58.                 notifyAll();
  59.                 throw new TimeoutException(msecs);
  60.               }
  61.             }
  62.             
  63.             try {
  64.               wait(timeLeft); 
  65.             }
  66.             catch (InterruptedException ex) { 
  67.               if (broken_ || departures_ > 0) { // interrupted after release
  68.                 Thread.currentThread().interrupt();
  69.                 break;
  70.               }
  71.               else {
  72.                 broken_ = true;
  73.                 departures_ = entries_;
  74.                 notifyAll();
  75.                 throw ex;
  76.               }
  77.             }
  78.           }
  79.         }
  81.       }
  83.       finally {
  85.         y = slots_[index];
  86.         
  87.         // Last one out cleans up and allows next set of threads in
  88.         if (--departures_ <= 0) {
  89.           for (int i = 0; i < slots_.length; ++i) slots_[i] = null;
  90.           entryGate_.release(entries_);
  91.           entries_ = 0;
  92.         }
  93.       }
  95.       // continue if no IE/TO throw
  96.       if (broken_)
  97.         throw new BrokenBarrierException(index);
  98.       else
  99.         return y;
  100.     }
  101.   }


既然这样，那提供一个共用的Barrier接口还有什么意义呢？Doug Lea也觉察出了这个问题。所以在即将在JDK1.5中作为标准并发工具包发布的java.util.concurrent中，就去除了Barrier接口。

最常用的Barrier实现是CyclicBarrier，下面是CyclicBarrier的一个简单使用实例，Rendezvous的使用实例可以参考concurrent包的API文档。

  1.  class Solver {
  2.     final int N;
  3.     final float[][] data;
  4.     final CyclicBarrier barrier;
  5.     
  6.     class Worker implements Runnable {
  7.        int myRow;
  8.        Worker(int row) { myRow = row; }
  9.        public void run() {
  10.           while (!done()) {
  11.              processRow(myRow);
  12.  
  13.              try {
  14.                barrier.barrier(); 
  15.              }
  16.              catch (InterruptedException ex) { return; }
  17.              catch (BrokenBarrierException ex) { return; }
  18.           }
  19.        }
  20.     }
  21.  
  22.     public Solver(float[][] matrix) {
  23.       data = matrix;
  24.       N = matrix.length;
  25.       barrier = new CyclicBarrier(N);
  26.       barrier.setBarrierCommand(new Runnable() {
  27.         public void run() { mergeRows(...); }
  28.       });
  29.       for (int i = 0; i < N; ++i) {
  30.         new Thread(new Worker(i)).start();
  31.       waitUntilDone();
  32.      }
  33.  }


原文转自：http://www.ltesting.net

评论列表（网友评论仅供网友表达个人看法，并不表明本站同意其观点或证实其描述）

相关文章

软件测试沙龙 More>>

新浪微博More>>

...

热门标签