ReentrantReadWriteLock可重入读写锁分析

ReentrantReadWriteLock 可重入的读写锁

什么叫可重入：就是同一个线程可以重复加锁，可以对同一个锁加多次，每次释放的时候回释放一次，直到该线程加锁次数为0，这个线程才释放锁。

什么叫读写锁： 也就是读锁可以共享，多个线程可以同时拥有读锁，但是写锁却只能只有一个线程拥有，而且获取写锁的时候，其他线程都已经释放了读锁，而且在该线程获取写锁之后，其他线程不能再获取读锁。

 

我们先看下下面两个示例： ReentrantReadWriteLock.java自带的两个示例

 

 * class CachedData {
 *   Object data;
 *   volatile boolean cacheValid;
 *   ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
 *
 *   void processCachedData() {
 *     rwl.readLock().lock();
 *     if (!cacheValid) {
 *        // Must release read lock before acquiring write lock
 *        rwl.readLock().unlock();
 *        rwl.writeLock().lock();
 *        // Recheck state because another thread might have acquired
 *        //   write lock and changed state before we did.
 *        if (!cacheValid) {
 *          data = ...
 *          cacheValid = true;
 *        }
 *        // Downgrade by acquiring read lock before releasing write lock
 *        rwl.readLock().lock();
 *        rwl.writeLock().unlock(); // Unlock write, still hold read
 *     }
 *
 *     use(data);
 *     rwl.readLock().unlock();
 *   }
 * }

 

 

如果要对cache的内容进行更新，则必须得加写锁，如果只是读取，则加读锁就可以了。

 

 

 

 * class RWDictionary {
 *    private final Map<String, Data> m = new TreeMap<String, Data>();
 *    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
 *    private final Lock r = rwl.readLock();
 *    private final Lock w = rwl.writeLock();
 *
 *    public Data get(String key) {
 *        r.lock();
 *        try { return m.get(key); }
 *        finally { r.unlock(); }
 *    }
 *    public String[] allKeys() {
 *        r.lock();
 *        try { return m.keySet().toArray(); }
 *        finally { r.unlock(); }
 *    }
 *    public Data put(String key, Data value) {
 *        w.lock();
 *        try { return m.put(key, value); }
 *        finally { w.unlock(); }
 *    }
 *    public void clear() {
 *        w.lock();
 *        try { m.clear(); }
 *        finally { w.unlock(); }
 *    }
 * }}

 

 这个示例，同样，对map进行get，allKeys读操作加读锁，put，clear操作加写锁。

 

我们先看下ReentrantReadWriteLock这个类的两个构造函数：

 

 

    public ReentrantReadWriteLock() {
        this(false);
    }

    /**
     * Creates a new {@code ReentrantReadWriteLock} with
     * the given fairness policy.
     *
     * @param fair {@code true} if this lock should use a fair ordering policy
     */
    public ReentrantReadWriteLock(boolean fair) {
        sync = (fair)? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

 

 

fair这个参数表示是否是创建一个公平的读写锁，还是非公平的读写锁。也就是抢占式还是非抢占式。

 

公平和非公平：公平表示获取的锁的顺序是按照线程加锁的顺序来分配获取到锁的线程时最先加锁的线程，是按照FIFO的顺序来分配锁的；非公平表示获取锁的顺序是无需的，后来加锁的线程可能先获得锁，这种情况就导致某些线程可能一直没获取到锁。

 

公平锁为啥会影响性能，从code上来看看公平锁仅仅是多了一项检查是否在队首会影响性能，如不是，那么又是在什么地方影响的？假如是闯入的线程，会排在队尾并睡觉(parking)等待前任节点唤醒，这样势必会比非公平锁添加很多paking和unparking的操作

 

一般的应用场景是： 如果有多个读线程，一个写线程，而且写线程在操作的时候需要阻塞读线程，那么此时就需要使用公平锁，要不然可能写线程一直获取不到锁，导致线程饿死。

 

我这边在一个项目就用到了读写锁：

 

一个KV引擎的java客户端：java客户端需要缓存KV引擎服务器的集群配置信息（master server，data node service）多个map的数据结构；更新时由后台一个写线程定时更新；而读取则同时可能几十个线程同时读，因为需要读取配置定时定位到一个key对应的data nodeserver去获取数据；为了更好的避免这些缓存的数据读写同步导致的问题，所以使用读写锁来解决同步的问题。读的时候加读锁，写的时候加写锁，在写的过程中需要阻塞读，因为写的过程非常快，所以可以阻塞读。关键是写的过程不能让读线程读取到一部分数据是旧的，一部分是新的，导致获取结果失败甚至出错。而这种case显然是读线程多，写线程只有一个，在测试过程中就发现写线程一直获取不到锁，因为用的是非公平锁，所以后来通过查询API，使用公平锁就可以了。

 

 

下面我们来看下具体读写锁的实现：

 

获取读锁的过程：

 

 

        protected final int tryAcquireShared(int unused) {
            /*
             * Walkthrough:
             * 1. If write lock held by another thread, fail
             * 2. If count saturated, throw error
             * 3. 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.
             * 4. If step 3 fails either because thread
             *    apparently not eligible or CAS fails,
             *    chain to version with full retry loop.
             */
            Thread current = Thread.currentThread();
            int c = getState();
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            if (!readerShouldBlock(current) &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != current.getId())
                    cachedHoldCounter = rh = readHolds.get();
                rh.count++;
                return 1;
            }
            return fullTryAcquireShared(current);
        }        /**
         * 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 = cachedHoldCounter;
            if (rh == null || rh.tid != current.getId())
                rh = readHolds.get();
            for (;;) {
                int c = getState();
                int w = exclusiveCount(c);
                if ((w != 0 && getExclusiveOwnerThread() != current) ||
                    ((rh.count | w) == 0 && readerShouldBlock(current)))
                    return -1;
                if (sharedCount(c) == MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                if (compareAndSetState(c, c + SHARED_UNIT)) {
                    cachedHoldCounter = rh; // cache for release
                    rh.count++;
                    return 1;
                }
            }
        }

 

 

如果写锁已经被获取了，则获取读锁失败；如果当前线程重入加锁次数达到MAX_COUNT，获取读锁失败；readerShouldBlock如果是公平锁，则判断当然线程是否是排在队列前面，如果不是，则等待，是则获取读锁；可重入性是通过class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> threadlocal来保存重入获取锁的次数；然后就是调用fullTryAcquireShared方法对当前线程获取锁的次数进行操作。

 

 

获取写锁的过程：

 

  公平锁获取锁的过程：

    /**

         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (isFirst(current) &&
                    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;
        }

 

  判断当前线程是否是等待队列的头线程，如果是，则把当前的排斥锁线程设置为当前线程，获取写锁成功；否则判断当前写锁的线程是不是当前线程，如果是则对写锁的重入数量进行加1操作；否则获取写锁失败。

 

非公平的写锁获取过程：

 

 

 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;
        }

 

跟公平写锁获取过程不同的是没有判断当前线程是否是等待队列线程的第一个。