Java并发编程一CountDownLatch、CyclicBarrier、Semaphore初使用

一时失言乱红尘 2023-07-07 03:36 62阅读 0赞

推荐：[Java并发编程汇总][Java]

## Java并发编程一CountDownLatch、CyclicBarrier、Semaphore初使用 ##

`CountDownLatch`、`CyclicBarrier`、`Semaphore`这些线程协作工具类是基于`AQS`的，看完这篇博客后可以去看下面这篇博客，了解它们是如何实现的。

[Java并发之AQS详解][Java_AQS]

**CountDownLatch**

`CountDownLatch`可以实现一个线程等待多个线程、多个线程等待一个线程、多个线程等待多个线程（这里不涉及）。

我们首先来看看怎么实现一个线程等待多个线程吧。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]

工厂中，对产品需要进行质检，`5`个工人进行检查，所有人都认为通过，这个产品才算通过。  
代码：

package flowcontrol.countdownlatch;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class CountDownLatchDemo1 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch latch = new CountDownLatch(5);
            ExecutorService service = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                final int no = i + 1;
                Runnable runnable = new Runnable() { 
    
                    @Override
                    public void run() { 
                        try { 
                            Thread.sleep((long) (Math.random() * 10000));
                            System.out.println("No." + no + "完成了检查。");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            latch.countDown();
                        }
                    }
                };
                service.submit(runnable);
            }
            System.out.println("等待5个人检查完.....");
            latch.await();
            System.out.println("所有人都完成了工作，进入下一个环节。");
        }
    }

输出：

等待5个人检查完.....
    No.4完成了检查。
    No.3完成了检查。
    No.2完成了检查。
    No.1完成了检查。
    No.5完成了检查。
    所有人都完成了工作，进入下一个环节。

输出也符合预期。

我们来看一看程序中使用的`CountDownLatch`构造器源码。

public CountDownLatch(int count) { 
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }

进入`Sync(count)`方法，`Sync`是`CountDownLatch`的内部类，它继承了`AbstractQueuedSynchronizer`，也就是传说中`AQS`，我们调用的`CountDownLatch`构造器传入的`count`其实对应于`AQS`的`state`，所以现在只需要知道怎么用`CountDownLatch`，知道用法后再去看底层就会更加容易理解，这里就不多说`AQS`了。

private static final class Sync extends AbstractQueuedSynchronizer { 
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) { 
                setState(count);
            }
    
            int getCount() { 
                return getState();
            }
    
            protected int tryAcquireShared(int acquires) { 
                return (getState() == 0) ? 1 : -1;
            }
    
            protected boolean tryReleaseShared(int releases) { 
                // Decrement count; signal when transition to zero
                for (;;) { 
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
        }

再来实现一下多个线程等待一个线程。

模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开始跑步。  
代码：

package flowcontrol.countdownlatch;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class CountDownLatchDemo2 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch begin = new CountDownLatch(1);
            ExecutorService service = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                final int no = i + 1;
                Runnable runnable = new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("No." + no + "准备完毕，等待发令枪");
                        try { 
                            begin.await();
                            System.out.println("No." + no + "开始跑步了");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                };
                service.submit(runnable);
            }
            //裁判员检查发令枪...
            Thread.sleep(5000);
            System.out.println("发令枪响，比赛开始！");
            begin.countDown();
        }
    }

输出：

No.1准备完毕，等待发令枪
    No.4准备完毕，等待发令枪
    No.3准备完毕，等待发令枪
    No.2准备完毕，等待发令枪
    No.5准备完毕，等待发令枪
    发令枪响，比赛开始！
    No.1开始跑步了
    No.4开始跑步了
    No.3开始跑步了
    No.2开始跑步了
    No.5开始跑步了

输出也符合预期。

结合前面两种实现，我们来实现一个复合版的。

模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开始跑步。当所有人都到终点后，比赛结束。  
代码：

package flowcontrol.countdownlatch;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class CountDownLatchDemo1And2 { 
    
        public static void main(String[] args) throws InterruptedException { 
            CountDownLatch begin = new CountDownLatch(1);
    
            CountDownLatch end = new CountDownLatch(5);
            ExecutorService service = Executors.newFixedThreadPool(5);
            for (int i = 0; i < 5; i++) { 
                final int no = i + 1;
                Runnable runnable = new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("No." + no + "准备完毕，等待发令枪");
                        try { 
                            begin.await();
                            System.out.println("No." + no + "开始跑步了");
                            Thread.sleep((long) (Math.random() * 10000));
                            System.out.println("No." + no + "跑到终点了");
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        } finally { 
                            end.countDown();
                        }
                    }
                };
                service.submit(runnable);
            }
            //裁判员检查发令枪...
            Thread.sleep(5000);
            System.out.println("发令枪响，比赛开始！");
            begin.countDown();
    
            end.await();
            System.out.println("所有人到达终点，比赛结束");
        }
    }

输出：

No.1准备完毕，等待发令枪
    No.4准备完毕，等待发令枪
    No.3准备完毕，等待发令枪
    No.2准备完毕，等待发令枪
    No.5准备完毕，等待发令枪
    发令枪响，比赛开始！
    No.1开始跑步了
    No.4开始跑步了
    No.3开始跑步了
    No.2开始跑步了
    No.5开始跑步了
    No.1跑到终点了
    No.4跑到终点了
    No.5跑到终点了
    No.3跑到终点了
    No.2跑到终点了
    所有人到达终点，比赛结束

输出也符合预期。

**CyclicBarrier**

`CyclicBarrier`可以让多个线程一起相互等待（阻塞），当相互等待的线程数量达到我们指定的数量时，它们就可以不用再阻塞了，可以接着执行了，不过在它们接着执行之前会触发一个我们指定的线程，具体看代码吧。

代码：

package flowcontrol.cyclicbarrier;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class CyclicBarrierDemo { 
        public static void main(String[] args) { 
            CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("所有人都到场了， 大家统一出发！");
                }
            });
            for (int i = 0; i < 10; i++) { 
                new Thread(new Task(i, cyclicBarrier)).start();
            }
        }
    
        static class Task implements Runnable{ 
            private int id;
            private CyclicBarrier cyclicBarrier;
    
            public Task(int id, CyclicBarrier cyclicBarrier) { 
                this.id = id;
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() { 
                System.out.println("线程" + id + "现在前往集合地点");
                try { 
                    Thread.sleep((long) (Math.random()*10000));
                    System.out.println("线程"+id+"到了集合地点，开始等待其他人到达");
                    cyclicBarrier.await();
                    System.out.println("线程"+id+"出发了");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } catch (BrokenBarrierException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

输出：

线程0现在前往集合地点
    线程2现在前往集合地点
    线程5现在前往集合地点
    线程3现在前往集合地点
    线程4现在前往集合地点
    线程9现在前往集合地点
    线程1现在前往集合地点
    线程8现在前往集合地点
    线程7现在前往集合地点
    线程6现在前往集合地点
    线程6到了集合地点，开始等待其他人到达
    线程2到了集合地点，开始等待其他人到达
    线程4到了集合地点，开始等待其他人到达
    线程0到了集合地点，开始等待其他人到达
    线程7到了集合地点，开始等待其他人到达
    所有人都到场了， 大家统一出发！
    线程7出发了
    线程6出发了
    线程2出发了
    线程4出发了
    线程0出发了
    线程5到了集合地点，开始等待其他人到达
    线程3到了集合地点，开始等待其他人到达
    线程8到了集合地点，开始等待其他人到达
    线程9到了集合地点，开始等待其他人到达
    线程1到了集合地点，开始等待其他人到达
    所有人都到场了， 大家统一出发！
    线程1出发了
    线程5出发了
    线程3出发了
    线程8出发了
    线程9出发了

输出符合预期。

**Semaphore**

`Semaphore`可以控制一起执行线程的数量，也就是并发数，当某个操作很耗资源时，我们不允许太大的并发量时，我们可以使用`Semaphore`来进行限制并发量，如下图所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]  
这是通过信号量来实现的，也对应于`AQS`的`state`，我们给某个服务设置一定数量的信号量，当一个线程来获取资源时，信号量就减一些（数量由我们自己指定，根据业务需求），没有信号量的线程就得阻塞。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]  
当线程成功获取资源后，也就是线程可以结束了，该线程会返还使用的信号量，以供其他线程使用。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 4]

代码：

package flowcontrol.semaphore;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Semaphore;
    
    public class SemaphoreDemo { 
    
        static Semaphore semaphore = new Semaphore(4, true);
    
        public static void main(String[] args) { 
            ExecutorService service = Executors.newFixedThreadPool(50);
            for (int i = 0; i < 100; i++) { 
                service.submit(new Task());
            }
            service.shutdown();
        }
    
        static class Task implements Runnable { 
    
            @Override
            public void run() { 
                try { 
                    semaphore.acquire(2);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "拿到了许可证");
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + "释放了许可证");
                semaphore.release(2);
            }
        }
    }

输出：

pool-1-thread-1拿到了许可证
    pool-1-thread-2拿到了许可证
    pool-1-thread-1释放了许可证
    pool-1-thread-2释放了许可证
    pool-1-thread-5拿到了许可证
    pool-1-thread-6拿到了许可证
    pool-1-thread-5释放了许可证
    pool-1-thread-6释放了许可证
    pool-1-thread-3拿到了许可证
    pool-1-thread-4拿到了许可证
    pool-1-thread-3释放了许可证
    pool-1-thread-4释放了许可证
    pool-1-thread-7拿到了许可证
    pool-1-thread-8拿到了许可证
    pool-1-thread-8释放了许可证
    pool-1-thread-7释放了许可证

输出也符合预期。

[Java]: https://kaven.blog.csdn.net/article/details/104233588
[Java_AQS]: https://blog.csdn.net/qq_37960603/article/details/104395660
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200220192722477.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200220171420448.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200220171910690.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200220172017902.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200220172304320.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70