CountDownLatch/CyclicBarrier/Semaphore 使用过吗？

CountDownLatch/CyclicBarrier/Semaphore 使用过吗？下面详细介绍用法：

一，（等待多线程完成的）CountDownLatch

     背景;

  • countDownLatch（同步援助）是在java1.5被引入，跟它一起被引入的工具类还有CyclicBarrier(同步援助)、Semaphore(计数信号量)、concurrentHashMap和BlockingQueue（阻塞队列）。
  • 存在于java.util.cucurrent包下。

    概念理解：

     让一些线程阻塞，直到另外一些线程完成一系列操作后才被唤醒。

   主要方法介绍

     通过一个计数器来实现的，计数器的初始值是线程的数量。await方法会被阻塞，每当一个线程执行完毕后，countDown方法会让计数器的值-1，不会阻塞，当计数器的值为0时，表示其他线程都执行完毕后，调用await方法的线程会被唤醒，继续执行。

  源码：

       (1)，countDownLatch类中只提供了一个构造器：

CountDownLatch(int count) 
构造一个 CountDownLatch初始化与给定的数。

      （2）类中有三个方法是最重要的： 

//调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
public void await() throws InterruptedException { };   
//和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };  
//将count值减1
public void countDown() { };

使用举例：

 public static void CountDownLatchUsed() throws InterruptedException {
        CountDownLatch downLatch = new CountDownLatch(6);
        for (int i = 0; i < 6; i++) {
          new Thread(()->{
      System.out.println(Thread.currentThread().getName()+"\t 下自习走人");
                downLatch.countDown();
            },String.valueOf(i)).start();
        }
        downLatch.await();
    System.out.println(Thread.currentThread().getName()+"自习室关门走人");
    }

运行结果：

  CountDownLatch/CyclicBarrier/Semaphore 使用过吗？

 

 

 小结：CountDownLatch和CyclicBarrier区别：

  1. countDownLatch是一个计数器，线程完成一个记录一个，计数器递减，只能使用一次。
  2. CyclicBarrier的计数器更像一个阀门，需要所有线程都到达，然后继续执行，计数器从0开始递增，同时提供reset()方法初始化计数值，可以多次使用。

了解更多：https://blog.csdn.net/chenssy/article/details/49794141#commentBox

二，（同步屏障）CyclicBarrier

   概念理解：

   CyclicBarrier字面上就是可循环使用的屏障。当一组线程得到一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才会继续工作。进入屏障通过await方法。

主要方法介绍：

           CountDownLatch/CyclicBarrier/Semaphore 使用过吗？

CyclicBarrier的内部是使用重入锁ReentrantLock和Condition。它有两个构造函数：    

  • CyclicBarrier(int parties)：创建一个新的 CyclicBarrier，它将在给定数量的线程处于等待状态时启动，但它不会在启动 barrier 时执行预定义的操作。（直到到达屏障时才会去执行）

  • CyclicBarrier(int parties, Runnable barrierAction)：创建一个新的 CyclicBarrier，它将在给定数量的线程处于等待状态时启动，并在启动 barrier 时执行给定的屏障操作，该操作由最后一个进入 barrier 的线程执行。

说明:

（1）parties:拦截线程的数量：

（2）barrierAction：为CyclicBarrier接收的Runnable命令，用于在线程到达屏障时，优先执行barrierAction ，用于处理更加复杂的业务场景。

   源码：

  private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }
  public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

在CyclicBarrier中最重要的方法莫过于await()方法，在所有参与者都已经在此 barrier 上调用 await 方法之前，将一直等待。如下：

 public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

await()方法内部调用dowait(boolean timed, long nanos)方法：

其实await()的处理逻辑还是比较简单的：如果该线程不是到达的最后一个线程，则他会一直处于等待状态，除非发生以下情况：

  1. 最后一个线程到达，即index == 0
  2. 超出了指定时间（超时等待）
  3. 其他的某个线程中断当前线程
  4. 其他的某个线程中断另一个等待的线程
  5. 其他的某个线程在等待barrier超时
  6. 其他的某个线程在此barrier调用reset()方法。reset()方法用于将屏障重置为初始状态。

应用场景：

   CyclicBarrier试用与多线程结果合并的操作，用于多线程计算数据，最后合并计算结果的应用场景。比如我们需要统计多个Excel中的数据，然后等到一个总结果。我们可以通过多线程处理每一个Excel，执行完成后得到相应的结果，最后通过barrierAction来计算这些线程的计算结果，得到所有Excel的总和。

应用实例：

  比如开会要等所有人到齐了，才会开会：

public class BlockQueueDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() {
            @Override
            public void run() {
                System.out.println("所有人都到齐了吧，咱们开会。。。");
            }
        });
        for (int i = 1; i <= 4; i++) {
            final int tempInt = i;
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "\t" + tempInt + "已到位");
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }, String.valueOf(i)).start();
        }
    }
}

运行结果：

 CountDownLatch/CyclicBarrier/Semaphore 使用过吗？

 

 

了解更多：https://blog.csdn.net/chenssy/article/details/70160595 

三，（控制并发线程数的）Semaphore 

        概念理解：

     信号量Semaphore是一个控制访问多个共享资源的计数器，它本质上是一个“共享锁”。

     理论知识：

    （1） Java并发提供了两种加锁模式，共享锁和独占锁。对于独占锁,他每次只能有一个线程持有，而共享锁则不同，它允许多个线程并行持有锁，并发的访问共享资源。

    （2）独占锁采用的是一种悲观的加锁策略，对于写而言为冲突是必须的，但对于读而言可以进行共享，因为它不印象数据的一致性，如果某个只读线程获取独占锁，则其他读线程都只能等待了，这种情况下就限制了不必要的并发性，降低了吞吐量。而共享锁则不同，它放宽了加锁的条件，采用了乐观锁机制，它是允许多个读线程同时访问同一个共享资源的。  

Semaphore官方解释：

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

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

CountDownLatch和Semephore优缺点对比

     CountDownLatch的问题是不能复用。比如count=3，那么加到3，就不能继续操作了。而Semaphore可以解决这个问题，比如6辆车3个停车位，对于CountDownLatch只能停3辆车，而Semaphore可以停6辆车，车位空出来后，其它车可以占有，这就涉及到了Semaphore.accquire()和Semaphore.release()方法。

    生活实例版本解释：

         我们假设只有三个理发师、一个接待人。一开始来了五个客人，接待人则安排三个客人进行理发，其余两个人必须在那里等着，此后每个来理发店的人都必须等待。一段时间后，一个理发师完成理发后，接待人则安排另一个人（公平还是非公平机制呢？？）来理发。在这里理发师则相当于公共资源，接待人则相当于信号量（Semaphore），客户相当于线程。

   深入理解：

       进一步讲，我们确定信号量Semaphore是一个非负整数（>=1）。当一个线程想要访问某个共享资源时，它必须要先获取Semaphore，当Semaphore >0时，获取该资源并使Semaphore – 1。如果Semaphore值 = 0，则表示全部的共享资源已经被其他线程全部占用，线程必须要等待其他线程释放资源。当线程释放资源时，Semaphore则+1；

       当信号量Semaphore = 1 时，它可以当作互斥锁使用。其中0、1就相当于它的状态，当=1时表示其他线程可以获取，当=0时，排他，即其他线程必须要等待。

源码分析

（1）Semaphore的结构如下：

      CountDownLatch/CyclicBarrier/Semaphore 使用过吗？

从上面可以看出，Semaphore和ReentrantLock一样，都是包含公平锁（FairySync）和非公平锁（NonfairSync），两个锁都是继承Sync，而Sync也是继承自AQS。其构造函数如下：

/**
     * 创建具有给定的许可数和非公平的公平设置的 Semaphore。  (默认是非公平锁)
     */
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }
    
    /**
     * 创建具有给定的许可数和给定的公平设置的 Semaphore。
     */
    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

（2）信号量的获取：acquire()

 

             在ReentrantLock中已经阐述过，公平锁和非公平锁获取锁机制的差别：对于公平锁而言，如果当前线程不在CLH队列（CLH锁是一个自旋锁。能确保无饥饿性。提供先来先服务的公平性）的头部，则需要排队等候，而非公平锁则不同，它无论当前线程处于CLH队列的何处都会直接获取锁。所以公平信号量和非公平信号量的区别也一样。

关于CLH 自旋队列锁 更多知识：https://blog.csdn.net/aesop_wubo/article/details/7533186

源码：

public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    } 
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

对于公平信号量和非公平信号量，他们机制的差异就体现在traAcquireShared()方法中：

公平锁 tryAcquireShared() 尝试获取信号量

源码： 

protected int tryAcquireShared(int acquires) {
        for (;;) {
            //判断该线程是否位于CLH队列的列头，如果是的话返回 -1,调用doAcquireSharedInterruptibly()
            if (hasQueuedPredecessors())
                return -1;
            //获取当前的信号量许可
            int available = getState();
            //设置“获得acquires个信号量许可之后，剩余的信号量许可数”
            int remaining = available - acquires;
            
            //如果剩余信号量 > 0 ,则设置“可获取的信号量”为remaining
            if (remaining < 0 || compareAndSetState(available, remaining))
                return remaining;
        }
    }

注意 ： tryAcquireShared是尝试获取 信号量，remaining表示下次可获取的信号量。

（2）hasQueuedPredecessors（） 是否有队列同步器

 

对于hasQueuedPredecessors、compareAndSetState在ReentrantLock中已经阐述了，hasQueuedPredecessors用于判断该线程是否位于CLH队列列头，compareAndSetState用于设置state的，它是进行原子操作的。代码如下：

public final boolean hasQueuedPredecessors() {
        Node t = tail; // Read fields in reverse initialization order
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }
 
    protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

 （3） doAcquireSharedInterruptibly源代码如下：

CountDownLatch/CyclicBarrier/Semaphore 使用过吗？
private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            /*
             * 创建CLH队列的node节点，Node.SHARED表示该节点为共享锁
             */
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {
                    //获取该节点的前继节点
                    final Node p = node.predecessor();
                    //当p为头节点时，基于公平锁机制，线程尝试获取锁
                    if (p == head) {
                        //尝试获取锁
                        int r = tryAcquireShared(arg);    
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);  
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    //判断当前线程是否需要阻塞，如果阻塞的话，则一直处于阻塞状态知道获取共享锁为止
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
View Code

相关文章：

  • 2021-07-08
  • 2021-11-15
  • 2022-12-23
  • 2022-12-23
  • 2021-12-21
  • 2021-12-09
猜你喜欢
  • 2018-03-15
  • 2022-12-23
  • 2020-04-29
  • 2022-12-23
  • 2022-12-23
  • 2021-11-10
  • 2021-11-27
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode