image
特性
- 显示锁，自旋锁，可重入锁
- aqs队列锁的实现
- 支持多条件唤醒
- 支持打断
- 支持公平,非公平锁
- 可尝试加锁
使用方式
ReentrantLock lock = new ReentrantLock();
try {
// 加锁
lock.lock();
// do something
} finally {
// 在finally中解锁,避免出现异常导致 没释放锁而死锁
lock.unlock();
}
源码
1. lock() 公平锁
final void lock() {
acquire(1);
}
AbstractQueuedSynchronizer. acquire(1)
public final void acquire(int arg) {
// !tryAcquire(arg) 尝试加锁
// acquireQueued(addWaiter(Node.EXCLUSIVE) 加入队列
// selfInterrupt 自我打断
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire(int acquires)
protected final boolean tryAcquire(int acquires) {
// 获取当前线程
final Thread current = Thread.currentThread();
// 判断当前锁的状态, 0 代表没有人持有
int c = getState();
if (c == 0) {
// 判断队列中是否有线程在等待,如果没有则cas改变锁的状态
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// cas成功,设置当前线程为持有锁的线程
setExclusiveOwnerThread(current);
return true;
}
}
// 有线程占有锁,判断持有线程是否有是自身,是的话重入
else if (current == getExclusiveOwnerThread()) {
// 锁状态 + 1
int nextc = c + acquires;
// 锁状态 < 0 ? 加锁次数太多,导致整形溢出？
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
// 设置锁的状态
setState(nextc);
return true;
}
// 抢锁失败
return false;
}
hasQueuedPredecessors()
判断队列中是否有线程在排队等锁
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
// 头结点一般都是 空结点
// 头结点 != 尾结点
// 头结点下的next为空 或者 头结点的后一个结点的线程不是当前线程
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
acquireQueued(final Node node, int arg) ： 入队睡眠
如果当前锁被其他线程持有(非自身重入), 或者队列中有线程在排队,则加锁失败，需要创建Node节点,入队(如果队列还没有初始化则会初始化队列)，并且入队之后,会判断自己Node的前驱节点是不是head,是的话，会自旋拿一次锁。
自旋拿锁成功会把自己做为新的头节点,空Node
addWaiter(Node mode) : 创建Node节点
// 创建线程为当前线程的Node对象
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节点的前节点设置为当前节点
node.prev = pred;
// 将新创建的Node节点设置为新的尾节点
if (compareAndSetTail(pred, node)) {
// 之前的尾节点的后驱节点设置为 新创建的Node对象
pred.next = node;
// 返回
return node;
}
}
// 如果尾节点为空,初始化队列
enq(node);
return node;
enq(Node mode) : 入队
如果队列还没有初始化,就会新建队列,队列的头结点和尾节点都为空的Node
将当前线程的Node节点接在之前的尾节点的后面，并设置成新的尾节点
private Node enq(final Node node) {
for (;;) {
// 尾节点
Node t = tail;
// 第一次进来,如果为空,说明队列还没有初始化
// 第二次进来就不为空了
if (t == null) { // Must initialize
// 第一次进来,创建空的Node,将头节点和尾节点都设置成这个空的Node
if (compareAndSetHead(new Node()))
tail = head;
} else {
// 不为空走这里,传进来的Node的前驱设置成之前的尾节点
node.prev = t;
// 传进来的Node成为新的尾节点,
if (compareAndSetTail(t, node)) {
// 之前的尾节点的后驱节点设置为传进来的Node
t.next = node;
return t;
}
}
}
}
acquireQueued ：自旋,失败后睡眠
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 拿前驱节点
final Node p = node.predecessor();
// 判断是不是头节点,是的自旋一次
if (p == head && tryAcquire(arg)) {
// 自己会变成空节点,并且成为头节点
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 设置前驱节点为需唤醒状态,第二次循环的时候,睡眠
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
shouldParkAfterFailedAcquire（Node pred, Node node）
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;
更改前驱节点的状态 ：
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// 判断前驱Node,是不是唤醒状态，是的话直接返回true,让当前Node去睡眠
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
// 如果是> 0，说明是取消状态
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
// 一直往前查找,知道找到非取消的状态的节点，作为当前Node的前驱
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.
*/
// 如果是0, cas将前驱Node的状态改为Node.SIGNAL, 外面第二次循环,直接返回true,让当前节点睡眠
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
parkAndCheckInterrupt()
阻塞当前线程,直到被打断,打断后会清空打断标记,在下一次循环中继续进来阻塞。造成一种无法打断的假象。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
加锁流程
-
当锁为无锁状态时 ：
image
-
线程t1来加锁 ： 持有锁的线程改成t1,锁的状态改为1:被占有状态
image
-
线程t1没有释放锁,线程t2来抢锁, t2会创建队列,并且排队
image
- 线程t1没有释放锁,线程t2入队,线程t3来抢锁
image
lock() : 非公平锁
非公平锁的加锁流程与公平锁不同的是,非公平锁在加锁的时候,不会判断队列中是否有线程在排队,直接cas加锁
final void lock() {
// cas改锁的状态
if (compareAndSetState(0, 1))
// 修改成功直接抢占锁,当锁的线程改为当前线程
setExclusiveOwnerThread(Thread.currentThread());
else
// 尝试加锁
acquire(1);
}
tryAcquire() & nonfairTryAcquire()
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 公平锁这里会判断队列是否有人在排序，非公平锁直接cas上手
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
再后面抢不到锁，进入队列之后和公平锁的逻辑是一模一样的。
公平和非公平之前在于 入队之前尝试加锁，看不看队列中有无线程排队，如果加锁失败,入队了,那么就再没有公平和非公平之分
2. unlock()
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
// 解锁, state一直减到0,锁才会被完全释放
if (tryRelease(arg)) {
Node h = head;
// 当锁完全释放后
if (h != null && h.waitStatus != 0)
// 叫醒后面排队的线程
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease
释放锁，state - 1,减到0,锁才会被完全释放,当前持有锁线程设置为null
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
// 状态为0,锁才完全释放, 重入时,state会被累加到 > 1,需要解锁多次
if (c == 0) {
// 完全释放
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
unparkSuccessor
:唤醒队列自己后面最近的一个非取消状态的线程
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.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 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;
// 则从尾节点向前遍历,找到自己后面最近的非取消状态
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
// 进行唤醒
if (s != null)
LockSupport.unpark(s.thread);
}
解锁流程
image
image
3.lockInterruptibly()
和lock方法不同的是,该加锁方法会响应打断异常
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
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;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// 如果再睡眠过程中被打断,会抛出异常
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
4.Condition 条件唤醒
await() : 等待
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 创建condition等待队列
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// 睡眠
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
addConditionWaiter()
如果尾节点为空，新建一个Node节点(持有当前线程),state为-2, 头节点和尾节点都等于新建的Node节点
如果尾节点不为空，新建一个Node节点(持有当前线程),接在尾节点的后面,并作为新的尾节点
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
signal() : 唤醒
唤醒条件队列的第一个Node
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 头节点
Node first = firstWaiter;
if (first != null)
// 唤醒
doSignal(first);
}
doSignal
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
transferForSignal(Node node)
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
// 将当前节点从条件队列加入到 等待队列中
// 返回是上一个节点
Node p = enq(node);
int ws = p.waitStatus;
// 然后将上一个节点的状态改为SIGNAL -1
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
// 如果cas失败，unpark
LockSupport.unpark(node.thread);
return true;
}
doSignalAll : 唤醒所有
将条件队列中所有的Node按顺序 从头到尾加到等待队列中
public final void signalAll() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignalAll(first);
}
private void doSignalAll(Node first) {
lastWaiter = firstWaiter = null;
// 无限循环，直到条件队列中没有元素
do {
Node next = first.nextWaiter;
first.nextWaiter = null;
// 从头到尾按顺序,加到等待队列中
transferForSignal(first);
first = next;
} while (first != null);
}
网友评论