CountDownLatch详解

2023-11-09

Java

从源码的角度来分析下它的工作原理

1、谁来决定公交车上的座位数？

公交车上的座位数是由汽车制造商决定的，在 CountDownLatch 中也会存在这样一个值 count，用来表示需要等待的线程个数。

count 值是在 CountDownLatch 的构造函数中进行初始化的

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

Sync(int count) {
 //设置 AQS 中的 state 为 count 值
 setState(count);
}
计数值 count 是一次性的，当它的值减为0后就不会再变化了，这也是其存在的不足之处。

2、谁来确定乘客全部到齐？

在汽车发车前检票员会对车上的乘客数量进行清点，如果满员了就会通知司机开车。

当然也可以采用这种方法：在得知车座位数的前提下，每上来一位乘客，座位数进行减一操作。CountDownLatch 就是采用的上述方法，它的 countDown() 方法会对 state 的值执行减1操作。

让我们从源码的角度来认识一下该方法。

public void countDown() {
 //释放共享锁
 sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
 if (tryReleaseShared(arg)) {
  doReleaseShared();
  return true;
 }
 return false;
}
先尝试释放锁，如果返回 true，则执行释放操作，反之不执行。我们分析下上边的两个方法

protected boolean tryReleaseShared(int releases) {
 for (;;) {
  //获取当前等待的线程数量
  int c = getState();
  //等待线程数为0，表示没有等待线程，故不需要释放锁资源
  if (c == 0)
   return false;
  //执行减1操作
  int nextc = c-1;
  //自旋+CAS将state的属性值-1
  if (compareAndSetState(c, nextc))
   return nextc == 0;
 }
}

最后一步中，如果减一之后为0，则说明没有其它线程等待，需要执行释放锁操作，返回 true，反之不需要。

在开始分析 doReleaseShared() 之前，我们先来补全一下 AQS 中 waitStatus 的状态说明

初始化状态：0，表示当前节点在同步队列中，等待获取锁；
CANCELLED：1，表示当前节点取消获取锁；
SIGNAL：-1，表示后续节点等待当前节点唤醒；
CONDITION：-2，表示当前线程正在条件等待队列中；
PROPAGATE：-3，共享模式，前置节点唤醒后续节点后，唤醒操作无条件传播下去；

/**
 * 释放锁：唤醒后续节点
 */
private void doReleaseShared() {
 for (;;) {
  Node h = head;
  //不是null 且不为尾节点，因为尾节点没有后续节点需要唤醒了
  if (h != null && h != tail) {
   int ws = h.waitStatus;
   //只有状态为 -1 才可以唤醒后续节点
   if (ws == Node.SIGNAL) {
    //将waitStatus设置为0失败会继续循环
    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
     continue;
    unparkSuccessor(h);
   }
   //将waitStatus设置为PROPAGATE失败会继续循环
   else if (ws == 0 &&
      !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    continue;                
  }
  if (h == head)                   
   break;
 }
}

unparkSuccessor() 方法用于唤醒 AQS 中被挂起的线程，在ReentrantLock的原理中讲过了，此处不再赘述。

小结：

当线程使用 countDown() 方法时，其实是使用了 tryReleaseShared() 方法以 CAS 的操作来减少 state ,直至 state 为 0 ，进而释放锁资源，唤醒后续节点。

③谁来发车？

肯定是司机来发车呀，那我们的 CountDownLatch 是如何实现的呢？

CountDownLatch 中的 await() 方法，就是等待线程的总开关，当发现 state 的值为0时会释放所有的等待线程，发车了。

我们从源码角度来看下它是如何工作的

public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//如果线程中断了，直接抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
//如果小于0，代表 state 不为0，即还有任务未执行完毕，会执行获取共享锁的操作
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}

protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
我们来看看它到底是如何获取共享锁的

private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//将当前线程封装成node放到队尾
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
//state为0，表示此时等待线程全部执行完毕，r为1。
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
failed = false;
return;
}
}
//从当前node节点向前寻找有效节点，并保证有效节点的waitStatus状态为-1
if (shouldParkAfterFailedAcquire(p, node) &&
//挂起线程
parkAndCheckInterrupt())
//在拿锁的期间，如果被中断了，那么会抛出异常，取消拿锁
throw new InterruptedException();
}
} finally {
if (failed)
//将当前节点设置为失效节点，并挂到最近的有效节点后边，上文中有图解
cancelAcquire(node);
}
}
其中最重要的就是 setHeadAndPropagate() 方法

private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
//将当前node设置为head，并将node的线程置为空
setHead(node);
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
//释放锁：唤醒后续节点
doReleaseShared();
}
}
小结：当线程使用 await() 方法时会将当前线程封装成 node 加入AQS 队列中，如果发现 state 不为0，说明还有任务未执行完成，继续阻塞；如果 state 为0，会释放掉所有的等待线程，执行 await() 之后的数据。

CountDownLatch 是 Java 中一个常用的并发工具类，位于 java.util.concurrent 包中。它允许一个或多个线程等待，直到在其他线程中执行的一组操作完成。
以下是 CountDownLatch 的基本使用步骤：
1.创建 CountDownLatch：指定计数器的初始值。
2.线程等待：调用 await() 方法，等待计数器变为零。
3.计数器递减：在其他线程中完成操作后，调用 countDown() 方法，将计数器的值减一。
代码示例
下面是一个简单的示例，演示如何使用 CountDownLatch：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchExample {

    public static void main(String[] args) {
        // 创建一个 CountDownLatch 对象，计数器初始化为3
        CountDownLatch latch = new CountDownLatch(3);

        // 创建并启动三个工作线程
        for (int i = 0; i < 3; i++) {
            new Thread(new Worker(latch)).start();
        }

        // 主线程等待，直到计数器减到0
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("所有工作线程已完成。主线程继续执行。");
    }

    static class Worker implements Runnable {
        private CountDownLatch latch;

        Worker(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                // 模拟一些工作
                Thread.sleep((long) (Math.random() * 1000));
                System.out.println(Thread.currentThread().getName() + " 完成工作");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 工作完成后，计数器减一
                latch.countDown();
            }
        }
    }
}

解释

1.创建 CountDownLatch：

1	CountDownLatch latch = new CountDownLatch(3);

这里计数器初始化为3，表示有三个操作需要完成。

2.创建并启动工作线程：

1
2
3

for (int i = 0; i < 3; i++) {
    new Thread(new Worker(latch)).start();
}

启动三个线程，每个线程会执行一些模拟工作。

3.等待所有线程完成：

1	latch.await();

主线程调用 await() 方法，进入等待状态，直到计数器变为零。

4.工作线程完成工作后，计数器递减：

1	latch.countDown();

每个工作线程在完成其任务后，调用 countDown() 方法将计数器减一。

5.主线程继续执行：

当计数器变为零时，主线程从 await() 方法返回，继续执行后续操作。
通过这种方式，可以确保主线程等待所有工作线程完成其任务后再继续执行。

< 高并发系统设计

Redis 的大 Key 对持久化有什么影响？ >