Java多线程中的CountDownLatch解析

2023-11-16 09:38:55 作者：进击的猫

这篇文章主要介绍了Java多线程中的CountDownLatch解析,CountDownLatch是一个阻塞部分线程直到其他线程执行完成后唤醒的同步计数器,核心是其内部类Sync继承于AQS,同时也是利用的AQS的同步原理,也称之为闭锁,需要的朋友可以参考下

一、概念简介

CountDownLatch是一个阻塞部分线程直到其他线程执行完成后唤醒的同步计数器

核心是其内部类Sync继承于AQS,同时也是利用的AQS的同步原理,也称之为闭锁。

二、使用场景

当主线程进行执行时,利用构造方法初始化一个同步数state(AQS原理),主线程调用await方法进行阻塞主线程即谁调用谁阻塞,其它线程调用countDown方法会对计数器减1直到0,会精准唤醒被阻塞线程即被await方法阻塞的线程。

(1)用于多种数据源数据汇总;

(2)等待某一时间点才执行逻辑如加载缓存、加载配置等;

注意：为了程序的健壮性,尽量给出合适的时间,防止子线程中断导致线程无法唤醒的情况发生。

三、特点

(1)子线程调用countDown方法只会减1,不会阻塞线程;

(2)主线程调用await方法会导致其被阻塞,当计数器state被其他线程调用countDown方法减至0会唤醒被阻塞的线程;

(3)当主线程发生中断会抛出异常,导致无法唤醒主线程即无法达到屏障点。

CountDownLatch简单使用

 public static void main(String[] args) {
        System.out.println("main 线程开始执行！");
        CountDownLatch latch = new CountDownLatch(5);//初始化同步数
        for (int i = 0; i < 5; i++) {
            int threadId = i+1;
            new Thread(()->{
                System.out.println("线程"+threadId+"执行！");
                latch.countDown();
            }).start();//java8 lamda表达式
        }
        System.out.println("即将被阻塞！");
        try {
            latch.await();//阻塞主线程,等待子线程将state减至0被唤醒
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("main 线程执行完毕！");
    }

四、CountDownLatch源码分析

(1)构造函数

 /**
  * CountDownLatch唯一的构造函数,实例化时只能使用指定同步数的构造方法
  */
  public CountDownLatch(int count) {
      if (count < 0) throw new IllegalArgumentException("count < 0");
      this.sync = new Sync(count);//利用内部类(继承AQS)对state进行设置初始化大小
  }

(2)await方法(核心)

CountDownLatch类：

public void await() throws InterruptedException {
    //核心成员变量sync调用AQS中的方法acquireSharedInterruptibly
    sync.acquireSharedInterruptibly(1);
}

AQS类：

  public final void acquireSharedInterruptibly(int arg)
              throws InterruptedException {
      if (Thread.interrupted())//判断是否有中断标志
          throw new InterruptedException();
          /**
          * 该方法是由子类重写,AQS强制其子类重写,否则报错
          * 根据if中的值判断是否需要阻塞操作 1代表不需要阻塞 -1代表需要阻塞
          */
      if (tryAcquireShared(arg) < 0)
          doAcquireSharedInterruptibly(arg);//调用AQS共享锁阻塞操作
  }

Sync类：

/**
 * 获取同步数并判断是否需要唤醒
 * 同步数state为0,则需要唤醒返回1即不需要阻塞
 * 同步数state不为1,则不需要唤醒,返回-1后的操作即阻塞
 */
 protected int tryAcquireShared(int acquires) {
     return (getState() == 0) ? 1 : -1;//获取AQS中的state进行返回是否需要进行阻塞操作
 }

//以共享锁的方式进行阻塞
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    /**
    * addWaiter方法主要是基于当前线程创建一个等待着并入队且会创建一个哨兵节点
    * addWaiter具体细节和其内部enq初始化队列方法请转入AQS分析
    */
    final Node node = addWaiter(Node.SHARED);//以共享锁创建一个等待者node
    boolean failed = true;
    try {
        for (;;) {//自旋,是否需要阻塞
            final Node p = node.predecessor();//当前线程的前继节点
            if (p == head) {//前继节点是否为头节点
                int r = tryAcquireShared(arg);//尝试获取共享锁即是否需要阻塞1和-1值
                if (r >= 0) {//当其大于等于时,r值只能时1或者-1,满足该条件时则说明不需要阻塞
                    setHeadAndPropagate(node, r);//设置新的头结点并释放共享锁
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            /**
            * shouldParkAfterFailedAcquire主要是改变前节点的等待信号量
            * parkAndCheckInterrupt在前者返回TRUE的情况下会直接调用LockSupport.park()进行阻塞
            * 上述两种方法在AQS分析中可找到详细解释
            */
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();//上述两个条件满足则代表线程被中断过
        }
    } finally {
        if (failed)//出现异常且未执行for循环中改变该failed值
            cancelAcquire(node);//取消超时节点和当前节点取消唤醒,AQS原理分析中详细讲解
    }
}

(3)countDown方法(核心)

//用于子线程调用将同步数-1
public void countDown() {
    sync.releaseShared(1);//通过内部成员变量sync调用内部Sync类继承AQS中的释放方法
}