读写锁

2018-04-12

ReentrantReadWriteLock 定义

ReentrantReadWriteLock: Reentrant(重入) Read (读) Write(写) Lock(锁), 从字面意思我们了解到, 这是一个可重入的读写锁;
ReentrantReadWriteLock 主要具有以下特点

读写锁都可重入, 线程可同时具有读写锁
线程同时获取读写锁时, 必须先获取 writeLock, 再获取 readLock (也就是锁的降级), 反过来的直接导致死锁(这个问题下面会重点分析)
ReentrantReadWriteLock支持公平与非公平机制, 主要依据是 AQS 类中的Sync Queue 里面是否有节点或 Sync Queue 里面的 head.next 是否是获取 writeLock 的线程节点; 公平模式就会依据获取的先后顺序在 SyncQueue 里面排队获取
读写锁互斥
获取 readLock 的过程中, 若此时有线程已获取写锁或 AQS 的 Sync Queue 里面有获取 writeLock 的线程, 则一定会等待获取writeLock成功并释放或放弃获取后才能获取(PS: 这里有个例外, 在死锁时, 已获取 readLock 的线程还是能重复获取 readLock)
获取 writeLock 时一定是在没有线程获取 readLock 或 writeLock 时才获取成功 (PS: 一个典型的死锁场景就是一个线程先获取readLock, 后又获取writeLock)
锁的获取支持线程中断, 且writeLock 中支持 Condition (PS: condition 只支持排他的场景)

参数

static final int SHARED_SHIFT   = 16;
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
// 高16位读锁，低16位写锁（总量）
/** Returns the number of shared holds represented in count  */
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count  */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
/**
 * A counter for per-thread read hold counts.
 * Maintained as a ThreadLocal; cached in cachedHoldCounter
 */
 //线程所拥有的锁
static final class HoldCounter {
    int count = 0;
    // Use id, not reference, to avoid garbage retention
    final long tid = getThreadId(Thread.currentThread());
}

读锁

final boolean tryReadLock() {
    Thread current = Thread.currentThread();
    for (;;) {
        int c = getState();
        // 同线程既可以获得读锁也可以获得写锁
        // 有独占锁且不是获得独占锁线程的将不能获得写锁
        if (exclusiveCount(c) != 0 &&
            getExclusiveOwnerThread() != current)
            return false;
        int r = sharedCount(c);
        if (r == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 单次在高16位增加1，所以在上层做一个MAX_COUNT == 的操作就足够了。不用改成>=
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            if (r == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                // 缓存上一个获得锁的线程（减少threadLocal.get的操作）
                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 true;
        }
    }
}
protected final int tryAcquireShared(int unused) {
   /*
    * Walkthrough:
    * 1. If write lock held by another thread, fail.
    * 2. Otherwise, this thread is eligible for
    *    lock wrt state, so ask if it should block
    *    because of queue policy. If not, try
    *    to grant by CASing state and updating count.
    *    Note that step does not check for reentrant
    *    acquires, which is postponed to full version
    *    to avoid having to check hold count in
    *    the more typical non-reentrant case.
    * 3. If step 2 fails either because thread
    *    apparently not eligible or CAS fails or count
    *    saturated, chain to version with full retry loop.
    */
   Thread current = Thread.currentThread();
   int c = getState();
   if (exclusiveCount(c) != 0 &&
       getExclusiveOwnerThread() != current)
       return -1;
   int r = sharedCount(c);
   // 公正锁情况下只要有前驱节点就要block
   // 而非公正锁的情况下，只有头结点是独占锁的情况下才会阻塞（代码详见下面）
   // 其他步骤和上述tryReadLock一样
   if (!readerShouldBlock() &&
       r < MAX_COUNT &&
       compareAndSetState(c, c + SHARED_UNIT)) {
       if (r == 0) {
           firstReader = current;
           firstReaderHoldCount = 1;
       } else if (firstReader == current) {
           firstReaderHoldCount++;
       } else {
           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;
   }
   return fullTryAcquireShared(current);
}
/**
 * Nonfair version of Sync
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -8159625535654395037L;
    final boolean writerShouldBlock() {
        return false; // writers can always barge
    }
    final boolean readerShouldBlock() {
        /* As a heuristic to avoid indefinite writer starvation,
         * block if the thread that momentarily appears to be head
         * of queue, if one exists, is a waiting writer.  This is
         * only a probabilistic effect since a new reader will not
         * block if there is a waiting writer behind other enabled
         * readers that have not yet drained from the queue.
         */
        return apparentlyFirstQueuedIsExclusive();
    }
}
/**
 * Returns {@code true} if the apparent first queued thread, if one
 * exists, is waiting in exclusive mode.  If this method returns
 * {@code true}, and the current thread is attempting to acquire in
 * shared mode (that is, this method is invoked from {@link
 * #tryAcquireShared}) then it is guaranteed that the current thread
 * is not the first queued thread.  Used only as a heuristic in
 * ReentrantReadWriteLock.
 */
final boolean apparentlyFirstQueuedIsExclusive() {
    Node h, s;
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}
/**
 * Fair version of Sync
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -2274990926593161451L;
    final boolean writerShouldBlock() {
        return hasQueuedPredecessors();
    }
    final boolean readerShouldBlock() {
        return hasQueuedPredecessors();
    }
}
/**
* Full version of acquire for reads, that handles CAS misses
* and reentrant reads not dealt with in tryAcquireShared.
*/
final int fullTryAcquireShared(Thread current) {
   /*
    * This code is in part redundant with that in
    * tryAcquireShared but is simpler overall by not
    * complicating tryAcquireShared with interactions between
    * retries and lazily reading hold counts.
    */
   HoldCounter rh = null;
   for (;;) {
       int c = getState();
       if (exclusiveCount(c) != 0) {
           if (getExclusiveOwnerThread() != current)
               return -1;
           // else we hold the exclusive lock; blocking here
           // would cause deadlock.
           // 应该被block的情况
           // 此时获取独占锁的且已经是当前线程的（可能锁已经被解放的了）
       } else if (readerShouldBlock()) {
           // Make sure we're not acquiring read lock reentrantly
           if (firstReader == current) {
               // assert firstReaderHoldCount > 0;
           } else {
               // threadLocal 变量需要remove 否则可能会造成内存泄露
               if (rh == null) {
                   rh = cachedHoldCounter;
                   if (rh == null || rh.tid != getThreadId(current)) {
                       rh = readHolds.get();
                       if (rh.count == 0)
                           // 惰性删除
                           readHolds.remove();
                   }
               }
               if (rh.count == 0)
                   return -1;
           }
       }
       if (sharedCount(c) == MAX_COUNT)
           throw new Error("Maximum lock count exceeded");
       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; // cache for release
           }
           return 1;
       }
   }
}

写锁

protected final boolean tryAcquire(int acquires) {
    /*
     * Walkthrough:
     * 1. If read count nonzero or write count nonzero
     *    and owner is a different thread, fail.
     * 2. If count would saturate, fail. (This can only
     *    happen if count is already nonzero.)
     * 3. Otherwise, this thread is eligible for lock if
     *    it is either a reentrant acquire or
     *    queue policy allows it. If so, update state
     *    and set owner.
     */
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);
    if (c != 0) {
        // 其他的线程已经获取读锁
        // (Note: if c != 0 and w == 0 then shared count != 0)
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // Reentrant acquire
        setState(c + acquires);
        return true;
    }
    // c==0
    // 非公正锁下 writerShouldBlock 无脑return false
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);
    return true;
}

参考ReentrantReadWriteLock 源码分析
参考读锁为什么不能升级成写锁