(juc系列)CountDownLatch源码阅读

java►juc , , 评论

前言

本文源码基于: JDK13

为了巩固AQS.　看一下CountDownLatch的源码.

简介

大部分都是直接翻译的官方代码注释，嘻嘻

一个同步器, 允许一个或者多个线程等待, 知道其他线程完成一系列操作.

初始化时提供一个数字. await方法将阻塞，直到别的线程通过调用countDown,达到给定的数字.

这个类是一个一次性，count数字不能被重新设置. 如果你需要一个可复用的版本，可以考虑使用CyclicBarrier.

CountDownLatch 可以用于以下目的:

  1. 初始化为N. 所有线程调用await等待，直到门被一个线程调用countDown来打开．
  2. 初始化为N, 一个线程等待，直到N个线程完成了一些动作，或者某个动作被完成了N次.

CountDownLatch的一个很有用的特性是:

所有调用countDown的线程不需要等待计数到达0. 他只是在await方法上阻塞所有想要通过的线程.

使用实例(来自官方文档):

两个类，使用两个CountDownLatch来完成以下功能.

  1. 第一个CountDownLatch, 是一个开始信号，告诉所有工作线程，驱动已经就绪，可以开始工作了。
  2. 第二个CountDownLatch, 是一个结束信号，允许驱动等待所有工作线程完成，之后进行其他工作.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class Driver { 
  
  void main() throws InterruptedException {
    CountDownLatch startSignal = new CountDownLatch(1);
    CountDownLatch doneSignal = new CountDownLatch(N);
 
    for (int i = 0; i < N; ++i) // create and start threads
      new Thread(new Worker(startSignal, doneSignal)).start();
 
    doSomethingElse();            // don't let run yet
    startSignal.countDown();      // let all threads proceed
    doSomethingElse();
    doneSignal.await();           // wait for all to finish
  }
}

class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;

    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    public void run() {
        try {
            startSignal.await();
            doWork();
            doneSignal.countDown();
        } catch (InterruptedException ex) {
        } // return;
    }

    void doWork() {

    }
}

另外一个典型应用是，将一个任务分割成N部分，每一个部分封装成一个任务，交给线程池。
然后一个协调线程，调用`await｀等到所有的子部分完成. 再通过.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class Driver2 { 
  void main() throws InterruptedException {
    CountDownLatch doneSignal = new CountDownLatch(N);
    Executor e = null; // some Executor

    for (int i = 0; i < N; ++i) // create and start threads
      e.execute(new WorkerRunnable(doneSignal, i));

    doneSignal.await();           // wait for all to finish
  }
}

class WorkerRunnable implements Runnable {
  private final CountDownLatch doneSignal;
  private final int i;
  WorkerRunnable(CountDownLatch doneSignal, int i) {
    this.doneSignal = doneSignal;
    this.i = i;
  }
  public void run() {
    try {
      doWork(i);
      doneSignal.countDown();
    } catch (InterruptedException ex) {} // return;
  }

  void doWork() {}
}

源码探究

最核心的实现，依然是继承自AQS的一个子类同步器Sync.

Sync

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
/**
 * Synchronization control For CountDownLatch.
 * Uses AQS state to represent count.
 */
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;
        }
    }
}

首先，初始化时传递的Count值，复用AQS中的状态State.

实现了AQS的共享模式加锁及共享模式解锁.

tryAcquireShared(int acquires)

共享模式的加锁，锁空闲就返回1. 锁非空闲就返回-1.

tryReleaseShared(int releases)

共享模式的解锁.

1
2
3
4
5
6
7
8
9
10
11
12
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;
    }
}

递减Count. 如果减1之后为0，就认为解锁成功. 通知.

如果减去1之后不为0. 返回false. 意味着解锁了，但是没有完全解锁成功.

构造方法

没啥说的，　将传入的Count值传入Sync.复用AQS的状态值来实现Count的控制.

countDown()

调用Sync同步器的释放共享锁方法，进行一次解锁操作.

await()

调用Sync同步器的获取共享锁方法，进行加锁操作.

总结

CountDownLatch是对AQS的共享模式的比较精巧的应用.

  1. 首先初始化时传入Count值. 设置AQS的State值.
  2. 由于自定义了共享锁的获取逻辑，当State值>0时，此锁不可再被获取.
  3. countDown操作，即释放一次锁操作. 每次释放State值减1.
  4. await操作，即加锁操作. 阻塞式加锁，在初始化之后，加锁会一直阻塞，直到调用N次的countDown之后，将锁完全释放. 此时获取锁成功，继续下一步，也就是await方法成功返回，成功通过CountDownLatch了.

为什么CountDownLatch是一次性的?

CountDownLatch中的同步器实现，并不是传统意义上的可以不断加锁或解锁。

只有在初始化时进行了设置State的操作，之后只可以进行读取/递减.

他的加锁操作，不会设置State的值，只是判断State是否大于1.

当解锁完成，State为0. 此时没有渠道去进行更新State的值.

如果重复的调用加锁，会不断的拿到”加锁成功”. 但是State数值并不会改变.

因此此时的”加锁成功”,其实意味着”门已经打开，可以无限进入”. 每一次的”加锁操作”, 约等于判断”门是否开着”的操作.

完.


完。





联系我

最后，欢迎关注我的个人公众号【 呼延十 】，会不定期更新很多后端工程师的学习笔记。
也欢迎直接公众号私信或者邮箱联系我，一定知无不言，言无不尽。



以上皆为个人所思所得，如有错误欢迎评论区指正。

欢迎转载，烦请署名并保留原文链接。

联系邮箱：huyanshi2580@gmail.com

更多学习笔记见个人博客或关注微信公众号 <呼延十 >——>呼延十

呼延十Developer

个人简介。