CountDownLatch源码解析
描述:
一种同步辅助工具,允许一个或多个线程等待在其他线程中执行的一组操作完成。
用给定的count
初始化 CountDownLatch 。因为调用了 countDown() 方法, await() 方法会一直阻塞,直到当前计数为零。在这之后,所有等待的线程都会被释放,后续对 await() 的调用都会立即返回。这是一种一次性现象 —— 计数无法重置。如果你需要一个重置计数的版本,可以考虑使用 CyclicBarrier 。
CountDownLatch 是一个通用的同步工具,可以用于许多目的。一个初始化为1的 CountDownLatch 相当于一个简单的 on/off闩锁,即门:调用 await() 的所有线程都会在门上等待,直到调用 countDown() 的线程打开门。初始化为N
的 CountDownLatch 可以用来使一个线程等待,直到 N
线程完成某个操作,或者某个操作完成N次。
CountDownLatch 有一个很有用的属性,它不要求调用 countDown 的线程等待计数为0后再继续,它不要求调用 countDown 的线程在继续之前等待计数为零,它只是阻止任何线程继续执行 await(),直到所有线程都可以通过。
示例用法:
下面是两个类,其中一组工作线程使用两个倒计时锁存器:
第一个是一个启动信号,阻止任何 worker 继续前进,直到 Driver 准备好让他们继续前进;
第二个是一个完成信号,它允许 Driver 等待直到所有工人都完成工作。
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 个部分,用一个 Runnable 对象来描述每个部分,该Runnable对象执行该部分,并在闩锁时计数,然后将所有的Runnable对象排队到一个执行器。当所有子部分完成时,协调线程将能够通过await。(当线程必须以这种方式重复计数时,请使用 CyclicBarrier。)
class Driver2 { // ...
void main() throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
Executor e = ...
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() { ... }
}
内存一致性影响:直到计数为零,在调用线程之前执行操作
源码:
public class CountDownLatch {
/**
* 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;
}
}
}
private final Sync sync;
/**
* 指定计数器的构造函数
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
/**
* 导致当前线程等待,直到闩锁计数到0,除非线程被中断。
*
* 1.如果当前计数为零,则此方法立即返回。
* 2.如果当前计数大于0,那么当前线程将被禁用,用于线程调度,并处于休眠状态,直到发生以下两种情况之一: 由于调用了 countDown 方法,计数器变为0 或 其他线程中断当前线程。
* 3.如果当前线程: 在进入此方法时设置其中断状态;或 当等待时,线程中断 , 然后 InterruptedException 被抛出,当前线程的中断状态被清除。
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* 除非线程是被中断,或者指定的等待时间已经过,否则当前线程将等待到锁存倒计时为零。
*
* 1.如果当前计数为零,则该方法立即返回值 true。
* 2.如果当前线程计数大于0,那么当前线程出于线程调度的目的被禁用,并处于休眠状态,直到发生以下三种情况之一: 由于调用了 countDown 方法,计数器变为0 或 其他线程中断当前线程。
* 3.如果当前线程: 在进入该方法时设置了它的中断状态;或 当等待时,线程中断 。然后 InterruptedException 被抛出,当前线程的中断状态被清除。
* 4.如果指定的等待时间过去了,则返回值 false 。如果时间小于或等于零,该方法将根本不等待。
*/
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
/**
* 减少锁存器的计数,如果计数为零,则释放所有等待线程。
*
* 1.如果当前计数值大于零,则减1。如果新的计数为0,则重新启用所有等待的线程,用于线程调度。
* 2.如果当前计数等于零,那么什么也不会发生。
*/
public void countDown() {
sync.releaseShared(1);
}
/**
* 返回当前计数。
*/
public long getCount() {
return sync.getCount();
}
public String toString() {
return super.toString() + "[Count = " + sync.getCount() + "]";
}
}
整体流程:
-
通过构造函数初始化 AQS 内部类的状态值 state
-
主线程调用 await 方法,当 state = 0 时,继续向下执行,否则进入阻塞队列,等待唤醒
-
子线程调用 countDown 方法,使 state-1 ,当减完之后的 state != 0 , 不做任何操作,否则是否所有等待的线程(释放主线程的阻塞)
设计思路:
主要根据 state 的值是否大于 0 ,
大于0主线程是否阻塞,当子线程逐步做state的减一操作,state = 0 时,释放主线程,主线程继续执行。以达到计数器到零之后再继续操作
小于等于0,主线程不阻塞,继续执行
标签:int,countDown,计数,源码,线程,doneSignal,CountDownLatch,解析 From: https://www.cnblogs.com/coolyang/p/17126155.html