Java并发读写锁ReentrantReadWriteLock 使用场景

class cacheddata {
   //被缓存的具体对象
   object data;
   //当前对象是否可用，使用volatile来保证可见性
   volatile boolean cachevalid;
   //reentrantreadwritelock 
   final 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();
       try {
         // recheck state because another thread might have
         // acquired write lock and changed state before we did.
         //经典处理之二，在独占锁内部要处理数据时，一定要做二次校验
         //因为可能同时有多个线程全都在获取写锁，
         //当时线程1释放写锁之后，线程2马上获取到写锁，此时如果不做二次校验那可能就导致某些操作做了多次
         if (!cachevalid) {
           data = ...
           //当缓存对象更新成功后，重置标记为true
           cachevalid = true;
         }
         // downgrade by acquiring read lock before releasing write lock
         //这里有一个非常神奇的锁降级操作，所谓降级是说当你持有写锁后，可以再次获取读锁
         //这里之所以要获取一次写锁是为了防止当前线程释放写锁之后，其他线程马上获取到写锁，改变缓存对象
         //因为读写互斥，所以有了这个读锁之后，在读锁释放之前，别的线程是无法修改缓存对象的
         rwl.readlock().lock();
       } finally {
         rwl.writelock().unlock(); // unlock write, still hold read
       }
     }
     try {
       use(data);
     } finally {
       rwl.readlock().unlock();
     }
   }
 }

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

public void processdata() {
    readlock.lock();
    if (!update) {
        // 必须先释放读锁
        readlock.unlock();
        // 锁降级从写锁获取到开始
        writelock.lock();
        try {
            if (!update) {
                // 准备数据的流程(略)
                update = true;
            }
            readlock.lock();
        } finally {
            writelock.unlock();
        }
        // 锁降级完成，写锁降级为读锁
    }
    try {
        // 使用数据的流程(略)
    } finally {
        readlock.unlock();
    }
}

public class reentrantreadwritelock implements readwritelock, java.io.serializable {}

abstract static class sync extends abstractqueuedsynchronizer {}

// 计数器
static final class holdcounter {
    // 计数
    int count = 0;
    // use id, not reference, to avoid garbage retention
    // 获取当前线程的tid属性的值
    final long tid = getthreadid(thread.currentthread());
}

// 本地线程计数器
static final class threadlocalholdcounter
    extends threadlocal<holdcounter> {
    // 重写初始化方法，在没有进行set的情况下，获取的都是该holdcounter值
    public holdcounter initialvalue() {
        return new holdcounter();
    }
}

abstract static class sync extends abstractqueuedsynchronizer {
    // 版本序列号
    private static final long serialversionuid = 6317671515068378041l;        
    // 高16位为读锁，低16位为写锁
    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;
    // 本地线程计数器
    private transient threadlocalholdcounter readholds;
    // 缓存的计数器
    private transient holdcounter cachedholdcounter;
    // 第一个读线程
    private transient thread firstreader = null;
    // 第一个读线程的计数
    private transient int firstreaderholdcount;
}

// 构造函数
sync() {
    // 本地线程计数器
    readholds = new threadlocalholdcounter();
    // 设置aqs的状态
    setstate(getstate()); // ensures visibility of readholds
}

public class reentrantreadwritelock
        implements readwritelock, java.io.serializable {
    // 版本序列号    
    private static final long serialversionuid = -6992448646407690164l;    
    // 读锁
    private final reentrantreadwritelock.readlock readerlock;
    // 写锁
    private final reentrantreadwritelock.writelock writerlock;
    // 同步队列
    final sync sync;
    private static final sun.misc.unsafe unsafe;
    // 线程id的偏移地址
    private static final long tid_offset;
    static {
        try {
            unsafe = sun.misc.unsafe.getunsafe();
            class<?> tk = thread.class;
            // 获取线程的tid字段的内存地址
            tid_offset = unsafe.objectfieldoffset
                (tk.getdeclaredfield("tid"));
        } catch (exception e) {
            throw new error(e);
        }
    }
}

public reentrantreadwritelock() {
    this(false);
}

public reentrantreadwritelock(boolean fair) {
    // 公平策略或者是非公平策略
    sync = fair ? new fairsync() : new nonfairsync();
    // 读锁
    readerlock = new readlock(this);
    // 写锁
    writerlock = new writelock(this);
}

static int sharedcount(int c)    { return c >>> shared_shift; }

static int exclusivecount(int c) { return c & exclusive_mask; }

static final int exclusive_mask = (1 << shared_shift) - 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) { // 状态不为0
        // (note: if c != 0 and w == 0 then shared count != 0)
        if (w == 0 || current != getexclusiveownerthread()) // 写线程数量为0或者当前线程没有占有独占资源
            return false;
        if (w + exclusivecount(acquires) > max_count) // 判断是否超过最高写线程数量
            throw new error("maximum lock count exceeded");
        // reentrant acquire
        // 设置aqs状态
        setstate(c + acquires);
        return true;
    }
    if (writershouldblock() ||
        !compareandsetstate(c, c + acquires)) // 写线程是否应该被阻塞
        return false;
    // 设置独占线程
    setexclusiveownerthread(current);
    return true;
}

/*
* note that tryrelease and tryacquire can be called by
* conditions. so it is possible that their arguments contain
* both read and write holds that are all released during a
* condition wait and re-established in tryacquire.
*/
protected final boolean tryrelease(int releases) {
    // 判断是否伪独占线程
    if (!isheldexclusively())
        throw new illegalmonitorstateexception();
    // 计算释放资源后的写锁的数量
    int nextc = getstate() - releases;
    boolean free = exclusivecount(nextc) == 0; // 是否释放成功
    if (free)
        setexclusiveownerthread(null); // 设置独占线程为空
    setstate(nextc); // 设置状态
    return free;
}

private illegalmonitorstateexception unmatchedunlockexception() {
    return new illegalmonitorstateexception(
        "attempt to unlock read lock, not locked by current thread");
}
// 共享模式下获取资源
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) // 写线程数不为0并且占有资源的不是当前线程
        return -1;
    // 读锁数量
    int r = sharedcount(c);
    if (!readershouldblock() &&
        r < max_count &&
        compareandsetstate(c, c + shared_unit)) { // 读线程是否应该被阻塞、并且小于最大值、并且比较设置成功
        if (r == 0) { // 读锁数量为0
            // 设置第一个读线程
            firstreader = current;
            // 读线程占用的资源数为1
            firstreaderholdcount = 1;
        } else if (firstreader == current) { // 当前线程为第一个读线程
            // 占用资源数加1
            firstreaderholdcount++;
        } else { // 读锁数量不为0并且不为当前线程
            // 获取计数器
            holdcounter rh = cachedholdcounter;
            if (rh == null || rh.tid != getthreadid(current)) // 计数器为空或者计数器的tid不为当前正在运行的线程的tid
                // 获取当前线程对应的计数器
                cachedholdcounter = rh = readholds.get();
            else if (rh.count == 0) // 计数为0
                // 设置
                readholds.set(rh);
            rh.count++;
        }
        return 1;
    }
    return fulltryacquireshared(current);
}

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) { // 写线程数量不为0
            if (getexclusiveownerthread() != current) // 不为当前线程
                return -1;
            // else we hold the exclusive lock; blocking here
            // would cause deadlock.
        } else if (readershouldblock()) { // 写线程数量为0并且读线程被阻塞
            // 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)) { // 计数器为空或者计数器的tid不为当前正在运行的线程的tid
                        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) { // 读线程数量为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 tryreleaseshared(int unused) {
    // 获取当前线程
    thread current = thread.currentthread();
    if (firstreader == current) { // 当前线程为第一个读线程
        // assert firstreaderholdcount > 0;
        if (firstreaderholdcount == 1) // 读线程占用的资源数为1
            firstreader = null;
        else // 减少占用的资源
            firstreaderholdcount--;
    } else { // 当前线程不为第一个读线程
        // 获取缓存的计数器
        holdcounter rh = cachedholdcounter;
        if (rh == null || rh.tid != getthreadid(current)) // 计数器为空或者计数器的tid不为当前正在运行的线程的tid
            // 获取当前线程对应的计数器
            rh = readholds.get();
        // 获取计数
        int count = rh.count;
        if (count <= 1) { // 计数小于等于1
            // 移除
            readholds.remove();
            if (count <= 0) // 计数小于等于0，抛出异常
                throw unmatchedunlockexception();
        }
        // 减少计数
        --rh.count;
    }
    for (;;) { // 无限循环
        // 获取状态
        int c = getstate();
        // 获取状态
        int nextc = c - shared_unit;
        if (compareandsetstate(c, nextc)) // 比较并进行设置
            // releasing the read lock has no effect on readers,
            // but it may allow waiting writers to proceed if
            // both read and write locks are now free.
            return nextc == 0;
    }
}

void processcacheddata() {
    rwl.readlock().lock();
    if (!cachevalid) {
        // must release read lock before acquiring write lock
        rwl.readlock().unlock();
        rwl.writelock().lock();
        try {
            // 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();
    } finally {
      rwl.writelock().unlock(); // unlock write, still hold read
    }
  }
  try {
    use(data);
  } finally {
    rwl.readlock().unlock();
  }
}

public class cache {
    private map<string, object> data = new hashmap<>();
    private readwritelock lock = new reentrantreadwritelock();
    public object get(string key) {
        lock.readlock().lock();
        try {
            return data.get(key);
        } finally {
            lock.readlock().unlock();
        }
    }
    public void put(string key, object value) {
        lock.writelock().lock();
        try {
            data.put(key, value);
        } finally {
            lock.writelock().unlock();
        }
    }
}

public class counter {
    private int count = 0;
    private stampedlock lock = new stampedlock();
    public int getcount() {
        long stamp = lock.tryoptimisticread();
        int value = count;
        if (!lock.validate(stamp)) {
            stamp = lock.readlock();
            try {
                value = count;
            } finally {
                lock.unlockread(stamp);
            }
        }
        return value;
    }
    public void increment() {
        long stamp = lock.writelock();
        try {
            count++;
        } finally {
            lock.unlockwrite(stamp);
        }
    }
}