ReentrantReadWriteLock读写锁源码分析

爱被打了一巴掌 2022-12-10 07:04 155阅读 0赞

# ReentrantReadWriteLock读写锁源码分析 #

## 读写状态的设计 ##

ReentrantReadWriteLock也是通过自定义AQS（抽象队列同步器）实现。同步器内部只有一个状态，而读写锁需要维护两个状态：读状态与写状态。

ReentrantReadWriteLock将同步器内部的状态state按位进行拆分：高16位代表读状态，低16位代表写状态。

java.util.concurrent.locks.ReentrantReadWriteLock.Sync

static final int SHARED_SHIFT   = 16; // 读锁偏移的位数
    static final int SHARED_UNIT    = (1 << SHARED_SHIFT); // 每多获取一次读锁，state=state+SHARED_UNIT，相当于左移一位
    static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1; // 写锁可重入的最大次数、读锁最大可获取次数（不等于获取读锁的线程数，一个线程可能获取多次）
    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; } // 写锁的计数，也就是它的重入次数

举个例子来说明一下state加锁时值的变化：

*  假如现在state=65537：
 *  读锁计数：将65537无符号右移16位，获得读锁的计数为1。
 *  写锁计数：将65537的高16位全部置为0，也就是65537&65535=1，获取写锁的计数为1。

### t1和t2同时加读锁 ###

读读共享并发。

package com.morris.concurrent.lock.reentrantreadwritelock.trace;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    /** * 两个线程加读锁 */
    @Slf4j
    public class ReadReadDemo { 
        private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        private static ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
    
        public static void main(String[] args) { 
            new Thread(ReadReadDemo::read, "t1").start();
            new Thread(ReadReadDemo::read, "t2").start();
        }
    
        private static void read() { 
            readLock.lock();
            try { 
                log.info(Thread.currentThread().getName() + " get readLock");
                TimeUnit.SECONDS.sleep(100);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } finally { 
                readLock.unlock();
            }
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

### t1读锁的重入 ###

package com.morris.concurrent.lock.reentrantreadwritelock.trace;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    /** * 读锁的重入 */
    @Slf4j
    public class ReadLockDemo2 { 
        private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        private static ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
    
        public static void main(String[] args) throws InterruptedException { 
            new Thread(() -> { 
                readLock.lock();
                try { 
                    log.info(Thread.currentThread().getName() + " get readLock");
    
                    readLock.lock();
                    try { 
                        log.info(Thread.currentThread().getName() + " get readLock");
                    } finally { 
                        readLock.unlock();
                    }
                } finally { 
                    readLock.unlock();
                }
            }, "t1").start();
            TimeUnit.SECONDS.sleep(1);
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

### t1写锁的重入 ###

package com.morris.concurrent.lock.reentrantreadwritelock.trace;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    @Slf4j
    public class Write2Demo { 
    
        private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        private static ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
    
        public static void main(String[] args) { 
    
            new Thread(() -> { 
                writeLock.lock();
                try { 
                    writeLock.lock();
                    try { 
                    } finally { 
                        writeLock.unlock();
                    }
                } finally { 
                    writeLock.unlock();
                }
            }, "t1").start();
    
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]

### t1写锁的降级 ###

锁降级指的是写锁降级成为读锁，锁降级是指当前线程持有写锁，然后再去获取到读锁，随后释放之前拥有的写锁的过程。如果当前线程拥有写锁，然后将其释放，最后再获取读锁，这种分段完成的过程不能称之为锁降级。

为什么可以降级？因为当线程持有写锁的时候没有其它线程可以获取读写锁，因此再获取读锁是安全的。而此时再释放写锁就会降级为读锁，其它线程也可以获得读锁。

为什么不可以升级？因为这会导致死锁，假设有A、B、C三个线程，它们都已持有读锁。假设线程A尝试从读锁升级到写锁。那么它必须等待B和C释放掉已经获取到的读锁。如果随着时间推移，B和C逐渐释放了它们的读锁，此时线程 A确实是可以成功升级并获取写锁。但是我们考虑一种特殊情况，假设线程A和B都想升级到写锁，那么对于线程A而言，它需要等待其他所有线程，包括线程B在内释放读锁。而线程B也需要等待所有的线程，包括线程A释放读锁。这就是一种非常典型的死锁的情况，谁都愿不愿意率先释放掉自己手中的锁。

package com.morris.concurrent.lock.reentrantreadwritelock.trace;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    @Slf4j
    public class WriteReadDemo { 
    
        private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
        private static ReentrantReadWriteLock.ReadLock readLock = reentrantReadWriteLock.readLock();
        private static ReentrantReadWriteLock.WriteLock writeLock = reentrantReadWriteLock.writeLock();
    
        public static void main(String[] args) { 
    
            new Thread(() -> { 
                writeLock.lock();
                try { 
                    readLock.lock();
                    try { 
                    } finally { 
                        readLock.unlock();
                    }
                } finally { 
                    writeLock.unlock();
                }
            }, "t1").start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 3]

## 线程重入计数器 ##

AQS的同步状态的高16位只能记录当前持有读锁的次数，那么每个线程对读锁的重入次数怎么记录呢？使用HoldCounter。

static final class HoldCounter { 
        int count = 0; // 读锁的重入次数
        final long tid = getThreadId(Thread.currentThread()); // 获取读锁的线程id
    }
    
    //使用ThreadLocal保证每个线程都有一份HoldCounter，线程安全
    static final class ThreadLocalHoldCounter
        extends ThreadLocal<HoldCounter> { 
        public HoldCounter initialValue() { 
            return new HoldCounter();
        }
    }
    
    private transient ThreadLocalHoldCounter readHolds; // 保存所有线程的HoldCounter，HoldCounter里面存的每个线程重入读锁的次数
    
    private transient HoldCounter cachedHoldCounter; // 缓存当前线程的HoldCounter
    
    private transient Thread firstReader = null; // 记录第一个获取读锁的线程，不会放入readHolds
    
    private transient int firstReaderHoldCount; // 记录第一个获取读锁的线程持有读锁的次数，不会放入readHolds
    
    Sync() { 
        readHolds = new ThreadLocalHoldCounter(); // 初始化ThreadLocal
        setState(getState()); // ensures visibility of readHolds
    }

## 写锁的获取 ##

写锁的逻辑主要在java.util.concurrent.locks.ReentrantReadWriteLock.Sync\#tryAcquire，其他逻辑跟ReentrantLock差不多。

protected final boolean tryAcquire(int acquires) { 
    
        Thread current = Thread.currentThread();
        int c = getState();
        int w = exclusiveCount(c); // 获取写锁的次数
        // c!=0 说明存在锁
        if (c != 0) { 
    
            // c != 0 && w == 0 说明存在写锁
            // current != getExclusiveOwnerThread() 表示当前线程不是获取锁的线程，不是重入，其他线程获取了写锁，所以返回false
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 锁的重入
            setState(c + acquires);
            return true;
        }
        // writerShouldBlock()由子类实现是否要公平
        if (writerShouldBlock() ||
            !compareAndSetState(c, c + acquires))
            return false;
        setExclusiveOwnerThread(current);
        return true;
    }

## 写锁的释放 ##

写锁的逻辑主要在java.util.concurrent.locks.ReentrantReadWriteLock.Sync\#tryRelease

protected final boolean tryRelease(int releases) { 
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;
        boolean free = exclusiveCount(nextc) == 0; // 判断写锁的次数是否为0，其他与独占锁的释放逻辑一致
        if (free)
            setExclusiveOwnerThread(null);
        setState(nextc);
        return free;
    }

## 读锁的获取 ##

java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock\#lock

public void lock() { 
        sync.acquireShared(1);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#acquireShared

public final void acquireShared(int arg) { 
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

java.util.concurrent.locks.ReentrantReadWriteLock.Sync\#tryAcquireShared

protected final int tryAcquireShared(int unused) { 
    
        Thread current = Thread.currentThread(); // 获取当前线程
        int c = getState(); // 获取锁的状态
    
        // exclusiveCount(c) != 0 有线程获取写锁
        // getExclusiveOwnerThread() != current // 获取写锁的线程不是自己
        if (exclusiveCount(c) != 0 &&
            getExclusiveOwnerThread() != current)
            return -1;
        // 到这里有两种情况
        // 1. 没有线程获取写锁
        // 2. 有线程获取了写锁，并且获取写锁的线程是自己，也就写锁降级为读锁
        int r = sharedCount(c); // 获取读锁的次数
        if (!readerShouldBlock() && // 处理公平与非公平
            r < MAX_COUNT &&
            compareAndSetState(c, c + SHARED_UNIT)) { 
            // 进入这里说明获取到了读锁
            if (r == 0) {  // 说明没有线程获取读锁
                firstReader = current; // 记录第一个获取读锁的线程
                firstReaderHoldCount = 1; // 记录第一个获取读锁的线程的重入次数
            } else if (firstReader == current) { 
                firstReaderHoldCount++; // 第一个获取读锁的线程的重入次数+1
            } else { 
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get(); // 缓存当前线程的HoldCounter
                else if (rh.count == 0)
                    readHolds.set(rh); // 这里只有最后一个获取锁的的线程先持有了锁，然后释放了，再获取才会进入（此时前面还有个线程持有了锁）
                rh.count++; // 重入次数+1
            }
            return 1;
        }
        return fullTryAcquireShared(current);
    }
    
    // 快速尝试获取读锁失败，则改为自旋获取，与上面的逻辑大同小异
    final int fullTryAcquireShared(Thread current) { 
    
        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.
            } else if (readerShouldBlock()) { 
                // Make sure we're not acquiring read lock reentrantly
                if (firstReader == current) { 
                    // assert firstReaderHoldCount > 0;
                } else { 
                    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;
            }
        }
    }

为了提高读锁的效率，会记录第一个获取锁的线程及它的重入次数，也会记录最后一个持有锁的线程的重入次数，而其他持有锁的线程的重入次数会存到每个线程对应的ThreadLocal中。

当上面尝试获取读锁失败后，后进入到队列然后开始休眠：

java.util.concurrent.locks.AbstractQueuedSynchronizer\#doAcquireShared

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) { 
                        setHeadAndPropagate(node, r); // 当前线程获取到读锁后会唤醒下一个获取读锁的线程
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                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); // 当前线程对应的节点会成为头节点，以前的头结点会被GC回收
    
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) { 
            Node s = node.next;
            if (s == null || s.isShared()) // 下一个节点是读
                doReleaseShared(); // 唤醒
        }
    }
    
    private void doReleaseShared() { 
        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;
        }
    }

## 读锁的释放 ##

java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock\#unlock

public void unlock() { 
        sync.releaseShared(1);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#releaseShared

public final boolean releaseShared(int arg) { 
        if (tryReleaseShared(arg)) {  只有读锁全部都被释放了才会返回true
            doReleaseShared(); // 唤醒获取写锁的线程
            return true;
        }
        return false;
    }

java.util.concurrent.locks.ReentrantReadWriteLock.Sync\#tryReleaseShared

protected final boolean tryReleaseShared(int unused) { 
        Thread current = Thread.currentThread();
        if (firstReader == current) { 
            // assert firstReaderHoldCount > 0;
            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) { 
                readHolds.remove(); // ThreadLocal用完要删除，避免内存泄漏
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
        for (;;) { 
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc)) // 获取读锁的次数-1
                return nextc == 0; // 没有线程持了锁的
        }
    }

## 公平的实现 ##

static final class FairSync extends Sync { 
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock() { 
            return hasQueuedPredecessors();
        }
        final boolean readerShouldBlock() { 
            return hasQueuedPredecessors();
        }
    }

writerShouldBlock和readerShouldBlock方法都表示当有别的线程也在尝试获取锁时，是否应该阻塞。

对于公平模式，hasQueuedPredecessors()方法表示前面是否有等待线程。一旦前面有等待线程，那么为了遵循公平，当前线程也就应该被加入同步队列，然后阻塞。

## 非公平的实现 ##

static final class NonfairSync extends Sync { 
        private static final long serialVersionUID = -8159625535654395037L;
        final boolean writerShouldBlock() { 
            return false; // 写锁一直能够被获取
        }
        final boolean readerShouldBlock() { 
            return apparentlyFirstQueuedIsExclusive();
        }
        
        // 如果同步队列中的第一个节点获取写锁，则读锁不能获取成功，为了避免写锁的饥饿
        final boolean apparentlyFirstQueuedIsExclusive() { 
            Node h, s;
            return (h = head) != null &&
                (s = h.next)  != null &&
                !s.isShared()         &&
                s.thread != null;
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/e23ef8e4909041fda7a4664b1f659068.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/641dd170327b44d4a9cdb2cc0da3219a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/08445ad397dc4634a10c06684435078f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/d985dc6a304b421ca26c97c8fa513194.png