java latch闭锁基本使用（结合future）

谁践踏了优雅 2022-01-21 08:57 274阅读 0赞

java闭锁用于多个线程共同执行后，统一执行一个动作。比如：多个线程执行计算操作，最后汇总到同一个线程执行汇总计算。需要注意的是，java中的闭锁是仅一次的。当闭锁打开后就会统一执行下面的动作。

下面简单的举一个例子：

worker线程进行工作，boss线程等到所有woker工作完成后，检查工作。

思路是这样的，首先使用java中的CountDownLatch，每个worker线程工作完成后都会对latch进行downLatch操作。boss线程启动后一直等待latch的计数降为0，然后对所有worker的工作进行检查。

代码如下：

package com.xueyou.demo.concurrent.latch;
    
    import java.util.concurrent.CountDownLatch;
    
    public class Worker implements Runnable {
        private CountDownLatch downLatch;
    
        public Worker(CountDownLatch downLatch) {
            this.downLatch = downLatch;
        }
    
        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getId() + " is working...");
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getId() + " finish the job.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                downLatch.countDown();
            }
        }
    }

package com.xueyou.demo.concurrent.latch;
    
    import java.util.concurrent.CountDownLatch;
    
    public class Boss implements Runnable {
        private CountDownLatch downLatch;
    
        public Boss(CountDownLatch downLatch) {
            this.downLatch = downLatch;
        }
    
        @Override
        public void run() {
            try {
                System.out.println("i am boss,i wait the job ok.....");
                downLatch.await();
                System.out.println("job ok haha.....");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

package com.xueyou.demo.concurrent.latch;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    public class LatchDemo {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            /**
             * 闭锁Latch 适用于单次进行等待的任务，有一个线程，等待其他线程工作完成再进行工作的场景下。
             * 由于latch在打开后不再关闭，所以不能重复的进行再次等待。
             *
             *
             * 下面的例子就是boss等待worker工作结束后在进行执行的例子。
             */
            CountDownLatch downLatch = new CountDownLatch(3);
            new Thread(new Boss(downLatch)).start();
            new Thread(new Worker(downLatch)).start();
            new Thread(new Worker(downLatch)).start();
            new Thread(new Worker(downLatch)).start();
        }    
    }

运行结果：

![20190609142231858.png][]

从上面的程序，能够看出，boss线程通过使用latch的await方法，被其他的worker阻塞了（因为他们共用同一个latch）。当所有的worker线程执行完成后，boss线程才会执行。

结合Future和FutureTask

现在boss有一个需求，就是获取每个worker的工作时长。需要如何处理呢？

首先每个worker线程需要返回自己的执行时间，同时boss能够接受到worker线程的结果。这里我们就需要使用future或者futuretask了。

直接上代码吧：

package com.xueyou.demo.concurrent.latch;
    
    import java.util.Random;
    import java.util.concurrent.Callable;
    import java.util.concurrent.CountDownLatch;
    
    public class WorkerWithResult implements Callable<Integer> {
        private CountDownLatch downLatch;
        private int workId;
    
        public WorkerWithResult(CountDownLatch downLatch, int workId) {
            this.downLatch = downLatch;
            this.workId = workId;
        }
    
        @Override
        public Integer call() throws Exception {
            int randomSleepTime = -1;
            try {
                System.out.println(workId + " is working...");
                randomSleepTime = new Random().nextInt(3000);
                Thread.sleep(randomSleepTime);
                System.out.println(workId + " finish the job.");
            } finally {
                downLatch.countDown();
                return randomSleepTime;
            }
        }
    }

package com.xueyou.demo.concurrent.latch;
    
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    public class BossWithResult implements Runnable {
        private CountDownLatch downLatch;
        //保存每个工人的工作时间
        private List<Future> workTimeUseList;
    
        public BossWithResult(CountDownLatch downLatch, List<Future> workTimeUseList) {
            this.downLatch = downLatch;
            this.workTimeUseList = workTimeUseList;
        }
    
        @Override
        public void run() {
            try {
                System.out.println("i am boss,i wait the job ok");
                downLatch.await();
                System.out.println("job ok haha.....");
                for (Future<Integer> workTimeFuture : workTimeUseList) {
                    System.out.println("work time is :" + workTimeFuture.get());
                }
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

注意上边的List<Future>这里没有使用List<Integer>。原因是如果使用了List<Integer>，那么等到future调用get的时候就会阻塞，导致boss线程不能提前启动，执行的顺序就会变成所有的worker线程执行完成，然后boss线程才会启动。

package com.xueyou.demo.concurrent.latch;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    public class LatchDemo {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            /**
             * 下面的程序使用的future对每个worker工作的时间进行统计，最后通过latch的countDown到0，
             * 通知boss，boss通过future传递的值能够知道每个worker的工作时间。
             * 这里有一点需要注意，在给boss传递参数的时候，可以让boss直接获得future中的值，但是如果使用
             * 这种方式，就没有必要使用latch了，因为在每个worker的值时需要使用future.get()，能够创建完成
             * 参数的时候，worker线程应该已经结束了。所以就没有必要使用latch了。
             *
             * 如果像下面程序传递的是future，然后在boss的线程中对future进行取值，就是需要latch的。因为在boss线程开
             * 开始的时候future没有执行完成，需要latch最后countDown到0，才能保证所有的future中都有结果了。
             */
            CountDownLatch downLatch = new CountDownLatch(4);
            ExecutorService executor = Executors.newFixedThreadPool(4);
            Future<Integer> future = executor.submit(new WorkerWithResult(downLatch, 1));
            Future<Integer> future1 = executor.submit(new WorkerWithResult(downLatch, 2));
            Future<Integer> future2 = executor.submit(new WorkerWithResult(downLatch, 3));
            FutureTask<Integer> futureTask = new FutureTask<>(new WorkerWithResult(downLatch, 4));
            executor.submit(futureTask);
            List<Future> workTimeList = new ArrayList<>();
            workTimeList.add(future);
            workTimeList.add(future1);
            workTimeList.add(future2);
            workTimeList.add(futureTask);
            executor.submit(new BossWithResult(downLatch, workTimeList));
            executor.shutdown();
        }
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbGQ0NmNhdA_size_16_color_FFFFFF_t_70][]

这样boss线程就能获得每个worker线程的执行时间了。

[20190609142231858.png]: /images/20220121/8d15cdc15b1847eb911006baf5ac3055.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dpbGQ0NmNhdA_size_16_color_FFFFFF_t_70]: /images/20220121/13d4bf12aa2f490db1e0b5149a4fe270.png