Java并发编程——CountDownLatch和CyclicBarrier

野性酷女 2022-03-01 15:00 324阅读 0赞

## 一、引言 ##

在Java 1.5中，提供了一些非常有用的辅助类来帮助我们进行并发编程，比如CountDownLatch，CyclicBarrier、Semaphore。下面就来学习一下这三个辅助类的用法，若有不正之处请多多谅解。

## 二、CountDownLatch ##

###### 1、CountDownLatch 是什么呢？ ######

CountDownLatch 在Java1.5中被引入的，跟它一起被引入的并发工具类还有CyclicBarrier、Semaphore、ConcurrentHashMap和BlockingQueue。它们都存在于java.util.concurrent包下。CountDownLatch 这个类能够使一个线程等待其他线程完成各自的工作后再执行。 利用它可以实现类似计数器的功能，比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能了。

###### 2、CountDownLatch 原理 ######

CountDownLatch 是通过一个计数器来实现的，计数器的初始值为线程的数量。这个值只能被设置一次，而且CountDownLatch没有提供任何机制去重新设置这个计数值。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在调用await()方法的等待线程就可以恢复执行任务。下面是CountDownLatch工作的伪代码：

//Main thread start
    //Create CountDownLatch for N threads
    //Create and start N threads
    //Main thread wait on latch
    //N threads completes there tasks are returns
    //Main thread resume execution

###### 3、CountDownLatch 用法 ######

CountDownLatch 类只提供了一个构造器：

public CountDownLatch(int count) {  };  //参数count为计数值，实际上就是闭锁需要等待的线程数量。

下面这3个方法是CountDownLatch类中最重要的方法：

//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行，除非线程被中断
    public void await() throws InterruptedException { };   
    
    //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
    public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };  
    
    //调用此方法则将count值减1
    public void countDown() { }; 
    
    //得到当前的计数
    public Long getCount();

###### 4、CountDownLatch 实例 ######

public class CountDownLatchExample { 
    
    	public static void main(String[] args) { 
    		final CountDownLatch latch = new CountDownLatch(2);
    
    		new Thread("T1") { 
    			public void run() { 
    				try { 
    					System.out.println("子线程" + Thread.currentThread().getName() + "正在执行");
    					Thread.sleep(1000);
    					System.out.println("子线程" + Thread.currentThread().getName() + "执行完毕");
    					latch.countDown();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    			};
    		}.start();
    
    		new Thread("T2") { 
    			public void run() { 
    				try { 
    					System.out.println("子线程" + Thread.currentThread().getName() + "正在执行");
    					Thread.sleep(3000);
    					System.out.println("子线程" + Thread.currentThread().getName() + "执行完毕");
    					latch.countDown();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    			};
    		}.start();
    
    		try { 
    			System.out.println("等待2个子线程执行完毕...");
    			latch.await();
    			System.out.println("2个子线程已经执行完毕");
    			System.out.println("继续执行主线程");
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    	}
    }

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70]  
说明：线程T1休眠了1秒，线程T2休眠了3秒，但T1执行完后，还要等T2线程执行完之后，主线程才能继续往下执行。

###### 5、小节 ######

与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。

其他 N 个线程必须引用闭锁对象，因为他们需要通知 CountDownLatch 对象，他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的count值就减1。所以当N个线程都调 用了这个方法，count的值等于0，然后主线程就能通过await()方法，恢复执行自己的任务。

## 三、CyclicBarrier ##

英文单词的字面意思：回环栅栏。通过它可以实现让一组线程等待至某个状态之后再全部同时执行。之所以叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier，当调用await()方法之后，线程就处于barrier了。

###### 1、CyclicBarrier 介绍 ######

CyclicBarrier类位于java.util.concurrent包下，CyclicBarrier提供2个构造器：

//参数parties指让多少个线程或者任务等待至barrier状态，参数barrierAction为当这些线程都达到barrier状态时会执行的内容，
    //先执行barrierAction，然后在释放所有线程
    public CyclicBarrier(int parties, Runnable barrierAction) { 
    }
     
    public CyclicBarrier(int parties) { 
    }

###### 2、方法介绍 ######

CyclicBarrier中最重要的方法就是 await 方法，它有2个重载版本：

public int await() throws InterruptedException, BrokenBarrierException {  };
    	
    	public int await(long timeout, TimeUnit unit) throws InterruptedException,
    											BrokenBarrierException,TimeoutException {  };
    											
    	public void reset()

第一个版本比较常用，用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；

第二个版本是让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。  
下面举几个例子就明白了；

reset 方法将屏障重置为其初始状态。如果所有参与者目前都在屏障处等待，则它们将返回，同时抛出一个BrokenBarrierException。

###### 3、CyclicBarrier实例 ######

假若有若干个线程都要进行写数据操作，并且只有所有线程都完成写数据操作之后，这些线程才能继续做后面的事情。

###### （1）、一个参数的构造函数 ######

public class CyclicBarrierExample { 
    
    	public static void main(String[] args) { 
    
    		final CyclicBarrier barrier = new CyclicBarrier(3);
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(2000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(4000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(6000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    	}
    }

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 1]  
结果说明：第一个线程休息2秒，第二个线程休息4秒，第三个线程休息6秒，但每个线程中的最后一句System.out.println，都会等到第三个线程休息完，才会同时执行。

###### （2）、二个参数的构造函数 ######

从上面输出结果可以看出，每个写入线程执行完写数据操作之后，就在等待其他线程写入操作完毕。当所有线程写入操作完毕之后，所有线程就继续进行后续的操作了，如果说想在所有线程写入操作完之后，进行额外的其他操作可以为 CyclicBarrier 提供Runnable参数。

public class CyclicBarrierExample { 
    
    	public static void main(String[] args) { 
    
    		final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("所有线程执行完毕，回调此处");
    				System.out.println("当前线程：" + Thread.currentThread().getName() + "");
    			}
    		});
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(2000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(4000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(6000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    	}
    }

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 2]  
结果说明：从结果可以看出，当三个线程都到达barrier状态后，会从三个线程中选择一个线程去执行Runnable。

###### （3）、为await指定时间 ######

public class CyclicBarrierExample { 
    
    	public static void main(String[] args) { 
    
    		final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("所有线程执行完毕，回调此处");
    				System.out.println("当前线程：" + Thread.currentThread().getName() + "");
    			}
    		});
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(2000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					try { 
    						barrier.await(2000, TimeUnit.MILLISECONDS);
    					} catch (TimeoutException e) { 
    						// TODO Auto-generated catch block
    						e.printStackTrace();
    					}
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(4000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    
    		new Thread(new Runnable() { 
    			@Override
    			public void run() { 
    				System.out.println("线程" + Thread.currentThread().getName() + "正在写入数据...");
    				try { 
    					Thread.sleep(6000); // 以睡眠来模拟写入数据操作
    					System.out.println("线程" + Thread.currentThread().getName() 
    							+ "写入数据完毕，等待其他线程写入完毕");
    					barrier.await();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				} catch (BrokenBarrierException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("所有线程写入完毕，继续处理其他任务...");
    			}
    		}).start();
    	}
    }

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 3]  
结果说明：第一个线程await()方法，只能等待2秒，该线程的时间到了，就会抛出一个TimeoutException异常，并继续执行后面的任务。

###### （4）、CyclicBarrier可重用 ######

public class CyclicBarrierExample2 { 
    
    	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(25000);
    		} 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(5000); // 以睡眠来模拟写入数据操作
    				System.out.println("线程" + Thread.currentThread().getName() 
    						+ "写入数据完毕，等待其他线程写入完毕");
    
    				cyclicBarrier.await();
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			} catch (BrokenBarrierException e) { 
    				e.printStackTrace();
    			}
    			System.out.println(Thread.currentThread().getName() 
    					+ "所有线程写入完毕，继续处理其他任务...");
    		}
    	}
    }

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 4]  
从执行结果可以看出，在初次的4个线程越过barrier状态后，又可以用来进行新一轮的使用。

## 四、总结 ##

CyclicBarrier与CountDownLatch比较：

（1）、CountDownLatch：一个线程(或者多个)，等待另外N个线程完成某个事情之后才能执行；CyclicBarrier：N个线程相互等待，任何一个线程完成之前，所有的线程都必须等待。

（2）、CountDownLatch：一次性的；CyclicBarrier：可以重复使用。

（3）、CountDownLatch基于AQS；CyclicBarrier基于锁和Condition。本质上都是依赖于volatile和CAS实现的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70]: /images/20220301/38dec41646dd415680acd2f51c437cdd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 1]: /images/20220301/09efc8ada9724ff0846af299df39ea96.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 2]: /images/20220301/952c3c9ae347475e94ee491e040d171b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 3]: /images/20220301/e8ca9cb25c2c40af8e0a213f6913d460.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4ZDE0MzU1MTM3NzU_size_16_color_FFFFFF_t_70 4]: /images/20220301/140e869e7c864918883efc83606f4841.png