多线程基础(二十一)：StampedLock源码分析
1.类结构及其注释
1.1 类结构
StampedLock是在java8中引入的一个新的并发工具，主要为了解决此前java7中Lock实现上的一些问题，如ReentrantReadWriteLock的锁饥饿问题。StampedLock是一个全新的Lock，其内部通过ReadWriteLockView、ReadLockView、writeLockView这三个内部类分别实现了Lock和ReadWriteLock接口。而WNode节点，则是一个链表的数据结构，其目的在于实现StanpedLock的CLH锁。
1.2 注释
一种基于容量、具有三种读/写控制mode的锁。StampLock由version和mode组成，锁获取的方法返回一个表示控制相对于锁状态的时间戳，这些方法的尝试版本可能返回特殊值0，以表示无法获取访问权限。锁的释放和转换需要使用stamp做为参数，如果这些stamps与锁的状态不匹配，则会失败。 三种模式分别如下：
- Writing模式：writeLock方法可能会阻止等待独占访问，返回可以再unlockWrite中使用的stamp，以释放锁。还提供了tryWriteLock的定时和非定时版本。当锁保持在写模式的时候，将无法获得任何读锁，并且所有乐观读验证都将失败。
- Reading模式：方法readLock可能会阻止等待的非排他性访问，从而返回可以在方法unlockRead中使用的stamp以释放锁。还提供了tryReadLock定时和非定时版本。
- Optimistic Reading模式：tryOptimisticRead仅当锁未处于写模式时才返回非零的stamp。如果自获取给定标记以来，未在写入模式下获得锁。则方法validate返回true。可以将此模式视为读锁的极弱版本，编写者可以随时将其破坏。对短的只读代码段使用乐观模式通常可以减少争用并提高吞吐量。但是，这种用法是非常脆弱的。乐观的读取部分应仅读取字段并将其保存在局部变量中，以供验证后使用。在乐观模式下读取的字段可能完全不一致，因此仅当您足够熟悉数据表示以检查一致性和/或重复调用方法 validate()时，用法才适用。例如，当首先读取对象或数组引用，然后访问其字段，元素或方法之一时，通常需要执行这些步骤。 此类还支持有条件的三种模式之间提供转换方法，如tryConvertToWriteLock尝试将锁升级。如果存在如下情况：
- 已处于写模式
- 处于读模式且没有其他读操作
- 处于乐观模式
则返回有效的写入stamo并且锁可用。这些方法的形式旨在帮助减少在基于重试的设计中原本会发生的某些代码膨胀。
StampedLocks设计为在开发线程安全组件时用作内部实用程序。它们的使用取决于对它们所保护的数据，对象和方法的内部属性的了解。它们不是可重入的，因此锁定的主体不应调用可能尝试重新获取锁的其他未知方法（尽管您可以将戳记传递给可以使用或转换它的其他方法）。读取锁定模式的使用依赖于相关的代码段无副作用。未经验证的乐观阅读节无法调用未知的方法来容忍潜在的不一致。stamp 使用有限表示，并且不是加密安全的（即有效的stamp可能是可猜测的）。stamp的价值可能会在（不超过）连续运行一年后回收。未经使用或验证而持有的stamp超过此期限可能无法正确验证。StampedLocks是可序列化的，但始终反序列化为初始解锁状态，因此它们对于远程锁定没有意义。 StampedLock的调度策略并不能始终如一地偏向读而不是写，反之亦然。所有“try”方法都是尽力而为，不一定符合任何调度或公平性策略。任何用于获取或转换锁的“try”方法的零返回值都不会携带有关锁状态的任何信息。随后的调用可能会成功。 由于它支持跨多种锁定模式的协调使用，因此此类不会直接实现Lock或ReadWriteLock接口。但是，在仅需要相关功能集的应用程序中，可以在asReadLock()、asWriteLock()或者asReadWriteLock()方法中进行转换。
用法示例： 下面说明了维护简单二维点的类中的一些用法惯用法。该示例代码说明了一些try /catch约定，即使此处严格不要求使用它们，因为它们的主体中不会出现异常。
class Point {
private double x, y;
private final StampedLock sl = new StampedLock();
void move(double deltaX, double deltaY) { // an exclusively locked method
long stamp = sl.writeLock();
try {
x += deltaX;
y += deltaY;
} finally {
sl.unlockWrite(stamp);
}
}
double distanceFromOrigin() { // A read-only method
long stamp = sl.tryOptimisticRead();
double currentX = x, currentY = y;
if (!sl.validate(stamp)) {
stamp = sl.readLock();
try {
currentX = x;
currentY = y;
} finally {
sl.unlockRead(stamp);
}
}
return Math.sqrt(currentX * currentX + currentY currentY);
}
void moveIfAtOrigin(double newX, double newY) { // upgrade
// Could instead start with optimistic, not read mode
long stamp = sl.readLock();
try {
while (x == 0.0 && y == 0.0) {
long ws = sl.tryConvertToWriteLock(stamp);
if (ws != 0L) {
stamp = ws;
x = newX;
y = newY;
break;
}
else {
sl.unlockRead(stamp);
stamp = sl.writeLock();
}
}
} finally {
sl.unlock(stamp);
}
}
}}2.成员变量及常量
2.1 算法说明
该设计采用了顺序锁的元素，在linux内核中使用，参见http://www.lameter.com/gelato2005.pdf 。或者http://www.hpl.hp.com/techreports/2012/HPL-2012-68.html。顺序的RW锁，请参考http://dl.acm.org/citation.cfm?id=2312015。 从概念上来说，锁的主要状态包含一个序列号，该序列号在写锁定的时候是奇数，但是，当读锁定的时候，读计数器将偏移非0值。当以乐观的seqlock-reader方式获得读锁的时候，将会忽略读计数器。因为我们必须为读取器使用少量有限的位数（当前为7），所以当读取器的数量超过count字段时，将使用补充读取器溢出字。为此，我们将最大读取器计数值（RBITS）视为保护溢出更新的自旋锁。 Waiters使用中使用的CLH锁的修改形式AbstractQueuedSynchronizer。有关AQS更详细的说， 参见其内部描述。其中每个节点都被标记为读或者写，等待读操作的集合被分组在一个公共节点filed cowait下，因此就大多数CLH机制而言，充当当个节点。由于队列结构，等待节点不需要实际携带序列号，我们知道每一个节点都比他的前一个节点大，这将调度策略简化为一个主要的FIFO方案，该方案包含了相位公平锁的元素（尤其参见Brandenburg&Anderson http://www.cs.unc.edu/~bbb/diss/） 特别地，我们使用阶段公平反讨价还价规则：如果一个传入的读操作在读锁被保持的情况下到达，但有一个排队的写操作，则该传入读操作将排队。（这个规则对方法获取者ead的一些复杂性负责，但是没有它，锁就会变得非常不公平）。方法发布本身不会（有时也不能）唤醒cowaiter。这是由主线程来完成的，但是在其他线程的帮助下，在方法acquireRead和acquireWrite中没有什么比这更好的了。 这些规则适用于实际排队的线程。所有tryLock的操作都会不顾一切的尝试获取锁，而不考虑偏向规则，因此可能会闯进来，在acquire方法中使用随机旋转来减少（越来越昂贵）上下文切换，同时也避免了许多线程之间的持续内存震荡。我们将旋转限制在队列的头部。线程旋转在阻塞之前等待最多旋转次数（每次迭代以50%的概率减少自旋计数）。如果在唤醒时，它无法获得锁，并且仍然（或成为）第一个等待线程（这表明其他线程阻塞并获得了锁），则它将提升旋转（最多旋转MAX_HEAD_spins），以减少继续丢失到barging线程的可能性。 几乎所有这些机制都是在acquireWrite和acquireRead方法中实现的，这是这类代码的典型特征，因为操作和重试依赖于一致的本地缓存读取集。 正如Boehm的论文（上面）所指出的，序列验证（主要是方法validate（））需要比普通易失性读取（state）更严格的排序规则。为了在验证之前强制对读取进行排序，以及在尚未强制执行的情况下强制执行验证本身，我们使用不安全装载围栏. 内存布局将锁状态和队列指针保持在一起（通常在同一缓存线上）。这通常适用于以读为主的加载。在大多数其他情况下，自适应spin-CLH锁减少内存争用的自然趋势会降低进一步扩展争用位置的动机，但可能会受到未来改进的影响。
2.2 常量
与CPU有关的常量如下：
//通过Runtime获得系统可用的CPU核数
/** Number of processors, for spin control */
private static final int NCPU = Runtime.getRuntime().availableProcessors();
/** Maximum number of retries before enqueuing on acquisition */
//在acquisition enque之前的最大重试次数
private static final int SPINS = (NCPU > 1) ? 1 << 6 : 0;
/** Maximum number of retries before blocking at head on acquisition */
//在acquisition block之前的最大重试次数
private static final int HEAD_SPINS = (NCPU > 1) ? 1 << 10 : 0;
/** Maximum number of retries before re-blocking */
//re-block之前的最大重试次数
private static final int MAX_HEAD_SPINS = (NCPU > 1) ? 1 << 16 : 0;
/** The period for yielding when waiting for overflow spinlock */
//等待自旋锁溢出 yield之前的周期 必须是2的幂-1
private static final int OVERFLOW_YIELD_RATE = 7; // must be power 2 - 1
与LG_READERS相关的常量：
/** The number of bits to use for reader count before overflowing */
//在reader 计数器overflowing之前的的最大字节数
private static final int LG_READERS = 7;
// Values for lock state and stamp operations
//锁状态和stamp操作值
//RUNIT状态
private static final long RUNIT = 1L;
//WBIT状态
private static final long WBIT = 1L << LG_READERS;
//RBITS状态
private static final long RBITS = WBIT - 1L;
//RFULL状态
private static final long RFULL = RBITS - 1L;
//ABITS状态
private static final long ABITS = RBITS | WBIT;
//SBITS状态
private static final long SBITS = ~RBITS; // note overlap with ABITS
// Initial value for lock state; avoid failure value zero
//初始化lock的状态，避免失败的0值
private static final long ORIGIN = WBIT << 1;
其他常量：
// Special value from cancelled acquire methods so caller can throw IE
//被取消的acquire方法的特殊值，以便抛出IE
private static final long INTERRUPTED = 1L;
// Values for node status; order matters
//节点的状态值，顺序性
//等待
private static final int WAITING = -1;
//取消
private static final int CANCELLED = 1;
// Modes for nodes (int not boolean to allow arithmetic)
//节点的模式，int并不代表boolean，此处允许算术计算
private static final int RMODE = 0;
private static final int WMODE = 1;2.3 变量
//CLH队列的head指针
private transient volatile WNode whead;
//CLH队列的尾指针
private transient volatile WNode wtail;
//视图 可以分为读锁、写锁、读写锁
transient ReadLockView readLockView;
transient WriteLockView writeLockView;
transient ReadWriteLockView readWriteLockView;
//锁序列和状态
private transient volatile long state;
//当读的count达到饱和的时候，额外的允许的读次数
private transient int readerOverflow;2.4 等待节点基本组成Node
组成等待队列的基本节点。本质是一个双向链表。
/** Wait nodes */
static final class WNode {
//指向链表上的前一个节点
volatile WNode prev;
//指向链表上的下一个节点
volatile WNode next;
//指向readers
volatile WNode cowait; // list of linked readers
//此处指向需要parked的线程，以便进行unpark
volatile Thread thread; // non-null while possibly parked
//基本的状态status
volatile int status; // 0, WAITING, or CANCELLED
//模式 指向存储的读或者写的模式 0表示读，1表示写
final int mode; // RMODE or WMODE
WNode(int m, WNode p) { mode = m; prev = p; }
}其结构如下：
通过这个结构，构成一个等待队列。
3.构造函数
StamepedLock只有一个构造函数：
public StampedLock() {
state = ORIGIN;
}该方法只会将state初始化为前面设置的ORIGIN状态。
4.关键实现方法
4.1 write相关方法
4.1.1 writeLock
在stampedLock中，write是通过独占的方式来获取锁。之后返回一个long类型的stamped戳。
public long writeLock() {
//定义stamp的next以及state
long s, next; // bypass acquireWrite in fully unlocked case only
//将s赋值为state，在一开始这个状态为ORIGN，通过将STATE与ABITS取& ，判断是否为0，
//如果为0，且此时通过cas操作，在当前对象的STATE偏移量处的state值如果为s,则将其修改为 next，且next的值为s+WBIT
//如果上述两个条件满足，则返回next，反之则调用acquireWrite方法，尝试无限等待获取写锁，这个方法会将当前线程进入等待队列
return ((((s = state) & ABITS) == 0L &&
U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
next : acquireWrite(false, 0L));
}可以看到这个方法，关键过程分为两部，首先是判断s的状态，由于是独占锁，需要判断是否有其他人已经获取了锁，这个计算过程如下：
可以看到，如果是初始状态，没有人获取到锁的话，那么s的值永远为0l，如果写锁被获取，则next会改变，返回的值可以看到最低7位为0。 如果写锁已经被获取，那么上述的位运算操作就不可能返回0，此时返回了一个next值，这个值加上了WBITS。此时则对acquireWrite方法进行调用。 还有一种可能是，为0判断的时候满足了，但是在并发的情况下，如果有其他的线程抢到了写锁，那么将会导致cas处更新值失败，因此也会进入acquireWrite方法。
4.1.2 tryWriteLock
此方法尝试获取排他的读锁，如果锁可用，则返回next，反之则返回0。可以根据返回值的状态来判断是否已经获取到了锁，这个方法不会将当前线程进入wait队列。
public long tryWriteLock() {
//定义s和next
long s, next;
//s为state，如果其值&ABITS为0L,且cas能成功，则获取到锁，返回next，反之则返回0
return ((((s = state) & ABITS) == 0L &&
U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
next : 0L);
}可以看到tryWriteLock的方法与writeLock方法最大的区别就是，tryWriteLock如果没有获取到锁会立即返回，stamp为0。而writeLock则不会，会被阻塞，进入Wait队列。
4.1.3 tryWriteLock(long time, TimeUnit unit)
支持超时的tryWriteLock方法。传入了time和timeUnit参数。 同时，此方法也支持InterruptedException异常。
public long tryWriteLock(long time, TimeUnit unit)
throws InterruptedException {
//将传入的time转为纳秒
long nanos = unit.toNanos(time);
//判断线程是否被中断，如果没有中断，则执行下面的逻辑
if (!Thread.interrupted()) {
//定义 next和deadline
long next, deadline;
//通过tryWriteLock方法来尝试获取锁，如果不为0则获取成功，返回next即可
if ((next = tryWriteLock()) != 0L)
return next;
//判断纳秒值的合法性
if (nanos <= 0L)
return 0L;
//通过deadline来计算是否触发阈值
if ((deadline = System.nanoTime() + nanos) == 0L)
deadline = 1L;
//调用acquireWrite方法，进入等待队列
if ((next = acquireWrite(true, deadline)) != INTERRUPTED)
return next;
}
throw new InterruptedException();
}此方法增加了超时时间，在规定的时间内，尝试获取锁，如果一次没有获取到，那么就调用acquireWrite方法，进入阻塞队列，通过设置超时时间来实现。 此方法还支持InterruptedException异常来响应中断。
4.1.4 writeLockInterruptibly
public long writeLockInterruptibly() throws InterruptedException {
long next;
//如果线程不是重点状态，且next通过acquireWrite方法无限等待的情况下的状态不为INTERRUPTED状态，则返回next,反之，此处抛出中断异常。
if (!Thread.interrupted() &&
(next = acquireWrite(true, 0L)) != INTERRUPTED)
return next;
throw new InterruptedException();
}此方法，是获取写锁的Inturrept的处理方法。调用acquireWrite方法，这个方法实际上不会抛出Inturrupt异常，如果在这个方法执行的过程中发现被中断，则只会返回一个值INTERRUPTED。
4.1.5 acquireWrite
这是获取读锁的核心方法。其代码如下：
private long acquireWrite(boolean interruptible, long deadline) {
//定义node和p
WNode node = null, p;
//spins初始化为-1 用于记录自旋次数
for (int spins = -1;;) { // spin while enqueuing
//定义m s ns
long m, s, ns;
//m为记录state的变量，用m与ABITS& 判断读锁是否可以独占，以此来判断是否可以获取读锁
if ((m = (s = state) & ABITS) == 0L) {
//如果可以，采用cas的方式更新state并返回
if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))
return ns;
}
//反之如果自旋次数小于0
else if (spins < 0)
//state为WBIT且等待队列中只有一个元素，则说明等待队列已被初始化，则spins=SPINS，反之则为0，需要多次自旋
spins = (m == WBIT && wtail == whead) ? SPINS : 0;
//如果spins大于0
else if (spins > 0) {
//采用LockSupport的nextSecondarySeed方法
if (LockSupport.nextSecondarySeed() >= 0)
//spins自减1
--spins;
}
//反之，如果p赋值为wtail且为null，说明等待队列中没有元素
else if ((p = wtail) == null) { // initialize queue
//初始化队列，调用WNode构造方法
WNode hd = new WNode(WMODE, null);
//采用cas方法将WHEAD改为hd
if (U.compareAndSwapObject(this, WHEAD, null, hd))
//之后wtail=hd 也就是说等待队列初始化之后，wtail的值为hd
wtail = hd;
}
//如果node为null
else if (node == null)
//new一个Node，状态为WMODE，p
node = new WNode(WMODE, p);
//如果node的前一个节点不是p
else if (node.prev != p)
//将prev指向p
node.prev = p;
//cas方法将WTAIL由p改为node
else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
//将next指向node并break
p.next = node;
break;
}
}
//前面这个循环的目的是判断当前节点的状态，要么获得锁，要么进入等待队列。
//继续循环，将等待队列中的队首元素判断能否获得锁
//这个地方满足FIFO
for (int spins = -1;;) {
WNode h, np, pp; int ps;
//h为whead 为p
if ((h = whead) == p) {
//如果spins小于0
if (spins < 0)
//spins的值赋值为HEAD_SPINS
spins = HEAD_SPINS;
//如果spins比MAX_HEAD_SPINS小
else if (spins < MAX_HEAD_SPINS)
//spins左移1位
spins <<= 1;
//内层循环 k = spins
for (int k = spins;;) { // spin at head
long s, ns;
//如果此时s为0说明可以获得锁
if (((s = state) & ABITS) == 0L) {
//cas的方式，修改STATE
if (U.compareAndSwapLong(this, STATE, s,
ns = s + WBIT)) {
//此时队列中的元素获得锁，并出队
whead = node;
node.prev = null;
return ns;
}
}
//将k自减并break
else if (LockSupport.nextSecondarySeed() >= 0 &&
--k <= 0)
break;
}
}
//如果h不为null 将协助释放waiters
else if (h != null) { // help release stale waiters
WNode c; Thread w;
//c赋值为h,cowait 判断是否为null
while ((c = h.cowait) != null) {
//cas方式将WCOWAIT从c修改为为c.cowait
if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
//此时如果线程不为null，则对该线程执行unpark
(w = c.thread) != null)
U.unpark(w);
}
}
//如果whead为h 说明h为等待对立的队首
if (whead == h) {
//np赋值为node.prev 如果不为p
if ((np = node.prev) != p) {
if (np != null)
//修改p的next指针
(p = np).next = node; // stale
}
//如果ps的状态为0
else if ((ps = p.status) == 0)
//cas方式将WSTATUS从0修改为WAITING
U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
//如果ps为CANCLED状态
else if (ps == CANCELLED) {
if ((pp = p.prev) != null) {
//将当前p节点出队
node.prev = pp;
pp.next = node;
}
}
//反之
else {
long time; // 0 argument to park means no timeout
//如果deadline为0
if (deadline == 0L)
//将等待时间改为0
time = 0L;
//反之，time为deadline减去当前的纳秒数
else if ((time = deadline - System.nanoTime()) <= 0L)
//如果小于0，则将等待的节点取消
return cancelWaiter(node, node, false);
//获得当前线程
Thread wt = Thread.currentThread();
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
//如果p的status小于0且p不为h
if (p.status < 0 && (p != h || (state & ABITS) != 0L) &&
whead == h && node.prev == p)
U.park(false, time); // emulate
//调用park将线程阻塞
LockSupport.park
node.thread = null;
U.putObject(wt, PARKBLOCKER, null);
//如果出现中断，也会将当前的等待队列中的节点取消
if (interruptible && Thread.interrupted())
return cancelWaiter(node, node, true);
}
}
}
}这个方法中将分为两个部分，两个for循环。其中第一个for循环中，首先会尝试获取写锁，如果能获取成功则返回，如果不能获取成功，则将当前线程创建一个WNode节点添加到等待队列的尾部。之后执行第二个for循环，在第二个for循环中，队队列的队首节点进行判断，判断队首节点能否获取到锁，此外对其线程栈中的读线程进行unpark唤醒。 之后在次自旋需要将之前写入的节点的线程进行阻塞。 StampedLock根据这个类中的成员变量state的状态，来决定其是否能被获得读锁或者是写锁。
4.2 read相关的方法
4.2.1 readLock
读锁将采用非独占的模式来获得锁，必要时会阻塞。
public long readLock() {
//定义s为state，next
long s = state, next; // bypass acquireRead on common uncontended case
//如果whead和wtail相等，则说明队列中只有一个初始节点，且s&ABITS小于RFULL，则说明可以获得读锁
//采用cas的方式获得读锁，返回next为s+RUNIT
//如果上述两个条件都能满足，则返回next获得读锁，反之，无限阻塞直到获得读锁为止
return ((whead == wtail && (s & ABITS) < RFULL &&
U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?
next : acquireRead(false, 0L));
}读锁的方式与获取写锁类似，只是不采用独占的方式进行。获取读锁的三个条件，一是等待队列只有一个节点，一是s&ABITS小于RFULL，此外就是cas的方式修改成功。
可以看到，最核心的条件是需要满足等待队列中whead=wtail这个条件，但是只要有写锁进入了等待队列，这个条件都将不满足，也就是说，只要有写锁被阻塞，那么读锁一定就会阻塞。如果这个条件满足之后，那么另外一个条件则是读的最大次数以及cas的操作成功的可能性。在前面学习写锁的过程中学过，一旦写锁获得成功，那么state的状态将远大于读锁，将会造成条件二不满足。
4.2.2 tryReadLock
tryReadLock采用尝试获取读锁的方法，如果获取不成功则返回0，立即返回。如果获取成功则返回next。
public long tryReadLock() {
//死循环
for (;;) {
long s, m, next;
。。如果s&ABITS为WBIT，则说明已经有写锁，则直接返回0
if ((m = (s = state) & ABITS) == WBIT)
return 0L;
//当m小于RFULL则说明可以获取读锁
else if (m < RFULL) {
//cas的方式获取并返回next
if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
return next;
}
else if ((next = tryIncReaderOverflow(s)) != 0L)
return next;
}
}读锁获得成功之后，每次在最低位加1，而写锁是在第8位加1。
写锁一次只能获取一个，而读锁，一次可以获取多个。
4.2.3 tryReadLock(long time, TimeUnit unit)
支持超时的方式，获得读锁。此方法中同样需要对中断进行处理，抛出InterruptException。
public long tryReadLock(long time, TimeUnit unit)
throws InterruptedException {
long s, m, next, deadline;
//将time转换为纳秒
long nanos = unit.toNanos(time);
//判断线程的中断状态
if (!Thread.interrupted()) {
//如果写锁没有被获取
if ((m = (s = state) & ABITS) != WBIT) {
//且m小于RFULL
if (m < RFULL) {
//cas的方式修改state，获得读锁，返回
if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
return next;
}
//反之，调用tryIncReaderOverflow
else if ((next = tryIncReaderOverflow(s)) != 0L)
//返回next
return next;
}
//如果纳秒数小于0
if (nanos <= 0L)
return 0L;
//如果deadline为0
if ((deadline = System.nanoTime() + nanos) == 0L)
deadline = 1L;
//如果acquireRead方法状态不为INTERRUPTED
if ((next = acquireRead(true, deadline)) != INTERRUPTED)
return next;
}
//反之则抛出中断异常
throw new InterruptedException();
}采用超时的方式来获取读锁，此方法需要对线程中断进行响应。抛出异常。
4.2.4 readLockInterruptibly
非独占的方式获得读锁，并支持中断。
public long readLockInterruptibly() throws InterruptedException {
long next;
//如果线程中断
if (!Thread.interrupted() &&
//next通过无限等待的方式获取读锁
(next = acquireRead(true, 0L)) != INTERRUPTED)
return next;
//抛出异常
throw new InterruptedException();
}此方法用于无限等待的方式获取读锁。并抛出异常。
4.2.5 tryOptimisticRead
乐观读操作。
public long tryOptimisticRead() {
long s;
return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
}
可以看到，s&WBIT这个操作，只要写锁没有被获取，就一定可以返回读锁，不需要关注其他读操作的状态。此外，也不需要关心队列中是否有等待的写操作。这是一种乐观的读操作。
4.2.6 acquireRead
采用超时等待的方式来获得读锁。
private long acquireRead(boolean interruptible, long deadline) {
WNode node = null, p;
//循环
for (int spins = -1;;) {
WNode h;
//如果h==p则等待队列中只有一个初始化节点
if ((h = whead) == (p = wtail)) {
//死循环
for (long m, s, ns;;) {
//如果可以获得读锁，则采用cas的方式获得
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
(m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
return ns;
//反之，如果m大于WBIT
else if (m >= WBIT) {
//如果spins大于0 将spins自减1
if (spins > 0) {
if (LockSupport.nextSecondarySeed() >= 0)
--spins;
}
else {
//如果spins为0，则获得锁，返回
if (spins == 0) {
WNode nh = whead, np = wtail;
if ((nh == h && np == p) || (h = nh) != (p = np))
break;
}
spins = SPINS;
}
}
}
}
//如果p为空，初始化队列
if (p == null) { // initialize queue
WNode hd = new WNode(WMODE, null);
//cas的方式
if (U.compareAndSwapObject(this, WHEAD, null, hd))
wtail = hd;
}
else if (node == null)
node = new WNode(RMODE, p);
else if (h == p || p.mode != RMODE) {
if (node.prev != p)
node.prev = p;
else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
p.next = node;
break;
}
}
//如果队列不空，且队尾是读节点，则将当前节点添加到队尾节点的cowait链表中。
else if (!U.compareAndSwapObject(p, WCOWAIT,
node.cowait = p.cowait, node))
node.cowait = null;
else {
for (;;) {
WNode pp, c; Thread w;
//尝试对队列头节点进行唤醒，判断是否能获得锁，如果是读锁，则通过循环将cowait链都释放。
if ((h = whead) != null && (c = h.cowait) != null &&
U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
(w = c.thread) != null) // help release
U.unpark(w);
if (h == (pp = p.prev) || h == p || pp == null) {
long m, s, ns;
do {
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s,
ns = s + RUNIT) :
(m < WBIT &&
(ns = tryIncReaderOverflow(s)) != 0L))
return ns;
} while (m < WBIT);
}
if (whead == h && p.prev == pp) {
long time;
if (pp == null || h == p || p.status > 0) {
node = null; // throw away
break;
}
if (deadline == 0L)
time = 0L;
else if ((time = deadline - System.nanoTime()) <= 0L)
return cancelWaiter(node, p, false);
Thread wt = Thread.currentThread();
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
if ((h != pp || (state & ABITS) == WBIT) &&
whead == h && p.prev == pp)
U.park(false, time);
node.thread = null;
U.putObject(wt, PARKBLOCKER, null);
if (interruptible && Thread.interrupted())
return cancelWaiter(node, p, true);
}
}
}
}
for (int spins = -1;;) {
WNode h, np, pp; int ps;
//如果当前线程是队首，则尝试获取读锁
if ((h = whead) == p) {
if (spins < 0)
spins = HEAD_SPINS;
else if (spins < MAX_HEAD_SPINS)
spins <<= 1;
for (int k = spins;;) { // spin at head
long m, s, ns;
//判断写锁是否被占用
if ((m = (s = state) & ABITS) < RFULL ?
U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
(m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)) {
WNode c; Thread w;
whead = node;
node.prev = null;
while ((c = node.cowait) != null) {
if (U.compareAndSwapObject(node, WCOWAIT,
c, c.cowait) &&
(w = c.thread) != null)
U.unpark(w);
}
return ns;
}
else if (m >= WBIT &&
LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
break;
}
}
//如果头节点的cowait不为空，则将所有读线程唤醒。
else if (h != null) {
WNode c; Thread w;
while ((c = h.cowait) != null) {
if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
(w = c.thread) != null)
U.unpark(w);
}
}
if (whead == h) {
if ((np = node.prev) != p) {
if (np != null)
(p = np).next = node; // stale
}
else if ((ps = p.status) == 0)
U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
else if (ps == CANCELLED) {
if ((pp = p.prev) != null) {
node.prev = pp;
pp.next = node;
}
}
else {//阻塞当前读线程
long time;
if (deadline == 0L)
time = 0L;
else if ((time = deadline - System.nanoTime()) <= 0L)
return cancelWaiter(node, node, false);
Thread wt = Thread.currentThread();
U.putObject(wt, PARKBLOCKER, this);
node.thread = wt;
//如果前一个节点为waiting状态，且写锁被占用，则阻塞当前线程。
if (p.status < 0 &&
(p != h || (state & ABITS) == WBIT) &&
whead == h && node.prev == p)
U.park(false, time);
node.thread = null;
U.putObject(wt, PARKBLOCKER, null);
if (interruptible && Thread.interrupted())
return cancelWaiter(node, node, true);
}
}
}
}acquireRead的读方法会比acquireWrite方法复杂，也是分为两部分，第一部分是判断如果不能获取读锁，则将当前线程包裹为WNODE节点插入双向链表中。如果队尾是读节点，那么将这个节点插入到读节点的cowait栈中，如果不是则直接插入队尾，之后将队列前面的元素修改为WAITING状态。 第二部分则是尝试唤醒队列头部的节点来获取锁，如果能够获取成功，则返回，如果获取的是读锁，那么将其cowait栈中的全部读操作进行释放。
4.3 unlock操作
4.3.1 unlockWrite
//对write操作进行解锁
public void unlockWrite(long stamp) {
WNode h;
//如果state不为此时的stamp或者stamp&WBIT不为0，则说明参数错误，将返回异常
if (state != stamp || (stamp & WBIT) == 0L)
throw new IllegalMonitorStateException();
//如果上述条件满足，且state加WBIT如果为0，则将state重置为ORIGIN，反之则返回stamp
state = (stamp += WBIT) == 0L ? ORIGIN : stamp;
//如果h不为空且其状态不为0，则调用release方法释放该节点。
if ((h = whead) != null && h.status != 0)
release(h);
}释放写锁的时候，首先判断state是否为加写锁之后的状态，state加写锁之后，再与WBIT相加，如果溢出，则返回为0，因此重置为ORIGN,反之则返回stamp的值。
4.3.2 unlockRead
释放非排他的读锁。
public void unlockRead(long stamp) {
long s, m; WNode h;
//死循环
for (;;) {
//如果s的state&SBITS不等于stamp&SBITS
if (((s = state) & SBITS) != (stamp & SBITS) ||
(stamp & ABITS) == 0L || (m = s & ABITS) == 0L || m == WBIT)
//抛出异常
throw new IllegalMonitorStateException();
if (m < RFULL) {
//采用cas的方式，将state的值从s改为s-RUNIT
if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) {
//如果m与RUNIT相等，且h不为空且h.status不为0
if (m == RUNIT && (h = whead) != null && h.status != 0)
//释放读锁
release(h);
break;
}
}
else if (tryDecReaderOverflow(s) != 0L)
break;
}
}如果只有读锁被获取，则State&SBITS与stamp&SBITS的状态始终都是第9位为1，8位以下全部为0，这两个值相等。如果写锁被获取则这两个值不可能相等。 如果不等则抛出异常。 反之，则说明可以释放读锁，判断m是否小于RFULL。 如果小于，则释放读锁。
4.3.3 unlock
此方法可以对读锁和写锁都进行unlock。如果传入的stamp与锁的state匹配，则释放相应的锁。
public void unlock(long stamp) {
long a = stamp & ABITS, m, s; WNode h;
//此处判断是写锁
while (((s = state) & SBITS) == (stamp & SBITS)) {
//对写锁进行释放
if ((m = s & ABITS) == 0L)
break;
//如果m为WBIT
else if (m == WBIT) {
if (a != m)
break;
//同样需要处理s+WBIT溢出的情况
state = (s += WBIT) == 0L ? ORIGIN : s;
if ((h = whead) != null && h.status != 0)
release(h);
return;
}
//如果为0则直接break
else if (a == 0L || a >= WBIT)
break;
//如果m小于RFULL
else if (m < RFULL) {
//采用cas的方式将STATE改为s-RUNIT
if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) {
if (m == RUNIT && (h = whead) != null && h.status != 0)
release(h);
return;
}
}
//调用tryDecReaderOverFlow
else if (tryDecReaderOverflow(s) != 0L)
return;
}
throw new IllegalMonitorStateException();
}4.3.4 tryUnlockWrite
采用try的方式释放写锁。
public boolean tryUnlockWrite() {
long s; WNode h;
//如果s&WBIT为0 则可以释放写锁
if (((s = state) & WBIT) != 0L) {
//判断state + WBIT是否溢出，如果溢出则重置为ORIGIN,否则返回s。
state = (s += WBIT) == 0L ? ORIGIN : s;
//如果h不为空，且h.status不为0，对写锁进行释放
if ((h = whead) != null && h.status != 0)
release(h);
return true;
}
return false;
}4.3.5 tryUnlockRead
采用try的方式释放读锁。
public boolean tryUnlockRead() {
long s, m; WNode h;
while ((m = (s = state) & ABITS) != 0L && m < WBIT) {
if (m < RFULL) {
if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) {
if (m == RUNIT && (h = whead) != null && h.status != 0)
release(h);
return true;
}
}
else if (tryDecReaderOverflow(s) != 0L)
return true;
}
return false;
}4.4 锁升降级
4.4.1 tryConvertToWriteLock
将锁升级为写锁。
public long tryConvertToWriteLock(long stamp) {
//a用于判断是否为读锁
long a = stamp & ABITS, m, s, next;
//判断s的状态如果是写锁
while (((s = state) & SBITS) == (stamp & SBITS)) {
//如果无锁或者锁溢出
if ((m = s & ABITS) == 0L) {
//如果a不为0，则返回
if (a != 0L)
break;
//cas的方式将STATE的s更新为写锁
if (U.compareAndSwapLong(this, STATE, s, next = s + WBIT))
return next;
}
//如果m与WBIT相等 存在写锁的场景
else if (m == WBIT) {
//如果此时不等了则返回
if (a != m)
break;
//直接返回写锁的stamp
return stamp;
}
//如果m为RUNIT且a不为0 cas的方式将s修改为s减去RUNIT
else if (m == RUNIT && a != 0L) {
if (U.compareAndSwapLong(this, STATE, s,
next = s - RUNIT + WBIT))
return next;
}
else
break;
}
return 0L;
}这个方法可以将读锁升级为写锁，完成了锁的升级。
4.4.1 tryConvertToReadLock
此方法可以将写锁降级为读锁。
public long tryConvertToReadLock(long stamp) {
//a可以判断是否为读锁
long a = stamp & ABITS, m, s, next; WNode h;
//判断当前版stamp再写锁区域的字段没有改变。
while (((s = state) & SBITS) == (stamp & SBITS)) {
//判断m的锁是否已经消除了
if ((m = s & ABITS) == 0L) {
if (a != 0L)
break;
//如果m小于RFULL
else if (m < RFULL) {
//cas的方式获得读锁，进行转换
if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
return next;
}
//如果读锁溢出，进行溢出处理
else if ((next = tryIncReaderOverflow(s)) != 0L)
return next;
}
//如果写锁存在
else if (m == WBIT) {
//版本号不等
if (a != m)
break;
//新的版本号，是当前的版本号加上WBIT+RUNIT
state = next = s + (WBIT + RUNIT);
//判断节点的状态
if ((h = whead) != null && h.status != 0)
release(h);
return next;
}
//说明读锁存在，且不会溢出，则直接返回stamp
else if (a != 0L && a < WBIT)
return stamp;
//其他方式直接退出
else
break;
}
return 0L;
}此方法可以将写锁转换为读锁。进行降级。
4.4.2 tryConvertToOptimisticRead
转换为乐观读锁。
public long tryConvertToOptimisticRead(long stamp) {
long a = stamp & ABITS, m, s, next; WNode h;
U.loadFence();
for (;;) {
if (((s = state) & SBITS) != (stamp & SBITS))
break;
//如果无锁 或者溢出，则直接返回
if ((m = s & ABITS) == 0L) {
if (a != 0L)
break;
return s;
}
//如果为写锁
else if (m == WBIT) {
//如果版本号不等则break
if (a != m)
break;
//对写锁释放
state = next = (s += WBIT) == 0L ? ORIGIN : s;
//判断节点状态，对节点head进行释放
if ((h = whead) != null && h.status != 0)
release(h);
return next;
}
//无所或者读锁溢出，break
else if (a == 0L || a >= WBIT)
break;
//读锁
else if (m < RFULL) {
if (U.compareAndSwapLong(this, STATE, s, next = s - RUNIT)) {
if (m == RUNIT && (h = whead) != null && h.status != 0)
release(h);
return next & SBITS;
}
}
//释放或者溢出
else if ((next = tryDecReaderOverflow(s)) != 0L)
return next & SBITS;
}
return 0L;
}4.5 其他方法
4.5.1 getReadLockCount
得到读锁的次数
private int getReadLockCount(long s) {
long readers;
//判断读锁是否存在
if ((readers = s & RBITS) >= RFULL)
//readers为RFULL+readerOverflow
readers = RFULL + readerOverflow;
return (int) readers;
}读锁的次数为RFULL+readerOverflow
4.5.2 isWriteLocked
判断当前锁是否为写锁。
public boolean isWriteLocked() {
return (state & WBIT) != 0L;
}通过state与WBIT取&判断。
4.5.3 isReadLocked
判断当前锁是否为读锁
public boolean isReadLocked() {
return (state & RBITS) != 0L;
}通过state与RBITS取&判断。
5.三个视图类
StampedLock提供了三个视图类： ReadLockView,WriteLockView,ReadWriteLockView。 这三个视图类，实际上是将StampedLock的方法独立出来，分为读锁、写锁和读写锁。这与ReentrentReadWriteLock有些类似。
final class ReadLockView implements Lock {
public void lock() { readLock(); }
public void lockInterruptibly() throws InterruptedException {
readLockInterruptibly();
}
public boolean tryLock() { return tryReadLock() != 0L; }
public boolean tryLock(long time, TimeUnit unit)
throws InterruptedException {
return tryReadLock(time, unit) != 0L;
}
public void unlock() { unstampedUnlockRead(); }
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
final class WriteLockView implements Lock {
public void lock() { writeLock(); }
public void lockInterruptibly() throws InterruptedException {
writeLockInterruptibly();
}
public boolean tryLock() { return tryWriteLock() != 0L; }
public boolean tryLock(long time, TimeUnit unit)
throws InterruptedException {
return tryWriteLock(time, unit) != 0L;
}
public void unlock() { unstampedUnlockWrite(); }
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
final class ReadWriteLockView implements ReadWriteLock {
public Lock readLock() { return asReadLock(); }
public Lock writeLock() { return asWriteLock(); }
}这三个视图类可以分别都通过asReadLock、asWriteLock、asReadWriteLock方法进行转换。
6.总结
本文对StampedLock的源码及注释部分进行了分析，可以知道，StampedLock实际上没有采用AQS来实现，而是采用了与AQS类似的一种比较简单的CLH队列来实现。这与ReadWriteLock相比起结构会简单许多。但是需要注意的是，StampedLock虽然有很多有点，但是并不支持重入。后续将对stampedLock和ReentrentReadWriteLock对比分析，及其应用场景。
腾讯云开发者
