java并发编程学习10--同步器--倒计时门栓

【同步器

java.util.concurrent包包含几个能帮助人们管理相互合做的线程集的类。这些机制具备为线程直间的共用集结点模式提供的‘预制功能’。若是有一个相互合做的线程知足这些行为模式之一，那么应该直接使用提供的类库而不是显示的使用锁与条件的集合。

【倒计时门栓

一个倒计时门栓（CountDownlatch）让一个线程集直到计数变为0.倒计时门栓是一次性的，一旦计数为0就不能再重用了。一个有用的特例是计数值为1的门栓。实现一个只能经过一次的门。线程在门外等待直到另外一个线程将计数值变为0。举例来说，假设一个线程集须要一些初始数据来完成工做。工做线程被启动并在，门外等候，另外一个线程准备数据，当数据准备好时，调用countDown()，全部的工做线程就能够继续工做了。而后再使用一个门栓检查何时工做线程所有运行完成。每一个工做线程在结束前将门栓计数器减一，门栓的计数变为0就代表工做完成。

【经常使用方法

• public void countDown():递减锁存器的计数，若是计数到达零，则释放全部等待的线程。若是当前计数大于零，则将计数减小。若是新的计数为零，出于线程调度目的，将从新启用全部的等待线程。若是当前计数等于零，则不发生任何操做。
• public boolean await():使当前线程在锁存器倒计数至零以前一直等待，若是当前计数为零，则此方法马上返回 true 值。

• public boolean await(long timeout,TimeUnit unit) throws InterruptedException:使当前线程在锁存器倒计数至零以前一直等待，除非线程被中断或超出了指定的等待时间。若是当前计数为零，则此方法马上返回 true 值。若是当前计数大于零，则出于线程调度目的，将禁用当前线程，且在发生如下三种状况之一前，该线程将一直处于休眠状态：

  • 因为调用 countDown() 方法，计数到达零；或者其余某个线程中断当前线程；或者已超出指定的等待时间。若是计数到达零，则该方法返回 true 值。
  • 若是当前线程：在进入此方法时已经设置了该线程的中断状态；或者在等待时被中断，则抛出 InterruptedException，而且清除当前线程的已中断状态。
  • 若是超出了指定的等待时间，则返回值为 false。若是该时间小于等于零，则此方法根本不会等待。

【例子

模拟一个应用程序：在正式开始工做前须要初始化数据，初始化数据使用三个线程，正式执行须要五个线程：

• 初始化线程

public class InitThread implements Runnable{

    private CountDownLatch downLatch;
    private String name;

    public InitThread(CountDownLatch downLatch, String name){
        this.downLatch = downLatch;
        this.name = name;
    }

    public void run() {
        this.doWork();
        try{
            TimeUnit.SECONDS.sleep(new Random().nextInt(10));
        }catch(InterruptedException ie){
        }
        System.out.println(this.name + "初始化数据完成");
        //计数器减一
        this.downLatch.countDown();
    }
    private void doWork(){
        System.out.println(this.name + "正在初始化数据... ...");
    }

}

• 初始化线程监视器

/**
 * 检测初始化数据监视器,由于须要判断是否初始化线程所有执行完毕，这里用callable返回结果。runnable不能返回值因此没法判断。
 */
public class InitMonitor implements Callable<String>{

    private ExecutorService executor;
    private CountDownLatch initLatch;
    private List<Runnable> initThreads;

    public InitMonitor(ExecutorService executor){
        this.executor = executor;
        //初始化线程：3个
        initLatch = new CountDownLatch(3);
        initThreads = Arrays.asList(new InitThread(initLatch,"InitOne"),
                                    new InitThread(initLatch,"InitTwo"),
                                    new InitThread(initLatch,"InitThree"));
    }

    public String call() {
        System.out.println("=========初始化START==========");
        initThreads.stream().forEach(initThread -> executor.submit(initThread));
        try {
            initLatch.await();
        } catch (InterruptedException e) {
        }
        System.out.println("***********初始化END*************");
        return "INIT_SUCCESS";
    }
}

• 工做线程

public class ExecuteThread implements Runnable{

    private CountDownLatch downLatch;
    private String name;

    public ExecuteThread(CountDownLatch downLatch, String name){
        this.downLatch = downLatch;
        this.name = name;
    }

    public void run() {
        this.doWork();
        try{
            TimeUnit.SECONDS.sleep(new Random().nextInt(10));
        }catch(InterruptedException ie){
        }
        System.out.println(this.name + "执行完成");
        //计数器减一
        this.downLatch.countDown();
    }
    private void doWork(){
        System.out.println(this.name + "正在执行... ...");
    }
}

• 工做线程监视器

public class ExecuteMonitor implements Callable<String>{

    private ExecutorService executor;
    private CountDownLatch executeLatch;
    private List<Runnable> executeThreads;

    public ExecuteMonitor(ExecutorService executor){
        this.executor = executor;
        //执行线程:5个
        executeLatch = new CountDownLatch(5);
        executeThreads = Arrays.asList(new ExecuteThread(executeLatch,"ExecuteOne"),
                                        new ExecuteThread(executeLatch,"ExecuteTwo"),
                                        new ExecuteThread(executeLatch,"ExecuteThree"),
                                        new ExecuteThread(executeLatch,"ExecuteFour"),
                                        new ExecuteThread(executeLatch,"ExecuteFive"));
    }

    public String call() {
        System.out.println("========执行START========");
        executeThreads.stream().forEach(executeThread -> executor.submit(executeThread));
        try {
            executeLatch.await();
        } catch (InterruptedException e) {
        }
        System.out.println("*********执行END*********");
        return "EXECUTE_SUCCESS";
    }
}

• 应用程序

public class Application implements Runnable{

    private ExecutorService executor;
    private InitMonitor initMonitor;
    private ExecuteMonitor executeMonitor;

    public Application(ExecutorService executor){
        this.executor = executor;
        initMonitor = new InitMonitor(executor);
        executeMonitor = new ExecuteMonitor(executor);
    }
    @Override
    public void run() {
        System.out.println("===============应用程序执行开始====================》》》");
        FutureTask<String> initTask = new FutureTask<String>(initMonitor);
        executor.submit(initTask);
        try {
            //若是初始化成功开始执行工做线程，在调用get()时，若是没有执行完成会自动阻塞，因此这里不须要使用isDone检测。
            if("INIT_SUCCESS".equals(initTask.get())){
                FutureTask<String> executeTask = new FutureTask<String>(executeMonitor);
                executor.submit(executeTask);
                if("EXECUTE_SUCCESS".equals(executeTask.get())){
                    executor.shutdown();
                    System.out.println("===============应用程序执行完毕====================");
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

• 客户端

客户端必定尽可能简介，全部细节所有屏蔽，这里只留下一个能够自定义线程池给用户自行选择

public class Test {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        Application application = new Application(executor);
        application.run();
    }
}