什么是闭锁
闭锁是一种java的工具类，闭锁可以用来确保某些活动知道其他活动都完成后才继续执行后续操作。比如一场电脑游戏，需要等待所有玩家的初始化工作都完成之后才能开始游戏。有些场景下则需要两个闭锁，比如一场跑步比赛，所有的运动远需要同时从起点出发，并且需要等所有的运动员都跑过终点才能公布比赛结果。CountDownLatch是一种灵活的闭锁实现，闭锁状态包含一个正数的计数器，表示需要等待的事件的数量。countDown方法递减计数器，await方法起到阻塞作用，await方法会一直阻塞直到计数器为零，或者等待中的线程中断，或者等待超时。
import java.util.concurrent.CountDownLatch;
public class RunningContest {
public void run(int i){
System.out.println("运动" + i + "到达比赛终点！");
}
public void end(){
System.out.println("公布比赛成绩！");
}
public void start(int n, RunningContest runningContest) throws InterruptedException{
final CountDownLatch startContest = new CountDownLatch(1);
final CountDownLatch endContest = new CountDownLatch(n);
for(int i = 1; i <= n; i++){
final int k = i;
new Thread(){
public void run(){
try{
startContest.await();
try{
runningContest.run(k);
}finally {
endContest.countDown();
}
}catch (InterruptedException e){}
}
}.start();
}
// 比赛开始
startContest.countDown();
// 等待所有运动员通过终点
endContest.await();
// 公布比赛结果
end();
}
public static void main(String[] args) throws InterruptedException {
RunningContest runningContest = new RunningContest();
runningContest.start(5, runningContest);
}
}
运动4到达比赛终点！
运动3到达比赛终点！
运动2到达比赛终点！
运动1到达比赛终点！
运动5到达比赛终点！
公不比赛成绩！
startContest用于控制所有运动员在起跑线等待，startContest.countDown()相当于比赛开始的哨声。endContest.await()等待所有远动员都通过终点才能进行之后公布比赛成绩的操作。
闭锁源码解析
再了解闭锁原理之前，需要先了解一下AQS的基本原理，因为闭锁的也是基于AQS实现的同步类，闭锁的实现原理相对于读写锁来说算是比较简单的了，闭锁的基本逻辑就是请求共享锁，如果状态值state大于0，则请求失败，线程被加入阻塞队列。如果状态值state==0，则请求成功，线程不会被阻塞并可以继续执行下面的代码。下面我们先来看下闭锁的基本参数。
private final Sync sync;
// Sync核心类，闭锁的基本业务逻辑都在里面实现，继承自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;
}
}
}
// 构造函数，初始化Sync类并设置AQS的state参数初值
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
// 阻塞方法，如果state的值大于0则阻塞线程，等于0
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
// 将state状态减1，如果减1操作后，状态值state的值等于0，则唤醒阻塞队列中的线程。
public void countDown() {
sync.releaseShared(1);
}
CountDownLatch初始化源码
CountDownLatch countDownLatch = new CountDownLatch(1);
/*CountDownLatch#CountDownLatch(int count)*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
/*CountDownLatch$Sync#Sync(int count)*/
Sync(int count) {
setState(count);
}
/*AbstractQueuedSynchronizer#setState(int newState)*/
protected final void setState(int newState) {
state = newState;
}
从上面的代码我们可以看到，最终还是调用AQS中的方法将state状态设置为1。
线程请求共享锁失败的过程
我们先将状态值state设置为1。这样thread1线程就会因为请求共享锁失败而加入阻塞队列，加入阻塞队列后，thread1线程不会立即被挂起，而是在尝试获取共享锁两次失败后被挂起。我们看下在thread1调用await()方法后的执行流程。
CountDownLatch countDownLatch = new CountDownLatch(1);
Thread thread1 = new Thread(new Runnable() {
@Override
public void run() {
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "Thread1");
thread1.start();
/*CountDownLatch#await()*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);// 步骤1
}
/*AbstractQueuedSynchronizer#acquireSharedInterruptibly(int arg)*/
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)// 步骤2：请求获取共享锁。
// 如果请求失败则线程Thread1加入阻塞队列
doAcquireSharedInterruptibly(arg);// 步骤3：将线程Thread1加入阻塞队列。
}
/*CountDownLatch$Sync#tryAcquireShared(int acquires)*/
protected int tryAcquireShared(int acquires) {// 步骤2：请求获取共享锁
// 如果状态值state的值等于0，则返回1，共享锁获取成果。如果不等于0（大于0），则共享锁获取失败，线程Thread1被加入阻塞队列。
return (getState() == 0) ? 1 : -1;
}
/*AbstractQueuedSynchronizer#doAcquireSharedInterruptibly(int arg)*/
private void doAcquireSharedInterruptibly(int arg)// 步骤3：将线程Thread1加入阻塞队列。
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);// 步骤4：创建线程Thread1对应的阻塞节点并加入阻塞队列，Node.SHARED表示当前节点的类型为共享节点，即Thread1线程请求的是共享锁。
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
// 如果当前阻塞节点的前一个节点是头节点，说明当前阻塞节点为队列中第一个阻塞的线程，可以尝试获取贡献锁
if (p == head) {
// 尝试获取共享锁
int r = tryAcquireShared(arg);
// 如果获取成功，则将当前节点设置为头节点。
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// shouldParkAfterFailedAcquire(p, node) 会判断当前阻塞节点的前一个阻塞节点的waitStatus（0）是否为-1，如果不为-1，则会将当前阻塞节点的前一个阻塞节点的waitStatus状态从0设为-1并返回false，再执行一遍for循环后，线程获取共享锁还是失败的话，就会执行parkAndCheckInterrupt()方法挂起线程。
if (shouldParkAfterFailedAcquire(p, node) &&// 步骤5：shouldParkAfterFailedAcquire(p, node)
parkAndCheckInterrupt())// 步骤6：挂起线程Thread1
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
/*AbstractQueuedSynchronizer#addWaiter(Node mode)*/
private Node addWaiter(Node mode) {// 步骤4：创建线程Thread1对应的阻塞节点并加入阻塞队列
Node node = new Node(Thread.currentThread(), mode);// mode = Node.SHARED
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
/*AbstractQueuedSynchronizer$Node#Node(Thread thread, Node mode)*/
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;//
this.thread = thread;
}
下图为执行步骤4：addWaiter(Node.SHARED)方法后阻塞队列的状态。
image
下图为调用shouldParkAfterFailedAcquire(p, node)方法后的阻塞队列状态。
image
阻塞队列中的线程被唤醒的过程
CountDownLatch countDownLatch = new CountDownLatch(1);
Thread thread1 = new Thread(new Runnable() {
@Override
public void run() {
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "Thread1");
thread1.start();
countDownLatch.countDown();// 调用countDown唤醒阻塞队列中的线程
/*CountDownLatch#countDown()*/
public void countDown() {
sync.releaseShared(1);
}
/*AbstractQueuedSynchronizer#releaseShared(int arg)*/
public final boolean releaseShared(int arg) {
// 如果解锁成功，则继续唤醒阻塞队列中的线程。
if (tryReleaseShared(arg)) {// 步骤1：将状态值state减1
doReleaseShared();// 步骤2：唤醒阻塞队列中的线程。
return true;
}
return false;
}
/*CountDownLatch#tryReleaseShared(int releases)*/
protected boolean tryReleaseShared(int releases) {// 步骤1：将状态值state减1
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
// 利用CAS将原来的状态值state减1
if (compareAndSetState(c, nextc))
// 状态值state减1后如果等于0，则解锁成功，返回true
return nextc == 0;
}
}
/*AbstractQueuedSynchronizer#doReleaseShared()*/
private void doReleaseShared() {// 步骤2：唤醒阻塞队列中的线程。
for (;;) {
Node h = head;
// 如果头节点不为空（阻塞队列不为空）
if (h != null && h != tail) {
// 获取头节点状态值
int ws = h.waitStatus;
// 如果头节点状态值等于-1
if (ws == Node.SIGNAL) {
// 利用CAS将头节点的状态值WaitStatus设为0
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 唤醒阻塞队列的第一个阻塞线程
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 第一个阻塞线程（Thread1）被唤醒后会将阻塞队列的head指针指向自己，这样h就不在等于head节点，就可以继续执行for循环，唤醒下一个阻塞的线程。
if (h == head) // loop if head changed
break;
}
}
总结
以上就是闭锁的底层原理，所到底就是对应AQS状态值state的操作已经对阻塞队列的操作。
