同步工具类闭锁，栅栏，信号量

时间：2021-04-15 15:16:39

概述

同步工具类可以是任何一个类，只要它根据其自身的状态来协调线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具还包括信号量（Semaphore），栅栏（Latch），闭锁(Latch).

闭锁

闭锁可以延迟线程的进度直到到达中止状态，闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并允许所有的线程通过。

CountDownLatch

API:

        CountDownLatch的构造函数接受一个int类型的参数作为计数器，如果你想要等待N个点完成，这里就传入N。

　　调用countDown()方法，N就会减一，所以在计数到达零之前，await()方法会阻塞当前线程。之后，会释放所有等待的线程,await()的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用CyclicBarria。由于countDown()方法可以用在任何地方，所以这里说的N个点可以是N个线程，也可以是１个线程里的N个执行步骤。

　　await(long time,TimeUnit unit),这个方法等待特定时间后，就不会阻塞当前线程。

简单入门：

public class CountDownLatchTest {
    private static java.util.concurrent.CountDownLatch c=new java.util.concurrent.CountDownLatch(2);
    public static void main(String[] args) throws InterruptedException {
    new Thread(
    ){
        @Override
        public void run(){
            System.out.println(1);
            c.countDown();
            System.out.println(2);
            c.countDown();
        }
    }.start();
    c.await();
        System.out.println(3);
}
}
输出：
１
２
３


进阶用法： 下面给出了两个类，其中一组 worker 线程使用了两个倒计数锁存器：

  • 第一个类是一个启动信号，在 driver 为继续执行 worker 做好准备之前，它会阻止所有的 worker 继续执行。
  • 第二个类是一个完成信号，它允许 driver 在完成所有 worker 之前一直等待。
    import java.util.concurrent.CountDownLatch;

public class Worker implements Runnable{
private final CountDownLatch startSignal;
private  final CountDownLatch doneSignal;
private CountDownLatch startSignal2;
Worker(CountDownLatch startSignal,CountDownLatch doneSignal){
    this.startSignal=startSignal;
    this.doneSignal=doneSignal;
}
public void  run(){
try {
    startSignal.await();
    work();
    doneSignal.countDown();
} catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
}

}public void work(){
    
}
}
  •  
    •  




public class Driver {
public void main() throws InterruptedException{
	CountDownLatch startSignal=new CountDownLatch(1);//开始信号
	CountDownLatch doneSignal=new CountDownLatch(5);//结束信号
for(int i=0;i<5;i++){
	new Thread(
			new Worker(startSignal,doneSignal)
			).start();
}
doSomethingElse();
startSignal.countDown();//开始所有工作
doSomethingElse();
doneSignal.await();//等待所有的工作结束
}
public void doSomethingElse(){}
}

 

 

另一种典型用法是，将一个问题分成 N 个部分，用执行每个部分并让锁存器倒计数的 Runnable 来描述每个部分，然后将所有 Runnable 加入到 Executor 队列。当所有的子部分完成后，协调线程就能够通过 await。（当线程必须用这种方法反复倒计数时，可改为使用CyclicBarrier。）
 

 

public class Worker1  implements Runnable{
	   private final CountDownLatch doneSignal;
	   private final int i;
	   Worker1(CountDownLatch doneSignal, int i) {
	      this.doneSignal = doneSignal;
	      this.i = i;
	   }
	   public void run() {
	      try {
	        doWork(i);
	        doneSignal.countDown();
	      } catch (InterruptedException ex) {} // return;
	   }

	 public  void doWork(int i) {		 
	 }
	 
}


public class Driver1 {
public void main() throws InterruptedException{
	CountDownLatch doneSignal=new CountDownLatch(1);
	Executor c=Executors.newCachedThreadPool();
	for(int i=0;i<4;i++){//create and start threads
		c.execute(new Worker1(doneSignal,i));
	}
	doneSignal.await();//wait all finish
}
}



 

FutureTask
 

 

为了理解FutureTask，让我们先来看看Future
 

Future 表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。计算完成后只能使用get 方法来获取结果，如有必要，计算完成前可以阻塞此方法。取消则由cancel 方法来执行。还提供了其他方法，以确定任务是正常完成还是被取消了。一旦计算完成，就不能再取消计算。
 

 

interface ArchiveSearcher { String search(String target); }
 class App {
   ExecutorService executor = ..
   ArchiveSearcher searcher = ...
   void showSearch(final String target)
       throws InterruptedException {
     Future<String> future
       = executor.submit(new Callable<String>() {
         public String call() {
             return searcher.search(target);
         }});
     displayOtherThings(); // do other things while searching
     try {
       displayText(future.get()); // use future
     } catch (ExecutionException ex) { cleanup(); return; }
   }
 }
 




 
FutureTask 类是 
Future 的一个实现， 
Future 可实现 
Runnable，所以可通过 
Executor 来执行。例如，可用下列内容替换上面带有 
submit 的构造： 

 

     FutureTask<String> future =
       new FutureTask<String>(new Callable<String>() {
         public String call() {
           return searcher.search(target);
       }});
     executor.execute(future);
 

 


信号量


 

Semaphore
 

一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个  acquire()，然后再获取该许可。每个  release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。
 
 

计算信号量的一种简化形式是二值信号量，即初始值为１的Semaphore，二值信号量可以用做互斥体，并具备不可重入的加锁语义。
 

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。例如，下面的类使用信号量控制对内容池的访问：



 

 class Pool {
   private static final int MAX_AVAILABLE = 100;
   private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);

   public Object getItem() throws InterruptedException {
     available.acquire();
     return getNextAvailableItem();
   }

   public void putItem(Object x) {
     if (markAsUnused(x))
       available.release();
   }

   // Not a particularly efficient data structure; just for demo

   protected Object[] items = ... whatever kinds of items being managed
   protected boolean[] used = new boolean[MAX_AVAILABLE];

   protected synchronized Object getNextAvailableItem() {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (!used[i]) {
          used[i] = true;
          return items[i];
       }
     }
     return null; // not reached
   }

   protected synchronized boolean markAsUnused(Object item) {
     for (int i = 0; i < MAX_AVAILABLE; ++i) {
       if (item == items[i]) {
          if (used[i]) {
            used[i] = false;
            return true;
          } else
            return false;
       }
     }
     return false;
   }

 }
 





栅栏（屏障）


 

CyclicBarrier
 

CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障(Barrier)。它要做的事情是，让一组线程到达一个屏障（也可以叫做同步点）时阻塞，
 

直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才会继续执行。同时，Barrier在释放等待线程之后可以重用。
 

CyclicBarrier支持一个可选的Runnable命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。
 

栅栏与闭锁的关键区别：
 

１．闭锁是一次性对象，栅栏可以重用，也可以reset()重置。
 

２．所有线程必须都到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。
 

 

示例用法：用户多线程计算数据,最后合并计算结果的场景。



　一个Excel保存了用户所有的银行流水，每个Sheet保存了一个账户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完了之后,得到每个sheet的日均银行流水,最后，再用barrierAction用这些线程的计算结果，计算整个Excel的日均银行流水。
 

代码如下：
 

 

<span style="color:#333333;">pa<span style="color:#330033;">ckage cn.dachao.thread;

import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by dachao on 16-7-28.
 */
public class BankWaterService implements Runnable{
   /*
   创建４个屏障,处理完之后执行当前类的run方法
    */
    private CyclicBarrier c=new CyclicBarrier(4,this);
    /*
    假设只有4个sheet，所以只启动４个线程
     */
    private Executor executor= Executors.newFixedThreadPool(4);
    /*
    保存每个sheet计算出的银流结果
     */
    private ConcurrentHashMap<String,Integer> count=new ConcurrentHashMap<>();
    private void count(){
        for(int i=0;i<4;i++){
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    //计算当前sheet的银行数据，代码省略
                    count.put(Thread.currentThread().getName(),1);
                    try {
                        c.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            });

        }
    }
    @Override
    public void run(){
        int result=0;
        //汇总每个sheet计算出的结果
        for(Map.Entry<String,Integer> sheet:count.entrySet()){
            result+=sheet.getValue();
        }
        //将结果输出
        count.put("result",result);
        System.out.println(result);
    }
    public static void main(String[]args){
        BankWaterService service=new BankWaterService();
        service.count();
    }
｝
输出结果：
<span style="color:#333333;">4</span>
</span></span>



Exchanger
 

 

可以在对中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给  exchange 方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。Exchanger 可能被视为  SynchronousQueue 的双向形式。Exchanger 可能在应用程序（比如遗传算法和管道设计）中很有用。  
 
 

用法示例：以下是重点介绍的一个类，该类使用 Exchanger 在线程间交换缓冲区，因此，在需要时，填充缓冲区的线程获取一个新腾空的缓冲区，并将填满的缓冲区传递给腾空缓冲区的线程。
 

 

class FillAndEmpty {
   Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
   DataBuffer initialEmptyBuffer = ... a made-up type
   DataBuffer initialFullBuffer = ...

   class FillingLoop implements Runnable {
     public void run() {
       DataBuffer currentBuffer = initialEmptyBuffer;
       try {
         while (currentBuffer != null) {
           addToBuffer(currentBuffer);
           if (currentBuffer.isFull())
             currentBuffer = exchanger.exchange(currentBuffer);
         }
       } catch (InterruptedException ex) { ... handle ... }
     }
   }

   class EmptyingLoop implements Runnable {
     public void run() {
       DataBuffer currentBuffer = initialFullBuffer;
       try {
         while (currentBuffer != null) {
           takeFromBuffer(currentBuffer);
           if (currentBuffer.isEmpty())
             currentBuffer = exchanger.exchange(currentBuffer);
         }
       } catch (InterruptedException ex) { ... handle ...}
     }
   }

   void start() {
     new Thread(new FillingLoop()).start();
     new Thread(new EmptyingLoop()).start();
   }
  }



标签：
相关文章