获得更好并发和线程安全性的场景和解决方案--第二部分 CountDownLatch(闭锁) and CyclicBarrier(循环关卡)

现实场景下我们可以借助java.util.concurrent包下的以下几个类做到更好。

    1.CountDownLatch(闭锁)

    2.CyclicBarrier(关卡)

以下有一些具体的现实场景可以使以下的类派上用场。

 

场景：主线程创建三个数据库连接，并且在主线程中为每个连接分配三个子线程。主线程必须等待子线程执行完成并且所有的数据库连接关闭后才能退出，所以，我们应该怎么去实现呢？

 

方案：因为我们已经知道了线程的具体数量，所以我们可以使用CountDownLatch(闭锁)。CountDownLatch可以在主线程中使用去实现等待子线程的功能。当前CountDownLatch可以使用数字3来初始化创建。

 

CountDownLatch countDownLatch = new CountDownLatch(MAX_THREADS);

 主线程生成若干子线程并且通过await方法等待count变成0的时候才开始执行

countDownLatch.await();

 

在每个子线程run()方法执行的过程时，只要子线程完成处理操作，count数就会递减

 

countDownLatch.countDown(); 

 

 

 

图上所示的是当若干工作线程使用startSignal(闭锁)等待其他工作线程来启动并且主线程通过stopSignal(闭锁)方法等待所有的工作线程

 

import java.util.concurrent.CountDownLatch;
 
public class Worker implements Runnable {
  
    private CountDownLatch startLatch;
    private CountDownLatch stopLatch;
 
    public Worker(CountDownLatch startLatch, CountDownLatch stopLatch) {
       this.startLatch = startLatch;
       this.stopLatch = stopLatch;
    }
 
    @Override
    public void run() {
        try {
            startLatch.await(); // 等待闭锁递减到0
            System.out.println("Running: " + Thread.currentThread().getName());
        } catch (InterruptedException ex)  {
            ex.printStackTrace();
        }
        finally {
         //闭锁递减到0，主线程继续执行
         stopLatch.countDown(); 
        }
 }
 
}

 

最后，定义创建工作线程的WaitForAllThreadsToStart类

 

import java.util.concurrent.CountDownLatch;
 
public class WaitForAllThreadsToStart {
  
 private static final int MAX_THREADS = 3;
  
 public static void main(String[] args) throws Exception {
     CountDownLatch startSignal = new CountDownLatch(1);   //闭锁递减从1到0
     CountDownLatch stopSignal = new CountDownLatch(MAX_THREADS); // 闭锁递减从3到0
         
     System.out.println("The main thread is going to spawn " + MAX_THREADS  + " worker threads.....");  
     for (int i = 1; i <= MAX_THREADS; i++) {
        Thread t = new Thread(new Worker(startSignal,stopSignal), "thread-" + i);
        Thread.sleep(300);
        t.start();
        System.out.println("Started: " + t.getName() + " but waits for other threads to start.");
     }
         
     //等待线程在闭锁从1递减到0后开始继续执行 
     startSignal.countDown();
     System.out.println("worker threads can now start executing as all worker threads have started.....");
     try{       
        stopSignal.await(); // 等待工作线程操作使闭锁递减到0
     } catch (InterruptedException ex){
        ex.printStackTrace();
     }
     System.out.println("finished executing the worker threads and the main thread is continuing.");
     System.out.println("The main thread can execute any task here.");
         
 }
}

 

输出如下：

 

The main thread is going to spawn 3 worker threads.....
Started: thread-1 but waiting for other threads to start.
Started: thread-2 but waiting for other threads to start.
Started: thread-3 but waiting for other threads to start.
worker threads can now start executing as all worker threads have started.....
Running: thread-1
Running: thread-3
Running: thread-2
finished executing the worker threads and the main thread is continuing.
The main thread can execute any task here.

 

场景：如果在以上的场景下增加一个特殊的需求让三个子线程互相等待？举个例子，如果三个线程中的每个都需要完成两个任务。在线程开始执行第二个任务之前，所有的三个线程都必须完成第一个任务。第一个任务是从数据库读取数据并且第二个任务对数据进行计算，最后所有的计算结果需要被合并由一个线程写回数据库.

 

方案：关卡可以在当前操作前有若干子进程并且需要实现等待所有子进程处理完成的场景下使用.这在需要多个并行进程完成一个串行处理的情况下很有用处.可以使用的方法如cyclicBarrier.await()和cyclicBarrier.reset()方法

 

 

以下的代码可以帮助你理解.首先是WorkerTask线程

 

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
 
public class WorkerTask implements Runnable {
  
 private CyclicBarrier barrier;
   
 public WorkerTask(CyclicBarrier barrier) {
  this.barrier = barrier;
 }
 
 @Override
 public void run() {
   String threadName = Thread.currentThread().getName();
   try {
         System.out.println(threadName + " is now performing Task-A");
         barrier.await();      //等待所有线程完成任务A的关卡
            
         System.out.println(threadName + " is now performing Task-B");
         barrier.await();     //等待所有关卡完成任务B的关卡
            
   } catch (BrokenBarrierException ex)  {
         ex.printStackTrace();
   } catch (InterruptedException e) {
      e.printStackTrace();
   }
  }
}

 

 

现在，测试类创建了执行任务A和任务B的工作线程，并且关卡线程完成合并处理

 

import java.util.concurrent.CyclicBarrier;
 
public class WaitForBarrierPoint {
  
 private static final int MAX_THREADS = 3;
 private static final int NO_OF_TASKS = 2;    //任务A和B
  
 private static int taskCount = 0;
  
 //创建一个监视关卡条件的线程，也就是等MAX_THREADS完成任务进行合并操作
//由匿名内部类来执行
 private static CyclicBarrier cb = new CyclicBarrier(MAX_THREADS, new Runnable() {
  
   @Override
   public void run() {
     System.out.println("All " + MAX_THREADS + " threads have reached the barrier point.");
     ++taskCount;
    
     //在所有任务完成后执行合并处理
     if(taskCount == NO_OF_TASKS) {
      System.out.println("The consolidation job can start now .... ");
   }
  }
 }); 
  
  
 public static void main(String[] args) {
  Thread t = null;
  //create 3 worker threads
  for (int i = 1; i <= MAX_THREADS; i++) {
   t = new Thread(new WorkerTask(cb), "Thread-" + i);
   t.start();
  }
   
  System.out.println("The main thread ends here.");
 }
  
}

 

 

Q.为什么称它为cyclic barrier(循环关卡)?

A.因为它的作用就是让若干线程等待彼此完成任务的关卡点.关卡被称为循环的原因是因为所有等待的工作线程被放行后在下个关卡点之前还可以被重用.

 

关卡点可以通过传递下面的参数给构造器来创建.

  1.参与并行操作的线程数目

  2.满足通过关卡条件的操作完成之后需要调用的处理

 

每步操作(或者迭代)时：

  每个线程完成属于自己工作的一部分来完成一步操作

  在完成了自己的工作部分后，线程会调用关卡的await方法

  await方法只有下列返回情况：

  1.三个线程都应调用过await()

  2.融合或者混合方法已经被执行(在对等待线程放行前，barrier已经在最后一个线程调用await()时调用过)

 

如果三个线程中的任何一个在等待barrier的时候被中断或者超时，这时关卡点已经被破坏掉，所以其他等待的线程会收到一个BrokenBarrierException的异常.这个异常会被传递到所有的线程并被其他步骤终止，或者只是被其中一个线程从外部中断掉.

 

Q.因此，什么时候使用CountDownLatch和什么时候可以使用CyclicBarrier呢？

A.CountDownLatch是通过一个初始数初始化的.线程可以等待闭锁的递减或者等待直到0.当最后到达0的时候，所有等待的线程都会被恢复.

 

如果你希望在一个点重复的执行一系列线程,你最好使用CyclicBarrier(关卡).举例来说，启动一批线程，在一个点会合，为了满足，比如一些如混合或者合并的工作，然后再重新会合，做一些验证，然后重复这些操作.

 

CountDownLatch只能被使用一次，在多步骤的操作中显得不太方便，如需要在多个阶段将不同的线程的中间结果进行合并.闭锁不应该显式的允许一个线程可以有通知其他线程停止等待的权利，虽然有时是有用的，比如如果线程之一的发生错误.

 

关卡在以下的场景下会比CountDownLatch(闭锁)更有用：

   1.多个阶段或者迭代的多线程操作

   2.多个阶段或者迭代的单线程操作中，例如，为了合并前面多线程阶段的结果而进行的处理