并发编程之读写锁ReentrantReadWriteLock实现

引子   

我们团队有维护这样一个类似比价的系统。其中有一个A方法需要查询好几个第三方的http接口，以获取某些信息进行汇总分析。虽然是通过多线程并发的去访问第三方接口，但是有些第三方系统不稳定，只要其中一个挂了就会影响整个方法的效率。团队成员经常在接到报警信息之后手动把不稳定的接口下线，这给大家的正常生活带来了麻烦，因为你有时候不得不不周末或者半夜起来操作。于是我做了一个自动下线功能，假如某个接口在1分钟内抛出的异常大于某个阈值之后自动下线一段时间，并在下线一段时间之后再自动上线，如果上线之后发现异常还没有减少则继续下线。为了实现这个功能，首先需要在抛异常的采集信息，并判断是否需要下线。A方法在请求第三方接口之前需要判断这个接口有没有下线，有下线之后则不再调用该http请求。在这种场景下，明显是读的行为比写的行为多，因为每次A方法都要读操作，而只有在抛异常的情况下才需要写操作。为了尽量保证方法的性能，想到了用读写锁来实现，但是之前没有用过读写锁，于是对读写锁的实现进行了学习。

读写锁

   读写锁实际是一种特殊的自旋锁，它把对共享资源的访问者划分成读者和写者，读者只对共享资源进行读访问，写者则需要对共享资源进行写操作。读写锁允许多个读者访问共享资源；读者和写者，多个写着只允许一个访问共享资源，因此在读操作多而写操作少的场景下，读写锁比普通的互斥锁能获得更高的并发能力。

 

读写锁规则

1.获得读锁：没有其他线程获得写锁，并且没有线程在请求写操作。

2.获得写锁：没有线程获得读锁和写锁。

3.可重入：如果一个线程获得读/写锁，那么该线程可以再次获得该锁。

4.读锁升级为写锁：有时候我们希望一个获得读锁的线程，也能获得写锁。读锁升级为写锁必须满足该线程是唯一拥有读锁的条件，即除了该线程之外，再没有其他线程拥有读锁。

5.写锁降级为读锁：即拥有写锁的线程，可以同时获得读锁。因为一个线程拥有了写锁，那么就不会有其他线程获得写锁和读锁了，而对于一个线程同时拥有读锁和写锁是没有什么危险的。

6.线程活跃度风险:如果读操作非常频繁，那么写操作可能一直获取不到写锁，从而产生写线程”饥饿“，那么需要一直机制去解决活跃度风险。

 

读写锁使用

java的读写锁是用ReentrantReadWriteLock实现的，ReentrantReadWriteLock使用比较简单，首先创建读写锁ReentrantReadWriteLock的实例，读方法用读锁锁住，写方法用写锁锁住，与所有显式锁一样，必须在finally中释放锁，下面是使用方法示例：

 

public ReadWriteLockExample {

       private final ReadWriteLock lock = new ReentrantReadWriteLock();
	private final Lock readLock = lock.readLock();
	private final Lock writeLock = lock.writeLock();


     public Object readMothed() {
          readLock.lock();
	     try {
		 your code;
	     } finally {
		readLock.unlock();
	    }
         }

    public void writeMothed() {
        writeLock.lock();
        try{
	    your code;
        } finally {
	    writeLock.unlock();
        }
    }
}

 

 

读写锁原理及实现

前面说过，java的读写锁是ReentrantReadWriteLock类实现的。他有3个自己实现的内部类ReadLock， writerLock，Sync。 同时Sync又有2个子类，一个是FairSync和NonFairSync，主要区别是获得读写锁的公平性，即解决线程饥饿的方法不同。

 

/** Inner class providing readlock */
    private final ReentrantReadWriteLock.ReadLock readerLock;
    /** Inner class providing writelock */
    private final ReentrantReadWriteLock.WriteLock writerLock;
    /** Performs all synchronization mechanics */
    private final Sync sync;

 

 

 

 顾名思义，ReadLock是读锁，writerLock是写锁，Sync是真正实现读写锁功能的类。ReadLock,writerLock的Lock,unLock方法都是委托Sync类来实现的。Sync扩展自抽象类AbstractQueuedSynchronizer，关于AQS的介绍，可以参考并发编程之AbstractQueuedSynchronizer原理剖析。使用一个int型state字段来管理读写请求线程数。高16位表示持有读锁的线程数，低16位表示持有写锁的线程数（0或1）以及请求写锁的请求数。使用threadlocal readHolds来记录当前线程持有的读锁数，同时还用了一个cachedHoldCounter来保存上一次成功获得读锁的线程及其读锁数量，还有一个exclusiveOwnerThread字段来保存拥有写锁的线程。最后Sync还继承出fairSync与nonfairSync类来解决线程活跃度问题。

读锁的实现

    读锁的lock是委托tryAcquireShared方法来实现的，如果tryAcquireShared返回小于0则通过自旋的方式继续调用tryAcquireShared方法直到获得锁或被中断为止。下面是tryAcquireShared的具体实现：

protected final int tryAcquireShared(int unused) {
            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);
        }

 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
                return -1;  表示如果请求和持有写锁的线程数不为0并且持有写锁的线程不是当前线程 则不能获得读锁。言外之意就是，如果没有线程持有写锁也没有其他线程请求写锁时，有机会获得读锁，同时如果持有写锁的线程是当前线程，那么当前线程也有机会获得读锁。这实现了上面读写锁的第1,5条规则。

 由于state的低16位表示持有读锁的线程数，如果该数超过了0xFFFF抛出 Error("Maximum lock count exceeded");错误。：

if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");

 如果前面两个条件都没有拦截住线程获取读锁的决心，那么最后一关来了：

 

 if (!readerShouldBlock(current) &&
                compareAndSetState(c, c + SHARED_UNIT))

 readerShouldBlock（current）方法判断当前线程需不需要继续阻塞，这个条件是为解决写锁活跃度风险。readerShouldBlock方法由Sync的子类fairSync和nonfairSync来实现。在nonfairSync中，如果阻塞队列中下一个请求是写请求，那么当前线程就不能获得读锁了。在fairSync中，如果阻塞队列为空或者当前线程是队列的head时，才允许获当前线程获得读锁。最后同时满足CAS条件时，这个线程才终于获得读锁了。

 

如果前面3个if条件都没能返回-1或1时，最终会调用fullTryAcquireShared方法再判断一遍完整的获取读锁逻辑，由于跟前面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;
                }
            }
        }

 

 

 

写锁的实现

写锁的lock是委托Sync的tryAcquire方法来实现的，但是由于fairSync和nonfairSync对了公平性策略不同而采用可不同的实现方法，所以这里分开来写。

nonfairSync的写锁调用的是nonfairTryAcquire方法

 

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

    首先，c==0表示当前既没有线程获得读锁，也没有线程获得写锁，那么只要当前线程对state的CAS操作成功，则就可以获得写锁，并设置当前线程为获得写锁的线程；如果有其他线程获得读锁或写锁时，如果获得写锁的线程是当前线程，那么可以继续获得写锁。这实现了前面读写锁规则的第2条和第3条。

 

fairSync的写锁实现

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

    该方法其实与nonfairSync的整体逻辑差不多，只是增加了一个是否阻塞的的判断isFirst(current)。当前既没有线程获得读锁，也没有线程获得写锁，那么当前线程有没有机会获得写锁，还得看阻塞队列的情况，如果阻塞队列为空，或者当前线程是阻塞队列的head，那么就有获得写锁的机会。