JUC包 (二) CountDownLatch、CyclicBarrier、Semaphore 工具类

痛定思痛。 2022-03-02 07:56 258阅读 0赞

### 前言 ###

在[Java 多线程(三) 线程通信][Java _]内我们介绍了可以使用`join()`方法来控制某个线程在一众线程后执行. 正这一节中,我们将介绍三种工具类,同样实现这一目标. 三种工具使用场景各部相同.

本章主要分为如下几个部分:

*  CountDownLatch类
 *  CyclicBarrier类
 *  Semaphore类

--------------------

### 正文 ###

##### join()方法实现 #####

在前文中,我们提及.使用`join()`方法可以非常容易的实现这部分的需求.

*  await() / await(long time, TimeUnit unit) : 指定时间的await()方法;
 *  countDown() : 计数器减少1;

实例代码如下所示:

/**
     * 使用Join方法完成 运行在几个线程之后的这种情况.
     * 
     * */
    class JoinThread extends Thread{
    	public void run(){
    		try {
    			Thread.sleep(2000);
    			System.out.println(Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    	}
    }
    public class JoinDemo {
    	public static void main(String[] args) {
    		Thread threadA = new JoinThread();
    		Thread threadB = new JoinThread();
    		Thread threadC = new JoinThread();
    		Thread threadD = new JoinThread();
    
    		threadA.start();
    		threadB.start();
    		threadC.start();
    		threadD.start();
    		
    		try {
    			threadA.join();
    			threadB.join();
    			threadC.join();
    			threadD.join();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    
    		
    		System.out.println("Main Thread"+Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    
    	}
    
    }
    //Thread-0 : 1553018593261
    //Thread-2 : 1553018593261
    //Thread-1 : 1553018593262
    //Thread-3 : 1553018593261
    //Main Threadmain : 1553018593263

##### CountDownLatch类 #####

我们同样可以使用`CountDownLatch类`来完成我们的需求.其中最主要的两个方法为:

*  `countDown()`减去1;
 *  `await()`进行等待;

import java.util.concurrent.CountDownLatch;
    
    /**
     * CountDownLatch来完成需求.
     * 
     * */
    class CountDownLatchThread extends Thread{
    	public void run(){
    		try {
    				Thread.sleep(2000);
    				System.out.println(Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    				CountDownLatchDemo.countDownLatch.countDown();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    	}
    }
    public class CountDownLatchDemo {
    	public static CountDownLatch countDownLatch = new CountDownLatch(4);
    	public static void main(String[] args) {
    		Thread threadA = new CountDownLatchThread();
    		Thread threadB = new CountDownLatchThread();
    		Thread threadC = new CountDownLatchThread();
    		Thread threadD = new CountDownLatchThread();
    
    		threadA.start();
    		threadB.start();
    		threadC.start();
    		threadD.start();
    		
    		try {
    			countDownLatch.await();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    		System.out.println("Main Thread"+Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    
    	}
    
    }
    //Thread-2 : 1553019240096
    //Thread-3 : 1553019240096
    //Thread-0 : 1553019240096
    //Thread-1 : 1553019240096
    //Main Threadmain : 1553019240097

##### CyclicBarrier类 #####

使用`CyclicBarrier`类也可以完成如上的要求.其主要方法为:

*  await(): 设置屏障点;
 *  reset(): 可以重新计数;
 *  getNumberWaiting(): 获取阻塞的线程数量;
 *  isBroken(): 阻塞的线程是否被中断;

import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * 使用CycleBarrier来实现需求.
     * 
     * */
    class CycleBarrierThread extends Thread{
    	public void run(){
    		try {
    				Thread.sleep(2000);
    				System.out.println(Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    				try {
    					CycleBarrierDemo.cycleBarrier.await();
    				} catch (BrokenBarrierException e) {
    					e.printStackTrace();
    				}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    	}
    }
    public class CycleBarrierDemo {
    	public static CyclicBarrier cycleBarrier = new CyclicBarrier(5);
    	public static void main(String[] args) {
    		Thread threadA = new CycleBarrierThread();
    		Thread threadB = new CycleBarrierThread();
    		Thread threadC = new CycleBarrierThread();
    		Thread threadD = new CycleBarrierThread();
    
    		threadA.start();
    		threadB.start();
    		threadC.start();
    		threadD.start();
    		
    		try {
    			cycleBarrier.await();
    		} catch (BrokenBarrierException e) {
    			e.printStackTrace();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    		System.out.println("Main Thread"+Thread.currentThread().getName()+" : "+System.currentTimeMillis());
    	}
    }
    // Thread-0 : 1553019850995
    // Thread-1 : 1553019850995
    // Thread-3 : 1553019850995
    // Thread-2 : 1553019850995
    // Main Threadmain : 1553019850996

`CyclicBarrier`有时还可以用于统计某几个线程运算的结果.(与`fork/join`有些类似?)

import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Executors;
    
    /**
     * 8-2-2 CyclicBarrier的实际使用场景.
     * 银行的流水系统.(CycleBarrier的实践)
     * 
     * */
    public class BankWaterService implements Runnable{
    	//创建4个屏障 完成后自动执行当前类的(this)的run方法.
    	private CyclicBarrier cycleBarrier = new CyclicBarrier(4,this);
    	// 线程池 固定线程池 设置4个线程.
    	private Executor executor = Executors.newFixedThreadPool(4);
    	
    	// 保存每个Sheet计算出的银行流量结果
    	private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();
    	private void count() {
    		for (int i = 0; i < 4; i++) {
    			executor.execute(new Runnable() {
    
    				@Override
    				public void run() {
    					// 计算当前sheet的银行流量数据.(具体逻辑代码省略)
    					sheetBankWaterCount.put(Thread.currentThread().getName(),1);
    					try {
    						cycleBarrier.await();
    					} catch (Exception e) {
    						e.printStackTrace();
    					}
    				}
    			});
    		}
    	}
    	
    	public void run(){
    		int result = 0;
    		// 汇总每个sheet的计算成果.
    		for(Map.Entry<String,Integer> sheet: sheetBankWaterCount.entrySet()){
    			result += sheet.getValue();
    		}
    		// 将结果输出
    		sheetBankWaterCount.put("result", result);
    		System.out.println(result);
    	}
    	
    	public static void main(String[] args) {
    		BankWaterService service = new BankWaterService();
    		service.count();
    	}
    
    }
    // 4

其中值得注意的是我们在构造函数`CyclicBarrier(int parties, Runnable barrierAction)`使得当阻塞停止时,就运行指定的业务.这点与`CountDownLatch类`略有写不太一致.

> CyclicBarrier类与CountDownLatch类有什么区别?  
> 在提及区别之前,我们先说下共同点: 都可以使某个线程阻塞,并在某个(某几个)线程结束后再进行执行.  
> 区别: 个人感觉`CyclicBarrier类`比`CountDownLatch类`封装的更完善.注意部分如下:1 .`CyclicBarrier类`能够将计数值重置; 2. `isBroken()`方法能够了解阻塞的线程是否被中断.3. 调用接口的写法不太一致, `CountDownLatch(int value)`而`CyclicBarrier(int value, Runnable runnable)`.可以直接指定完成后执行某个线程.

##### Semaphore类 #####

`Semaphore类`经常是用来控制流量. 比如有4个线程,但是当前每次运行的线程数目为2个,便可以使用`Semaphore类`来进行控制.我们经常通过`tryAcquire()`或`acquire()`方法获取权限,使用`release()`释放资源.

*  acquire() 获取资源
 *  tryAcquire() 尝试获取资源
 *  release()释放资源
 *  int availablePermits() 返回此信号量

import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    
    /**
     * 8-3 Semaphore控制线程并发数目.
     * 
     * */
    public class SemaphoreDemo {
    	private static final int THREAD_COUNT=30;
    	
    	private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
    	
    	private static Semaphore s = new Semaphore(10);
    	
    	public static void main(String[] args) {
    		for(int i=0;i<THREAD_COUNT;i++){
    			threadPool.execute(new Runnable() {
    				
    				@Override
    				public void run() {
    					try {
    						s.acquire();
    						System.out.println(Thread.currentThread().getName()+"save data");
    						s.release();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    					}					
    				}
    			});
    		}
    		threadPool.shutdown();
    	}
    }

##### Exchanger 类 #####

`Exchanger类`主要用于线程之间的数据交换与数据校对、遗传算法.

*  exchang(Object object): 数据跟随管道进行传递.

示例如下所示:

import java.util.concurrent.Exchanger;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * 8-4 线程间交换数据的Exchanger.
     * (银行流水的相互校验)
     * */
    public class ExchangerDemo {
    	private static final Exchanger<String> exchanger = new Exchanger<>();
    	private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    	
    	public static void main(String[] args) {
    		// 载入第一个线程
    		threadPool.execute(new Runnable() {
    			
    			@Override
    			public void run() {
    				String A = "银行流水A";
    				try {
    					exchanger.exchange(A);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		
    		// 载入第二个线程
    		
    		threadPool.execute(new Runnable() {
    			
    			@Override
    			public void run() {
    				String B = "银行流水B";
    				try {
    					String A = exchanger.exchange(B);
    					System.out.println("A和B数据是否一致:"+A.equals(B)+". A录入的数据为:"+A+". B录入的数据为:"+B);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		
    		threadPool.shutdown();
    	}
    
    }
    
    // A和B数据是否一致:false. A录入的数据为:银行流水A. B录入的数据为:银行流水B

--------------------

### Reference ###

\[1\]. Java 多线程编程核心技术  
\[2\]. Java并发编程的艺术  
\[3\]. [什么时候使用CountDownLatch][CountDownLatch]

[Java _]: https://yanxml.blog.csdn.net/article/details/88659177
[CountDownLatch]: http://www.importnew.com/15731.html