Java ReentrantLock
会飞的汤姆猫 人气:0ReentrantLock 原理
概念
基于AQS实现的可重入锁实现类。
核心变量和构造器
public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
public ReentrantLock() {
// 默认为非公平锁。为何默认为非公平锁？因为通过大量测试下来，发现非公平锁的性能优于公平锁
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
// 由fair变量来表明选择锁类型
sync = fair ? new FairSync() : new NonfairSync();
}
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()) {
// 利用state整形变量进行次数记录
int nextc = c + acquires;
// 如果超过了int表示范围，表明符号溢出，所以抛出异常0111 1111 + 1 = 1000 0000
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
// 返回false 表明需要AQS来将当前线程放入阻塞队列，然后进行阻塞操作等待唤醒获取锁
return false;
}
// 公平锁和非公平锁公用方法，因为在释放锁的时候，并不区分是否公平
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
// 如果当前线程不是上锁的那个线程
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
// 不是重入锁，那么当前线程一定是释放锁了，然后我们把当前AQS用于保存当前锁对象的变量ExclusiveOwnerThread设置为null，表明释放锁成功
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
// 注意：此时state全局变量没有改变，也就意味着在setState之前，没有别的线程能够获取锁，这时保证了以上的操作原子性
setState(c);
// 告诉AQS，我当前释放锁成功了，你可以去唤醒正在等待锁的线程了
return free;
}
protected final boolean isHeldExclusively() {
return getExclusiveOwnerThread() == Thread.currentThread();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
}
static final class NonfairSync extends Sync {
// 由ReentrantLock调用获取锁
final void lock() {
// 非公平锁，直接抢锁，不管有没有线程排队
if (compareAndSetState(0, 1))
// 上锁成功，那么标识当前线程为获取锁的线程
setExclusiveOwnerThread(Thread.currentThread());
else
// 抢锁失败，进入AQS的标准获取锁流程
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
// 使用父类提供的获取非公平锁的方法来获取锁
return nonfairTryAcquire(acquires);
}
}
static final class FairSync extends Sync {
// 由ReentrantLock调用
final void lock() {
// 没有尝试抢锁，直接进入AQS标准获取锁流程
acquire(1);
}
// AQS调用，子类自己实现获取锁的流程
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
// 此时有可能正好获取锁的线程释放了锁，也有可能本身就没有线程获取锁
if (c == 0) {
// 注意：这里和非公平锁的区别在于：hasQueuedPredecessors看看队列中是否有线程正在排队，没有的话再通过CAS抢锁
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;
}
// 返回false 表明需要AQS来将当前线程放入阻塞队列，然后进行阻塞操作等待唤醒获取锁
return false;
}
}
}核心方法
获取锁操作：
public void lock() {
// 直接通过sync同步器上锁
sync.lock();
}释放锁操作：
public void unlock() {
sync.release(1);
}ReentrantReadWriteLock 原理
用例
将原来的锁，分割为两把锁：读锁、写锁。适用于读多写少的场景，读锁可以并发，写锁与其他锁互斥。写写互斥、写读互斥、读读兼容。
public class ThreadDemo {
static volatile int a;
public static void readA() {
System.out.println(a);
}
public static void writeA() {
a++;
}
public static void main(String[] args) {
ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
Thread readThread1 = new Thread(() -> {
readLock.lock();
try {
readA();
} finally {
readLock.unlock();
}
});
Thread readThread2 = new Thread(() -> {
readLock.lock();
try {
readA();
} finally {
readLock.unlock();
}
});
Thread writeThread = new Thread(() -> {
writeLock.lock();
try {
writeA();
} finally {
writeLock.unlock();
}
});
readThread1.start();
readThread2.start();
writeThread.start();
}
}核心变量和构造器
该接口用于获取读锁和写锁对象
public interface ReadWriteLock {
// 用于获取读锁
Lock readLock();
// 用于获取写锁
Lock writeLock();
}readerLock和writerLock变量用于支撑以上描述的ReadWriteLock接口的读锁和写锁方法。通过构造方法得知，读写锁对象的创建和用例均依赖于公平锁或者非公平锁同步器。
public class ReentrantReadWriteLock implements ReadWriteLock {
// 读锁对象
private final ReentrantReadWriteLock.ReadLock readerLock;
// 写锁对象
private final ReentrantReadWriteLock.WriteLock writerLock;
// 同步器
final Sync sync;
// 默认构造器，创建了非公平锁
public ReentrantReadWriteLock() {
this(false);
}
// 根据fair变量，来选择创建不同的锁：公平锁 FairSync 和非公平锁 NonfairSync
public ReentrantReadWriteLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
// 用同步器来创建读写锁对象
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
public ReentrantReadWriteLock.ReadLock readLock() { return readerLock; }
}Sync类
核心变量和构造器
我们说读锁可以多个线程同时持有，而写锁只允许一个线程持有，此时我们称 读锁-----共享锁 写锁------互斥锁（排他锁）。然后我们在AQS中了解到一个变量state，它是32位的值，那么我们这里将其切割为高16位和低16位。
abstract static class Sync extends AbstractQueuedSynchronizer {
// 高16位用于表示读锁
static final int SHARED_SHIFT = 16;
// 用于对高16位操作：加1 减1
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
// 最大读锁量
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
// 用于获取低16位的值。例如 获取低八位：0000 0000 1111 1111
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
/** 获取当前持有读锁的线程数量 */
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
/** 获取当前持有写锁的线程数量 */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
// 高16位为所有读锁获取，那么我想知道每个线程对于读锁重入的次数？采用ThreadLocal来进行统计，每个线程自己统计自己的
static final class HoldCounter {
int count = 0;
final long tid = getThreadId(Thread.currentThread());
}
// 继承自ThreadLocal，重写了其中的initialValue方法，该方法将在线程第一次获取该变量时调用初始化HoldCounter计数器
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
// 创建ThreadLocal对象
private transient ThreadLocalHoldCounter readHolds;
// 缓存最后一个线程获取的读锁数量
private transient HoldCounter cachedHoldCounter;
// 保存获取到该锁的第一个读锁线程
private transient Thread firstReader = null;
// 保存第一个该锁的第一个读锁线程获取到的读锁数量
private transient int firstReaderHoldCount;
Sync() {
// 构造器中初始化ThreadLocalHoldCounter ThreadLocal对象
readHolds = new ThreadLocalHoldCounter();
// 用于保证可见性，使用了state变量的volatile语义
setState(getState());
}
}tryAcquire获取写锁的流程
由AQS调用，用于子类实现自己的上锁逻辑，和原有获取互斥锁保持一致，
protected final boolean tryAcquire(int acquires) {
// 获取当前线程
Thread current = Thread.currentThread();
// 获取当前状态值和互斥锁的数量
int c = getState();
int w = exclusiveCount(c);
// 状态值有效
if (c != 0) {
// 有线程获取到了读锁或者当前线程不是持有互斥锁的线程
if (w == 0 || // 有线程获取到了读锁
current != getExclusiveOwnerThread()) // 有线程获取到了写锁
// 返回false 让AQS执行阻塞操作
return false;
// 写锁重入，而又由于写锁的数量保存在低16位，所以直接加就行了
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
setState(c + acquires);
return true;
}
// 既没有读锁，也没有写锁
if (writerShouldBlock() || // 由子类实现判断当前线程是否应该获取写锁
!compareAndSetState(c, c + acquires)) // 通过CAS抢写锁
return false;
// 获取写锁成功，那么将当前线程标识为获取互斥锁的线程对象
setExclusiveOwnerThread(current);
return true;
}tryAcquireShared获取读锁的流程获取写锁的流程
protected final int tryAcquireShared(int unused) {
// 获取到当前线程对象
Thread current = Thread.currentThread();
// 获取到当前状态值
int c = getState();
if (exclusiveCount(c) != 0 && // 有没有线程持有写锁
getExclusiveOwnerThread() != current) // 如果有线程获取到了互斥锁，那么进一步看看是不是当前线程
// 不是当前线程，那么直接返回-1，告诉AQS获取共享锁失败
return -1;
// 获取到读锁的持有数量
int r = sharedCount(c);
if (!readerShouldBlock() && // 让子类来判定当前获取读锁的线程是否应该被阻塞
r < MAX_COUNT && // 判断是否发生了溢出
compareAndSetState(c, c + SHARED_UNIT)) { // 直接CAS 增加state的高16位的读锁持有数量
// 增加高16位之前的计数为0，此时表明当前线程就是第一个获取读锁的线程
if (r == 0) {
// 注意：持有两个变量来优化threadlocal
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
// 当前获取读锁的线程就是一个线程，那么此时表明：锁重入，直接++计数位即可
firstReaderHoldCount++;
} else {
// 当前线程不是第一个读线程，此时将其获取读锁的次数保存在ThreadLocal中
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return 1;
}
// 有很多同学走到这里，直接懵逼？不知道这是啥情况？经验：在看doug lea写的代码时，请注意：经常做优化，就是把一些常见的场景前置，保证性能
return fullTryAcquireShared(current);
}fullTryAcquireShared完全获取读锁流程
final int fullTryAcquireShared(Thread current) {
HoldCounter rh = null;
for (;;) {
int c = getState();
// 当前已经有线程获取到写锁且当前获取写锁的线程不是，当前线程
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
} else if (readerShouldBlock()) {
// 子类判断当前线程应该阻塞
if (firstReader == current) {
// 当前线程就是第一个获取到读锁的线程
} else {
// 获取到当前线程记录读锁重入次数的HoldCounter对象
if (rh == null) {
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current)) {
rh = readHolds.get();
if (rh.count == 0)
readHolds.remove();
}
}
// 当前读锁重入次数为0时，表明没有获取读锁，此时返回-1，阻塞当前线程
if (rh.count == 0)
return -1;
}
}
// 读锁获取次数溢出
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// CAS增加读锁次数
if (compareAndSetState(c, c + SHARED_UNIT)) {
if (sharedCount(c) == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
if (rh == null)
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh;
}
return 1;
}
}
}tryRelease释放写锁的流程
protected final boolean tryRelease(int releases) {
// 没有获取写锁，为啥能释放写锁呢？
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
// 释放完毕后，写锁状态是否为0（锁重入），因为此时计算的不是当前state，是nextc
boolean free = exclusiveCount(nextc) == 0;
// 如果下一个状态值为0，此时表明当前线程完全释放了锁，也即锁重入为0，那么将当前线程对象从OwnerThread中移除
if (free)
setExclusiveOwnerThread(null);
// 此时设置全局state变量即可
setState(nextc);
// 如果返回为true，那么由AQS完成后面线程的唤醒
return free;
}tryReleaseShared释放读锁的流程
释放时，需要考虑：重入多少次，就释放多少次。总结：先完成自己的释放，然后再完成共享的高16位的释放。
protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
// 当前线程是第一个获取到读锁的线程
if (firstReader == current) {
// 当前重入次数为1，代表什么？代表可以直接释放，如果不是1，那么表明还持有多个读锁，也即重入多次，那么直接--
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
int count = rh.count;
if (count <= 1) {
// 当前线程已经释放完读锁，那么不需要在ThreadLocal里持有HoldCounter对象
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
--rh.count;
}
for (;;) {
// CAS释放高16位计数
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
// 释放完毕后是否为0，为无锁状态，此时需要干啥？由AQS来唤醒阻塞的线程
return nextc == 0;
}
}readerShouldBlock和writerShouldBlock模板方法公平锁实现
判断条件只有一个：hasQueuedPredecessors()方法，就是看看AQS的阻塞队列里是否有其他线程正在等待，如果有排队去。
总结：有人在排队，那么不插队。w->r->r->r 此时来了个r：w->r->r->r->r， 此时来了个w：w->r->r->r->w。
static final class FairSync extends Sync {
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
} // w->r->r r获取锁 w->r->r-r
}readerShouldBlock和writerShouldBlock模板方法非公平锁实现
写线程永远false，因为读写锁本身适用的是读多写少，此时不应该 让写线程饥饿，而且非公平，写锁永远不阻塞，让它抢，不管前面是否有人排队，先抢了再说。apparentlyFirstQueuedIsExclusive()第一个排队的是不是写线程。r(10)，当前线程是第十一个，此时已经有一个写线程排队，r(10)->w，此时排队去。r(10)->w->r。
static final class NonfairSync extends Sync {
final boolean writerShouldBlock() {
return false;
}
final boolean readerShouldBlock() {
return apparentlyFirstQueuedIsExclusive();
} // w->r->r r获取锁 r->r->r
}加载全部内容