可重入读写锁ReentrantReadWriteLock源码解析

public interface ReadWriteLock {
    /**
     * 返回读锁
     */
    Lock readLock();
 
    /**
     * 返回写锁
     */
    Lock writeLock();
}

/**
 * 每个线程读取计数器的计数
 * 维持在ThreadLocal或缓存在cachedHoldCounter中
 */
static final class HoldCounter {
	int count = 0;
	// 使用id而不是引用,以避免垃圾保留
	final long tid = getThreadId(Thread.currentThread());
}

/**
 * ThreadLocal的子类. 本地线程计数器
 */
static final class ThreadLocalHoldCounter
	extends ThreadLocal<HoldCounter> {
	// 重写初始化方法
	public HoldCounter initialValue() {
		return new HoldCounter();
	}
}

/*
* 读取和写入时计数所需常量
* 从逻辑上锁状态被分为两个无符号short:
* 低位表示写锁计数,高位表示读锁计数
*/
// 共享计数偏移量
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;

/**
 * 当前线程持有的重入读锁的数目
 * 只能在构造函数和readObject方法中被初始化
 * 当计数降至0时被删除
 */
private transient ThreadLocalHoldCounter readHolds;

/**
 * 缓存的线程计数
 * 获取readLock的最后一个线程的保留计数
 * 通常情况下保存在ThreadLocal里
 */
private transient HoldCounter cachedHoldCounter;

/**
 * 获取读锁定的第一个线程
 * 更确切地说,firstReader是上次将共享计数从0更改为1的唯一线程
 */
private transient Thread firstReader = null;
// 第一个读线程的计数
private transient int firstReaderHoldCount;

/**
 * 设置本地线程计数器和State的状态
 */
Sync() {
	readHolds = new ThreadLocalHoldCounter();
	setState(getState()); // ensures visibility of readHolds
}

// 获取到读锁占有的线程数量
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }

// 写锁占有的线程数量
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

// 重写的tryAcquire方法
protected final boolean tryAcquire(int acquires) {
	/*
	 * Walkthrough:
	 * 1.如果读取计数非零或写入计数非零,并且所有者非零,则会失败
	 * 2.如果计数饱和也会失败(仅当count已为非零时才会发生)
	 * 3.否则,如果该线程是可重入的获取或队列策略允许的,则它有资格
	 *   获得锁定,如果是,则更新状态和设置所有者
	 */
	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)
		// 写线程数量为0(写锁还没有被获取)或当前线程不是独占资源的线程
		if (w == 0 || current != getExclusiveOwnerThread())
			return false;
		// 判断是否超过最高写线程数量	
		if (w + exclusiveCount(acquires) > MAX_COUNT)
			throw new Error("Maximum lock count exceeded");
		// Reentrant acquire
		// 设置state值,支持重入
		setState(c + acquires);
		return true;
	}
	// 此时c==0,空闲读锁和写锁都没用被获取
	if (writerShouldBlock() ||
		!compareAndSetState(c, c + acquires))
		return false;
	setExclusiveOwnerThread(current);
	return true;
}

/*
 * tryAcquire和tryRelease都可以被Conditions调用
 */
// 重写的tryReleases方法
protected final boolean tryRelease(int releases) {
	// 当前线程是否是独占资源的
	if (!isHeldExclusively())
		throw new IllegalMonitorStateException();
	// state值减去releases	
	int nextc = getState() - releases;
	// 判断写锁状态是否为0,如果为0就完全释放,否则就false
	boolean free = exclusiveCount(nextc) == 0;
	if (free)
		setExclusiveOwnerThread(null);
	setState(nextc);
	return free;
}

// 共享模式下获取资源,参数是unused,因为读锁的重入技术是内部维护的
protected final int tryAcquireShared(int unused) {
	Thread current = Thread.currentThread();
	int c = getState();
	// 先获取低16位写锁,存在写锁并且当前线程不是获取写锁的线程,返回-1即获取失败
	if (exclusiveCount(c) != 0 &&
		getExclusiveOwnerThread() != current)
		return -1;
	// 再去获取高16位的读锁	
	int r = sharedCount(c);
	/**
	 * 先判断当前线程是否应该被阻塞
	 * MAX_COUNT为获取读锁的最大数量,16位的最大值,即判断读锁占有数量是否小于MAX_COUNT
	 * 在上述两个都为true的情况下,CAS设置c高位+1
	 */
	if (!readerShouldBlock() &&
		r < MAX_COUNT &&
		compareAndSetState(c, c + SHARED_UNIT)) {
		// r为0表示是firstReader,计数不会放到readHolds,这样在读锁只有一个的情况下就避免了查找readHolds
		if (r == 0) {
			firstReader = current;
			firstReaderHoldCount = 1;
		// firstReader重入	
		} else if (firstReader == current) {
			firstReaderHoldCount++;
		// 非firstReader读锁重入计数更新	
		} else {
			// 读锁重入计数缓存,基于ThreadLocal实现
			HoldCounter rh = cachedHoldCounter;
			// 计数缓存为null或其tid不是当前线程
			if (rh == null || rh.tid != getThreadId(current))
				// 获取当前线程对应的计数器
				cachedHoldCounter = rh = readHolds.get();
			else if (rh.count == 0)
				// 计数为0则设置
				readHolds.set(rh);
			rh.count++;
		}
		return 1;
	}
	/**
	 * 第一次获取读锁失败有两种情况:
	 * 1).没有写锁被占用时,尝试通过一次CAS去获取时,更新失败(说明有其他读锁在申请)
	 * 2).当前线程占有写锁,并且有其他写锁在当前线程的下一个节点等待获取写锁,除非
	 * 当前线程的下一个节点被取消,否则fullTryAcquireShared也获取不到读锁
	 */
	return fullTryAcquireShared(current);
}

protected final boolean tryReleaseShared(int unused) {
    // 获取当前线程
    Thread current = Thread.currentThread();、
    // 当前线程为第一读线程
    if (firstReader == current) {
        // assert firstReaderHoldCount > 0;
        // 第一个读线程占用资源数为1,表示释放掉此线程即可
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else // 否则得减少第一个读线程重入个数
            firstReaderHoldCount--;
    } else { // 当前线程不是第一个读线程
        // 获取缓存计数器
        HoldCounter rh = cachedHoldCounter;
        // 计数值为空或计数器的tid不为当前线程的tid
        if (rh == null || rh.tid != getThreadId(current))
            // 获取到当前线程的计数器
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            // 计数<=1直接移除
            readHolds.remove();
            // 此时计数<=0抛出异常
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        // 减少计数
        --rh.count;
    }
    for (;;) { // 自旋CAS
        // 获取状态
        int c = getState();
        // 减去1<<16
        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;
    }
}

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

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -8159625535654395037L;
    // 直接返回false,可插队
    final boolean writerShouldBlock() {
        return false; // writers can always barge
    }
    final boolean readerShouldBlock() {
        return apparentlyFirstQueuedIsExclusive();
    }
}

//当head节点不为null且head节点的下一个节点s不为null且s是独占模式(写线程)且s的线程不为null时,返回true.
//目的是不应该让写锁始终等待.作为一个启发式方法用于避免可能的写线程饥饿,这只是一种概率性的作用,因为如果有一个等待的写线程在其他尚未从队列中出队的读线程后面等待,那么新的读线程将不会被阻塞
 final boolean apparentlyFirstQueuedIsExclusive() {
     Node h, s;
     return (h = head) != null &&
         (s = h.next)  != null &&
         !s.isShared()         &&
         s.thread != null;
}

/**
 * ReadLock内部类
 */
public static class ReadLock implements Lock, java.io.Serializable {
	private static final long serialVersionUID = -5992448646407690164L;
	private final Sync sync;

	/**
	 * 子类构造方法
	 */
	protected ReadLock(ReentrantReadWriteLock lock) {
		sync = lock.sync;
	}

	/**
	 * 尝试以共享方式获取读锁(具体AQS中的实现)
	 * 如果写锁没有被另外线程占有,则可以获取读锁
	 */
	public void lock() {
		sync.acquireShared(1);
	}

	/**
	 * 尝试以共享方式获取读锁,但是可以中断的
	 */
	public void lockInterruptibly() throws InterruptedException {
		sync.acquireSharedInterruptibly(1);
	}

	/**
	 * 尝试获取读锁(Sync类的获取读锁)
	 */
	public boolean tryLock() {
		return sync.tryReadLock();
	}

	/**
	 * 如果写锁在给定的时间内没有被另一个线程锁定,则读锁可以获取
	 * 如果锁以公平方式来获取,则其他线程在等待时当前获取的则被锁定
	 */
	public boolean tryLock(long timeout, TimeUnit unit)
			throws InterruptedException {
		return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
	}

	/**
	 * 尝试以共享方式去释放锁
	 */
	public void unlock() {
		sync.releaseShared(1);
	}

	/**
	 * 不支持创建条件等待队列
	 */
	public Condition newCondition() {
		throw new UnsupportedOperationException();
	}

	/**
	 * 返回锁的字符串描述和锁状态值
	 */
	public String toString() {
		int r = sync.getReadLockCount();
		return super.toString() +
			"[Read locks = " + r + "]";
	}
}

/**
 * WriteLock内部类
 */
public static class WriteLock implements Lock, java.io.Serializable {
	private static final long serialVersionUID = -4992448646407690164L;
	private final Sync sync;

	/**
	 * 子类内部构造方法
	 */
	protected WriteLock(ReentrantReadWriteLock lock) {
		sync = lock.sync;
	}

	/**
	 * 尝试以独占方式获取写锁
	 * 因为读锁和写锁互斥,只有在读锁和写锁都未被锁定时,则获取成功并且
	 * state值为1
	 */
	public void lock() {
		sync.acquire(1);
	}

	/**
	 * 尝试以独占方式获取写锁,并且是可中断的
	 */
	public void lockInterruptibly() throws InterruptedException {
		sync.acquireInterruptibly(1);
	}

	/**
	 * 尝试获取写锁(Sync的获取写锁方法)
	 */
	public boolean tryLock( ) {
		return sync.tryWriteLock();
	}

	/**
	 * 在给定时间内当前线程没有被中断,如果写锁没有被另一个线程持有
	 * 则获取写锁
	 */
	public boolean tryLock(long timeout, TimeUnit unit)
			throws InterruptedException {
		return sync.tryAcquireNanos(1, unit.toNanos(timeout));
	}

	/**
	 * 尝试释放写锁
	 */
	public void unlock() {
		sync.release(1);
	}

	/**
	 * 返回实例的条件等待队列
	 */
	public Condition newCondition() {
		return sync.newCondition();
	}

	/**
	 * 返回写锁的字符串及其锁定状态值
	 */
	public String toString() {
		Thread o = sync.getOwner();
		return super.toString() + ((o == null) ?
								   "[Unlocked]" :
								   "[Locked by thread " + o.getName() + "]");
	}

	/**
	 * 查询当前线程是否持有写锁定
	 */
	public boolean isHeldByCurrentThread() {
		return sync.isHeldExclusively();
	}

	/**
	 * 通过当前查询写锁的持有数
	 * @return 当前线程对锁的持有数或 如果当前线程为保持锁定则返回零
	 */
	public int getHoldCount() {
		return sync.getWriteHoldCount();
	}
}

private static final long serialVersionUID = -6992448646407690164L;
/** 内部类提供的读锁 */
private final ReentrantReadWriteLock.ReadLock readerLock;
/** 内部类提供的写锁 */
private final ReentrantReadWriteLock.WriteLock writerLock;
/** 同步队列,下方所有同步机制都是由此执行 */
final Sync sync;

// Unsafe机制
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);
}

public class ReadWrite {
    private Map<String, Object> map = new HashMap<>();
    private ReadWriteLock rwl = new ReentrantReadWriteLock();

    private Lock r = rwl.readLock();
    private Lock w = rwl.writeLock();

    public Object get(String key) {
        r.lock();
        System.out.println(Thread.currentThread().getName()+" 读操作正在执行...");
        try {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return map.get(key);
        } finally {
            r.unlock();
            System.out.println(Thread.currentThread().getName()+" 读操作执行完毕...");
        }
    }

    public void put(String key, Object value) {
        w.lock();
        try {
            System.out.println(Thread.currentThread().getName()+" 写操作正在执行...");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            map.put(key, value);
        } finally {
            w.unlock();
            System.out.println(Thread.currentThread().getName()+" 写操作执行完毕...");
        }
    }
}

public class ReadWriteTest {
    public static void main(String[] args) {
        final ReadWrite readWrite = new ReadWrite();

        new Thread(new Runnable() {
            @Override
            public void run() {
                readWrite.put("key1", "value1");
            }
        }, "t0").start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                //readWrite.put("key2", "value2");
                System.out.println(readWrite.get("key1"));
            }
        }, "t1").start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                readWrite.put("key3", "value3");
            }
        }, "t2").start();

    }
}