分享牛原创,CountDownLatch类的使用,CountDownLatch是一个工具类,运行主线程开启子线程的时候,子线程还没有结束的时候,主线程可以一直等待,直到初始化的现成的计数器count为0,主线程就可以不用等待继续执行了。count只能在实例化CountDownLatch类的时候初始化,没有其他的办法count注入值。
下面看一下java.util.concurrent.CountDownLatch如何使用。
1.1.1. CountDownLatch类中的方法
CountDownLatch类中的主要访法如下图所示:
构造方法如下所示:
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
在这里要特别注意,实例化CountDownLatch类的时候,参数值一定要与启动的线程个数一直。假如启动三个线程,我们这里传入的值是4,那么每个线程结束的时候调用countDown方法的时候,计数器会减去1,这样三个完成之后,计数器时1,则程序就是一直等待状态。永远不会向下执行。
countDown方法,当前线程调用此方法,则计数在实例化CountDownLatch中传入的值基础上减一。
awaint方法,调用此方法会一直阻塞当前线程,不会向下执行,直到计时器的值为0的时候程序才会继续向下执行。
1.1.2. 测试
下面的程序开始测试CountDownLatch的使用。
package com.shareniu.zkTest; import java.util.concurrent.CountDownLatch; public class ShareniuDriver { private static final int N = 2; public static void main(String[] args) throws InterruptedException { CountDownLatch doneSignal = new CountDownLatch(N); for (int i = 0; i < N; ++i) // create and start threads { new Thread(new Worker(doneSignal,i)).start(); } doneSignal.await(); doSomethingElse(); // don't let run yet } private static void doSomethingElse() { System.out.println("分享牛 我在做其他事情....."); } static class Worker implements Runnable { private CountDownLatch doneSignal; private int i; public Worker( CountDownLatch doneSignal,int i) { this.doneSignal = doneSignal; this.i=i; } public void run() { try { doWork(); } catch (Exception e) { } finally { doneSignal.countDown(); } } private void doWork() { System.out.println("开始工作....."+i); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
上面的程序可以得出一个结论:count为0的时候,await()之后的方法才会去执行,否则一直处于等待状态。
上面的是启动2个线程,当我们把N设置为3的时候,也就是大于2的任意数的时候,启动2个线程,其他地方不变化,发现程序一直等待,不执行。这个也间接说明了,只有count为0的时候,程序才会继续走下去。
1.1.3. 工作原理
工作原理如下图所示:
1.1.4. 使用场景
程序启动的时候,根据业务开启几个线程去执行检查服务是否正常,主线程一直等待,当检查的线程都结束的时候,主线程才能去判断对应的所有的线程返回结果,检查是否正常。
比如有一个任务A,它要等待其他4个任务执行完毕之后才能执行,此时就可以利用CountDownLatch来实现这种功能了。
具体的程序如下:
1.1.4.1. BaseHealthChecker 类,所有类的基类
package com.shareniu.zkTest; import java.util.concurrent.CountDownLatch; public abstract class BaseHealthChecker implements Runnable { private CountDownLatch latch; private String serviceName; private boolean serviceUp; public BaseHealthChecker(CountDownLatch latch, String serviceName) { this.latch = latch; this.serviceName = serviceName; } public void run() { try { verifyService(); serviceUp = true; } catch (Exception e) { e.printStackTrace(System.err); serviceUp = false; }finally{ if(latch != null) { latch.countDown(); } } } public abstract void verifyService(); public CountDownLatch getLatch() { return latch; } public String getServiceName() { return serviceName; } public boolean isServiceUp() { return serviceUp; } }
1.1.4.2. NetworkHealthChecker 类
package com.shareniu.zkTest; import java.util.concurrent.CountDownLatch; public class NetworkHealthChecker extends BaseHealthChecker { public NetworkHealthChecker(CountDownLatch latch, String serviceName) { super(latch, serviceName); } @Override public void verifyService() { System.out.println("开始检查 " + this.getServiceName()); try { Thread.sleep(7000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(this.getServiceName() + " 检查结束"); } }
1.1.4.3. CacheHealthChecker 类
package com.shareniu.zkTest; import java.util.concurrent.CountDownLatch; public class CacheHealthChecker extends BaseHealthChecker { public CacheHealthChecker(CountDownLatch latch, String serviceName) { super(latch, serviceName); } @Override public void verifyService() { System.out.println("开始检查 CacheHealthChecker" + this.getServiceName()); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(this.getServiceName() + "CacheHealthChecker 检查结束 "); } }
1.1.4.4. 测试类
package com.shareniu.zkTest; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class AppTest { private static List<BaseHealthChecker> services; private static CountDownLatch latch; public static void main(String[] args) throws InterruptedException { latch = new CountDownLatch(2); services=new ArrayList<BaseHealthChecker>(); services.add(new CacheHealthChecker(latch, "CacheHealthChecker")); services.add(new NetworkHealthChecker(latch, "NetworkHealthChecker")); ExecutorService executor = Executors.newFixedThreadPool(services.size()); for(final BaseHealthChecker v : services) { executor.execute(v); } //等待 指导所有的线程执行完毕 latch.await(); for(final BaseHealthChecker v : services) { boolean serviceUp = v.isServiceUp(); System.out.println(v.getServiceName()+":"+serviceUp); } } }
运行结果如下所示:
开始检查 CacheHealthCheckerCacheHealthChecker
开始检查 NetworkHealthChecker
CacheHealthCheckerCacheHealthChecker 检查结束
NetworkHealthChecker 检查结束
CacheHealthChecker:true
NetworkHealthChecker:true