一、ReentrantLock
ReentrantLock是Java提供的并发同步组件，内部也是基于队列同步器AQS来实现的，它具有两个特点：
- 可重入：当一个线程调用
lock方法并成功获取到锁之后，如果该线程再次调用lock方法仍能获得该锁而不会被阻塞。 - 支持公平 & 非公平锁的选择：公平锁指的是等待时间最长的线程最优先获取锁，反之就是不公平的。
1.1 重进入
重进入需要解决两个问题：
- 再次获取锁：需要去识别获取锁的线程是否为当前占据锁的线程。
- 最终释放：线程重复获取了
n次锁，随后在第n次释放该锁后，其他线程能够获取到该锁。
ReentrantLock内部声明了两个内部类FairSync和NonfairSync，它们都继承于AbstractQueuedSynchronizer，分别对应公平锁和非公平锁的实现。我们以NonfairSync来看一下它是如何实现重进入的功能，为了便于理解这里只保留了和加锁相关的部分。
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {
abstract void lock();
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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;
}
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
}
static final class NonfairSync extends Sync {
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
public void lock() {
sync.lock();
}
public void unlock() {
sync.release(1);
}
}
lock的流程如下：
- 调用到
NonfairSync.lock()，判断当前是否已经有线程获取了同步状态，如果没有，那么直接设置当前线程。 - 如果没有获取成功，那么调用
acquire方法，这个是AQS提供的独占式获取同步状态方法，它会回调抽象的tryAcquire(int acquires)，参见 多线程知识梳理(11) - 队列同步器实现原理 & 应用，最终走到nonfairTryAcquire(acquires)这一步，开始正式的处理逻辑。这里面有下面三种情况：- 如果
getState() = 0，表示当前没有线程获取到同步状态，那么直接设置当前线程即可。 -
如果已经获取到同步状态的线程是当前线程，那么不会创建新的结点，仅仅是将同步状态值增加并返回
true，表示获取同步状态成功，这里就是实现可重入的关键。 - 其余情况返回
false。
- 如果
unlock的流程如下：
-
sync.release(int releases)会调用到tryRelease方法，并依赖该方法的返回值决定是否释放同步状态。 - 这里面的判断逻辑是，如果该锁被获取了
n次，那么前n-1次的tryRelease(int releases)必须返回false，而只有同步状态完全释放了，才能返回true。
1.2 公平锁与非公平锁
公平性是针对获取锁而言的，如果一个锁是公平的，那么锁的获取顺序应当符合请求的绝对时间顺序，也就是FIFO。
ReentrantLock提供了公平锁和非公平锁的选择，从实现上来说，就是组合的队列同步器Sync的类型是FairSync还是NonfairSync，非公平锁的实现我们之前已经分析过了，下面来看一下公平锁的实现。
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
//区别一：没有直接根据当前同步状态返回。
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//区别二：判断是否有前驱结点。
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
这里的区别就是在没有任何线程获取到同步状态的时候，加入了是否有前驱结点的判断，如果该函数返回true，那么表示有线程比当前线程更早地请求获取锁，因此需要等待前驱线程获取并释放锁后才能继续获取锁。
1.3 对比
- 公平锁保证了锁的获取按照
FIFO原则，而代价是进行大量的线程切换。 - 非公平锁虽然可能造成线程饥饿，但极少的线程切换，保证了其更大的吞吐量。
1.4 API 接口
由于ReentrantLock实现了Lock接口，因此其API的定义和Lock的定义相同：
-
void lock()：获取锁，调用该方法当前线程会获取锁，当锁获得后，从该方法返回。 -
void lockInterruptibly() throws InterruptedException：可中断地获取锁，和lock()区别在于该方法会响应中断，即在锁的获取过程中可以中断当前线程。 -
boolean tryLock()：尝试非阻塞地获取锁，调用该方法后立即返回，如果能够获取返回true，否则返回false。 -
boolean tryLock(long time, TimeUnit unit) throws InterruptedException：超时地获取锁，在以下三种情况下会返回：- 超时时间内获取了锁，
true - 超时时间内被中断，抛出异常，没有返回值
- 超时时间结束，
false
- 超时时间内获取了锁，
-
void unlock()：释放锁 -
Condition newCondition()：获取等待通知组件，该组件与当前的锁绑定，当前线程只有获取了锁，才能调用组件的await()方法，调用后，当前线程将释放锁。
这里比较容易混淆的是 响应中断 的概念，中断的情况分为两种：
- 调用
lock之前已经被中断：-
lock()不会抛出异常。 -
lockInterruptibly()会立刻抛出InterruptedException异常。
-
- 调用
lock之后，在同步队列中排队时被中断：-
lock()不会抛出异常，但是会被唤醒，将for(;;)循环中的interrupted设置为true。之后继续在同步队列中等待获取到同步状态，它和正常获取的区别就是在成功获取到同步状态返回后，Thread的interrupted标志位为true。 -
lockInterruptibly()会被唤醒，并且立即抛出InterruptedException异常。
-
//1.不响应中断的处理方式。
final boolean acquireQueued(final Node node, long arg) {
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null;
return interrupted;
}
//在同步队列中遇到中断会被唤醒。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
//仅仅是设置表示位。
interrupted = true;
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
//2.响应中断的处理方式。
private void doAcquireInterruptibly(long arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null;
return;
}
//在同步队列中遇到中断会被唤醒。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
//抛出异常。
throw new InterruptedException();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
网友评论