JAVA里的锁之二独占锁与共享锁实现分析
【JAVA里的锁之二独占锁与共享锁实现分析】接着上一篇文章再来分析下同步是如何完成线程同步的，主要内容有：同步队列，独占式同步状态获取与释放，共享式同步状态获取与释放，超时获取同步状态。
1，同步器状态同步整体说明
同步器依赖内部的同步队列(FIFO双向队列)业完成同步状态的管理，当前线程获取同步状态失败时，同步器会将当前线程及等待状态等信息构造成一个节点Node并将其加入同步队列，同时阻塞当前线程，当同步状态释放时，会将节点中的线程唤醒，使其再次获取同步状态。
文章图片
同步器包含了两个节点类型的引用，一个指向头节点，另一个指向尾节点。获取同步状态时只会有一个线程能获取成功，因此设置头节点时不需要使用CAS，只需要将首节点设置成为原首节点的后继节点并断开原首节点的next引用。首节点的线程在释放同步状态时，会唤醒后继节点，后继节点在获取到同步状态后将自己设置为首节点。
文章图片
其它没获取到头节点的线程会被构造成节点加入到同步队列的尾部，因为有多个线程所以使用了CAS设置尾节点。
文章图片
2，同步器源码说明
2.1 Node节点类
同步队列中的节点(Node)用来保存获取同步状态失败的线程引用，等待状态及前驱和后继节点。
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED =1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL= -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
/**
* Returns true if node is waiting in shared mode.
*/
final boolean isShared() {
return nextWaiter == SHARED;
}/**
* Returns previous node, or throws NullPointerException if null.
* Use when predecessor cannot be null.The null check could
* be elided, but is present to help the VM.
*
* @return the predecessor of this node
*/
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}Node() {// Used to establish initial head or SHARED marker
}Node(Thread thread, Node mode) {// Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
waitStatus，等待状态，包含如下状态：
CANCELLED(1)：由于在同步队列中等待的线程超时或被中断需要从同步队列中取消等待，节点进入该状态将不会变化
SIGNAL(-1)：后继的线程处于等待状态，如果当前节点的线程释放了同步状态或者被取消，将会通知后继节点使后继节点的线程得以运行。
CONDITION(-2)：节点处于等待队列中，节点线程等待在Condition上，当其它线程对Condition调用了signal后，该节点将会从等待队列中转移到同步队列，加入到对同步状态的获取中。
PROPAGATE(-3)：表示下一次共享式同步状态获取将会无条件被传播下去。
INITIAL(0):初始状态。注意，负值表示结点处于有效等待状态，而正值表示结点已被取消。所以源码中很多地方用>0、<0来判断结点的状态是否正常。
Node prev，前驱节点，当节点加入同步队列时被设置（尾部添加）。
Node next，后继节点。
Node nextWaiter，等待队列中的后继节点，如果当前节点是共享的，那么这个字段是一个SHARED常量，也就是说节点类型（分独占和共享）和等待队列中的后继节点共用一个字段。
Thread thread，获取同步状态的线程。
3，独占锁及共享锁的实现
3.1.1 acquire(int) 此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源，线程直接返回，否则进入等待队列，直到获取到资源为止，且整个过程忽略中断的影响。这也正是lock()的语义，当然不仅仅只限于lock()。获取到资源后，线程就可以去执行其临界区代码了。下面是acquire()的源码：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
流程如下：
- tryAcquire()尝试直接去获取资源，如果成功则直接返回（&&为短路运算符，这里体现了非公平锁，每个线程获取锁时会尝试直接抢占加塞一次，而CLH队列中可能还有别的线程在等待）；如果失败则进一步执行acquireQueued()
- addWaiter()将该线程加入等待队列的尾部，并标记为独占模式。
- acquireQueued()使线程阻塞在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
- 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
AQS这里只定义了一个接口，具体资源的获取交由自定义同步器去实现了（通过state的get/set/CAS）。
private Node addWaiter(Node mode) {
//以给定模式构造结点。mode有两种：EXCLUSIVE（独占）和SHARED（共享）
Node node = new Node(Thread.currentThread(), mode);
// 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;
}
}
//CAS自旋加入队尾
enq(node);
return node;
}
//加入队尾
private Node enq(final Node node) {
//CAS"自旋"，直到成功加入队尾
for (;
;
) {
Node t = tail;
if (t == null) {// 队列为空，创建一个空的标志结点作为head结点，并将tail也指向它。
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {//放入队尾
t.next = node;
return t;
}
}
}
}
//获取同步状态
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
//标记是否成功拿到资源，true未获取；false已获取
try {
boolean interrupted = false;
//标记等待过程中是否被中断过
for (;
;
) {
final Node p = node.predecessor();
//前驱节点
//如果前驱节点是头节点，才有资格(可能是被头节点唤醒或被中断)去获取同步状态
if (p == head && tryAcquire(arg)) {
//将同步器节点指向当前节点，当前节点也就成为了头节点
setHead(node);
//将之前的头节点的next置为null，也就是断开与下一节点的连接
//这个操作是将之前的头节点从队列里移除了
p.next = null;
//HELP GC
failed = false;
//成功获取资源
return interrupted;
}
//如果前驱节点不是头节点且尝试获取同步状态失败，则说明需要park
//从而进入waiting状态直到被unpark()
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//如果等待过程中没有成功获取资源（如timeout，或者可中断的情况下被中断了），那么取消结点在队列中的等待。
if (failed)
cancelAcquire(node);
}
}
/*
*此方法主要用于检查状态，看看自己是否真的可以去休息了，
*以免队列前边的线程都放弃了而自己一直盲等。
*整个流程中，如果前驱结点的状态不是SIGNAL，那么自己就不能安心去休息，
*需要去找个安心的休息点，同时可以再尝试下看有没有机会轮到自己拿号
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//获取前驱节点状态
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
* 如果已经告诉前驱拿完号后通知自己一下，那就可以安心休息了
* 可以看下上面2.1 Node类里关于waitStatus的说明
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
* 前驱节点状态大于0说明前驱节点因为等待超时或被中断放弃了等待
* 如果前驱放弃了，那就一直往前找，直到找到最近一个正常等待的状态，
* 并排在它的后边。
* 注意：那些放弃的结点，由于被自己“加塞”到它们前边，
* 它们相当于形成一个无引用链，稍后就会被GC回收
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE.Indicate that we
* need a signal, but don't park yet.Caller will need to
* retry to make sure it cannot acquire before parking.
* 如果前驱正常，那就把前驱的状态设置成SIGNAL，告诉它拿完号后
* 通知自己一下。
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}/*
*如果线程找好安全休息点后，那就可以安心去休息了。此方法就是让线程去休息，
*真正进入等待状态。
*/
private final boolean parkAndCheckInterrupt() {
//park()会让当前线程进入waiting状态。在此状态下，有两种途径可以唤醒该线程：1）被unpark()；2）被interrupt()。
LockSupport.park(this);
//Thread.interrupted()会清除当前线程的中断标记位。
return Thread.interrupted();
}
这里总结下acquireQueued()方法的流程：
acquire()方法也做一个总结：
- 结点进入队尾后，检查状态，找到安全休息点；
- 调用park()进入waiting状态，等待unpark()或interrupt()唤醒自己；
- 被唤醒后，看自己是不是有资格能拿到号。如果拿到，head指向当前结点，并返回从入队到拿到号的整个过程中是否被中断过；如果没拿到，继续流程1。
- 调用自定义同步器的tryAcquire()尝试直接去获取资源，如果成功则直接返回；
- 没成功，则addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
- acquireQueued()使线程在等待队列中休息，有机会时（轮到自己，会被unpark()）会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
- 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。
文章图片
3.1.2 release(int) 该方法为释放同步状态，在释放了之后还会唤醒其后继节点，这样后继节点才有机会重新尝试获取同步状态。
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
//找头节点
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
//唤醒下一节点
return true;
}
return false;
}
/*
*跟tryAcquire()一样，这个方法是需要独占模式的自定义同步器去实现的
*/
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling.It is OK if this
* fails or if status is changed by waiting thread.
* ws为当前节点的状态，此时node还是头节点
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//将自身状态置为0/*
* Thread to unpark is held in successor, which is normally
* just the next node.But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
Node s = node.next;
//下一节点
//如果为空或等待的节点状态已无效（等待超时或被中断，需要从队列里取消等待）
if (s == null || s.waitStatus > 0) {
s = null;
//从后往前找，waitStatus小于0的等待是有效的
for (Node t = tail;
t != null && t != node;
t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
//将找到的合适节点唤醒
}
总结一下release(int)方法就是：
一句话概括，用unpark()唤醒等待队列中最前边的那个未放弃线程上面分析了独占锁的获取与释放，下面来看下共享锁的获取与释放。
文章图片
3.2.1 acquireShared(int) 共享式获取与独占式获取最主要区别在于同一时刻能否有多个线程同时获取同步状态。
public final void acquireShared(int arg) {
/*
*tryAcquireShared依然需要子类去实现，但返回值的含义已定义好：
*负值代表获取失败；
*0代表获取成功，但没有剩余资源；
*正数表示获取成功，还有剩余资源，其他线程还可以去获取
*/
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}private void doAcquireShared(int arg) {
//构造共享节点并加入到队尾
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;
;
) {
final Node p = node.predecessor();
//前驱节点
if (p == head) {
int r = tryAcquireShared(arg);
//尝试获取资源
if (r >= 0) {//获取成功
//将head指向自己，还有剩余资源可以再唤醒之后的线程
setHeadAndPropagate(node, r);
p.next = null;
// help GC
//如果等待过程中被打断过，此时将中断补上
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
//判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
//将自己设置为头节点，如果还有剩余资源还会唤醒下一临近节点
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
// Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
*Propagation was indicated by caller,
*or was recorded (as h.waitStatus either before
*or after setHead) by a previous operation
*(note: this uses sign-check of waitStatus because
*PROPAGATE status may transition to SIGNAL.)
* and
*The next node is waiting in shared mode,
*or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
* 如果还有剩余量，继续唤醒下一个邻居线程
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
与独占模式相比，共享式获取到资源后，如果有剩余资源还会去唤醒后继节点。
3.2.2 releaseShared() 如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//尝试释放资源，需要子类实现
doReleaseShared();
//唤醒后继节点
return true;
}
return false;
}private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases.This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;
;
) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// loop to recheck cases
unparkSuccessor(h);
//唤醒后继
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
// loop on failed CAS
}
if (h == head)// loop if head changed
break;
}
}
这里再提一下关于超时获取资源贴一下代码：
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
//截止时间
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.EXCLUSIVE);
//独占式
boolean failed = true;
try {
for (;
;
) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {//尝试获取
setHead(node);
p.next = null;
// help GC
failed = false;
return true;
}
//减去尝试获取所花时间
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)//时间到了或已超时，需要立即返回
return false;
//获取失败后需要等待，且需要等待的时间仍大于1s
//则等待nanosTimeout
//说明下，如果需等待时间已小于1s也进入等待的话，那经过下一轮的
//计算误差会很大，所以只有大于1s才进入等待。
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
好了，这里分析完了独占锁与共享的获取与释放，下一篇文章再分析下重入锁，读写锁及锁降级相关的知识点。
参考的文章：
Java并发之AQS详解
沉淀再出发：关于java中的AQS理解
推荐阅读
- 热闹中的孤独
- JAVA（抽象类与接口的区别&重载与重写&内存泄漏）
- 放屁有这三个特征的，请注意啦！这说明你的身体毒素太多
- 一个人的旅行，三亚
- 布丽吉特，人生绝对的赢家
- 慢慢的美丽
- 尽力
- 一个小故事，我的思考。
- 家乡的那条小河
- 《真与假的困惑》???|《真与假的困惑》??? ——致良知是一种伟大的力量