并发工具类CountDownLatch、CyclicBarrier、Semaphore、Exchanger详解

比眉伴天荒 2022-05-28 06:49 289阅读 0赞

# 一、CountDownLatch #

一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。

CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。

CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。

# 我们先来2个例子： #

## 不用CountDownLatch来完成一个文本所有数字的累积求和 ##

**nums.txt**文本文件

10,20,30,33,12,23
    21,12,18,45,11
    23,45,67,78,89
    12,12

import java.io.BufferedReader;
    import java.io.FileReader;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    public class Demo { 
    
        private int[] nums;
    
        public Demo(int line) {
            nums = new int[line];
        }
    
        public void calc(String line, int index) {
            String[] nus = line.split(","); // 切分出每个值
            int total = 0;
            for (String num : nus) {
                total += Integer.parseInt(num);
            }
            nums[index] = total; // 把计算的结果放到数组中指定的位置
            System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为：" + total);
        }
    
        public void sum() {
            System.out.println("汇总线程开始执行... ");
            int total = 0;
            for (int i = 0; i < nums.length; i++) {
                total += nums[i];
            }
            System.out.println("最终的结果为：" + total);
        }
    
        public static void main(String[] args) {
    
            List<String> contents = readFile();
            int lineCount = contents.size();
    
            Demo d = new Demo(lineCount);
            for (int i = 0; i < lineCount; i++) {
                final int j = i;
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        d.calc(contents.get(j), j);
                    }
                }).start();
            }
    
            while (Thread.activeCount() > 1) {
    
            }
    
            d.sum();
        }
    
        private static List<String> readFile() {
            List<String> contents = new ArrayList<>();
            String line = null;
            BufferedReader br = null;
            try {
                br = new BufferedReader(new FileReader("d:\\nums.txt"));
                while ((line = br.readLine()) != null) {
                    contents.add(line);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (br != null) {
                    try {
                        br.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
            return contents;
        }

![这里写图片描述][70]

用CountDownLatch之后

import java.io.BufferedReader;
    import java.io.FileReader;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    public class Demo2 { 
    
        private int[] nums;
    
        public Demo2(int line) {
            nums = new int[line];
        }
    
        public void calc(String line, int index, CountDownLatch latch) {
            String[] nus = line.split(","); // 切分出每个值
            int total = 0;
            for (String num : nus) {
                total += Integer.parseInt(num);
            }
            nums[index] = total; // 把计算的结果放到数组中指定的位置
            System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为：" + total);
            latch.countDown();
        }
    
        public void sum() {
            System.out.println("汇总线程开始执行... ");
            int total = 0;
            for (int i = 0; i < nums.length; i++) {
                total += nums[i];
            }
            System.out.println("最终的结果为：" + total);
        }
    
        public static void main(String[] args) {
    
            List<String> contents = readFile();
            int lineCount = contents.size();
    
            CountDownLatch latch = new CountDownLatch(lineCount);
    
            Demo2 d = new Demo2(lineCount);
            for (int i = 0; i < lineCount; i++) {
                final int j = i;
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        d.calc(contents.get(j), j, latch);
                    }
                }).start();
            }
    
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            d.sum();
        }
    
        private static List<String> readFile() {
            List<String> contents = new ArrayList<>();
            String line = null;
            BufferedReader br = null;
            try {
                br = new BufferedReader(new FileReader("d:\\nums.txt"));
                while ((line = br.readLine()) != null) {
                    contents.add(line);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (br != null) {
                    try {
                        br.close();
                    } catch (IOException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
            return contents;
        }
    
    }

![这里写图片描述][70 1]

![这里写图片描述][70 2]  
这里其实就是把传统的做法变化了一下、CountDownLatch利用一个计数器来完成，这里面最主要的方法就是上面图片中所标识的2个方法、在上述实例中先完成行数的统计传入CountDownLatch，在线程每完成一个就调用CountDownLatch的countDown方法来将数量减一、达到零时，则释放所有等待线程。

## CountDownLatch实现原理 ##

CountDownLatch是继承了AbstractQueuedSynchronizer类，一个共享锁的实现  
![这里写图片描述][70 3]  
![这里写图片描述][70 4]  
![这里写图片描述][70 5]  
上图的await方法就是判断getState的值是否等于0等于0执行不等于0返回-1所以不执行  
![这里写图片描述][70 6]  
这里countDown就是值减一，而我们总行数减到0之后所有线程便执行了

# 二、CyclicBarrier #

一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环 的 barrier。

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作 很有用。  
![这里写图片描述][70 7]  
如：一个开会的例子、不管你什么时候到的，一定要等到规定人数到了之后才进行、否则就一直等待，直到人数达到规定数量才继续向下执行

import java.util.Random;
    import java.util.concurrent.CyclicBarrier;
    
    public class Demo {
    
        Random random = new Random();
    
        public void meeting(CyclicBarrier barrier) {
            try {
                Thread.sleep(random.nextInt(4000));
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " 到达会议室，等待开会..");
    
    
            try {
                barrier.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    
        public static void main(String[] args) {
            Demo demo = new Demo();
    
            CyclicBarrier barrier = new CyclicBarrier(10, new Runnable() {
                @Override
                public void run() {
                    System.out.println("好！我们开始开会...");
                }
            });
    
            for (int i = 0; i < 10; i++) {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        demo.meeting(barrier);
                    }
                }).start();
            }
    
        }
    
    }

## CyclicBarrier实现原理 ##

![这里写图片描述][70 8]  
这里的CyclicBarrier其实和CountDownLatch差不多、这里还传入了一个事件，就是当条件达到的时候需要做什么事  
![这里写图片描述][70 9]  
这个的话其实我们就是调用await方法、而await方法就调用了dowait方法，这个dowait方法就是整个CyclicBarrier类的核心

private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                final Generation g = generation;
                //通过Generation的broken方法了判断是否中断，如果中断了，抛出一个中断的异常
                if (g.broken)
                    throw new BrokenBarrierException();
                //判断这个线程有没有被中断过
                if (Thread.interrupted()) {
                    breakBarrier();
                    throw new InterruptedException();
                }
                //进来一个线程，count就递减，说明等待的线程就少一个
                int index = --count;
                if (index == 0) {  // tripped
                //当index等待线程为0则叫醒所有的等待线程
                    boolean ranAction = false;
                    try {
                        //在判断barrierAction是否有事件传入，有就执行
                        final Runnable command = barrierCommand;
                        if (command != null)
                            command.run();
                        ranAction = true;
                        nextGeneration();
                        return 0;
                    } finally {
                        //如果执行事件报错则打破屏障叫醒所有线程
                        if (!ranAction)
                            breakBarrier();
                    }
                }
                //后面就是不为0的处理判断是否重置超时什么的
                // loop until tripped, broken, interrupted, or timed out
                for (;;) {
                    try {
                        if (!timed)
                            trip.await();
                        else if (nanos > 0L)
                            nanos = trip.awaitNanos(nanos);
                    } catch (InterruptedException ie) {
                        if (g == generation && ! g.broken) {
                            breakBarrier();
                            throw ie;
                        } else {
                            // We're about to finish waiting even if we had not
                            // been interrupted, so this interrupt is deemed to
                            // "belong" to subsequent execution.
                            Thread.currentThread().interrupt();
                        }
                    }
    
                    if (g.broken)
                        throw new BrokenBarrierException();
    
                    if (g != generation)
                        return index;
    
                    if (timed && nanos <= 0L) {
                        breakBarrier();
                        throw new TimeoutException();
                    }
                }
            } finally {
                lock.unlock();
            }
        }

这里的话引用了ReentrantLock进行加锁，来保证线程是安全的，而Generation就是为了实现reset重置功能的  
![这里写图片描述][70 10]

# 三、Semaphore #

一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。

import java.util.concurrent.Semaphore;
    
    public class Demo {
    
        public void method (Semaphore semaphore) {
    
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            System.out.println(Thread.currentThread().getName() + " is run ...");
    
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            semaphore.release();
        }
    
    
        public static void main(String[] args) {
    
            Demo d = new Demo();
            Semaphore semaphore = new Semaphore(10);
    
            while(true) {
    
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        d.method(semaphore);
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                }).start();
            }
    
    
        }
    
    }

每次只也许n个线程执行、其余的线程将被等待在那不释放也不有些、直到前面n个线程有一个运行完毕后释放资源后、其余线程去竞争运行。与线程池的差不多，只不过Semaphore会创建线程等待在那、而线程池不会创建线程、都是规定只能有n个线程执行。

# 四、Exchanger #

可以在对中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给 exchange 方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序（比如遗传算法和管道设计）中很有用。

import java.util.concurrent.Exchanger;
    
    public class Demo {
    
        public void a (Exchanger<String> exch) {
    
            System.out.println("a 方法执行...");
    
            try {
                System.out.println("a 线程正在抓取数据...");
                Thread.sleep(2000);
                System.out.println("a 线程抓取到数据...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            String res = "12345";
    
            try {
                System.out.println("a 等待对比结果... 抓取数据为：" + res);
                String value = exch.exchange(res);
                System.out.println("a获取b交换过来的数据:" + value);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
        }
    
        public void b (Exchanger<String> exch) {
            System.out.println("b 方法开始执行...");
            try {
                System.out.println("b 方法开始抓取数据...");
                Thread.sleep(4000);
                System.out.println("b 方法抓取数据结束...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String res = "1234567";
    
            try {
                String value = exch.exchange(res);
                System.out.println("开始进行比对...");
                System.out.println("b获取a交换过来的数据:" + value);
                System.out.println("比对结果为：" + value.equals(res));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
    
            Demo d = new Demo();
            Exchanger<String> exch = new Exchanger<>();
            new Thread(new Runnable() {
                @Override
                public void run() {
                    d.a(exch);
                }
            }).start();
    
            new Thread(new Runnable() {
                @Override
                public void run() {
                    d.b(exch);
                }
            }).start();
    
        }
    
    }

![这里写图片描述][70 11]  
大家可以发现，其实Exchanger作用就是线程与线程之间的数据交换

[70]: /images/20220528/a0d0392e7d7a44b594350cfdc0e35313.png
[70 1]: /images/20220528/4a25a1032e7647c39191db07c886efa9.png
[70 2]: /images/20220528/8d5f0ab8b5ca4e2197f1f262ab77c96d.png
[70 3]: /images/20220528/60b113fff5ad4b249b0a37a2f8842a52.png
[70 4]: /images/20220528/81a0d1dc38bb426e8bf09c3c114d3e95.png
[70 5]: /images/20220528/42a98974d0dc4e6eafcc69abd3637582.png
[70 6]: /images/20220528/975501e88827448e8fdc7ac39a29da24.png
[70 7]: /images/20220528/9fdf139564634bfe8eafbde6e65050cf.png
[70 8]: /images/20220528/f6125ac2138f49d1b7f9570c9c8a5b48.png
[70 9]: /images/20220528/9ba8d1cbc4e64c49943527cf46ee9134.png
[70 10]: /images/20220528/878db9fadcd8446caf97ddd6ce426e17.png
[70 11]: /images/20220528/ce92181c5f994a58956ab0ba61ba67b0.png