在聊聊高并发（九）实现几种自旋锁（四）中实现的限时队列锁是一个基于链表的限时无界队列锁，它的tryLock方法支持限时操作和中断操作，无饥饿，保证了先来先服务的公平性，在多个共享状态上自旋，是低争用的。但是它的一个缺点是牺牲了空间，为了让线程可以多次使用锁，每次Lock的时候都要new
QNode，并设置给线程，而不能重复使用原来的节点。
这篇说说限时有界队列锁，它采用了有界队列，并且和ArrayLock不同，它不限制线程的个数。它的特点主要有
1. 采用有界队列，减小了空间复杂度，L把锁的空间复杂度在最坏的情况下(有界队列长度为1)是O(L)
2. 非公平，不保证先来先服务，这也是一个很常见的需求
3. 因为是有界队列，所以在高并发下存在高争用，需要结合回退锁来降低争用
它的实现思路是:
1. 采用了一个有界的等待队列，等待队列的每个节点都有多种状态，每个节点是可复用的
2. 采用了一个工作队列，Tail指针指向工作队列的队尾节点。获取和是否锁的操作是在工作队列中的节点之间进行
3. 由于是限时队列，并支持中断，所以队列中的节点都是可以退出队列的
4. 算法分为三步，第一步是线程从有界的等待队列中获得一个节点，并设置为WAITING，如果没有获得，就自旋
第二步是把这个节点加入工作队列，并获得前一个节点的指针
第三步是在前一个节点的状态上自旋，直到获得锁，并把前一个节点RELEASED状态改为FREE
节点有4种状态：
1. FREE： 表示节点可以被获得。当前一个节点释放锁，并设置状态为RELEASED的时候，后一个节点需要把前一个节点设置为FREE。当节点在没有进入工作队列时超时，也被设置为FREE.
2. RELEASED：节点释放锁时设置为RELEASED，需要后续节点把它设置为FREE。如果是工作队列的最后一个节点，那么RELEASED状态的节点在第一步时可被获得
3. WAITING：表示获得了锁或在工作队列中等待锁。是在第一步中被设置的，第一步的结果就是获得一个状态为WAITING的节点
4. ABORTED：工作队列中的节点超时或者中断的节点被设置为ABORTED。 队尾的ABORTED节点可以被第一步获得，队中的ABORTED节点不能被第一步获取，只能把它的preNode指针指向它的前一个节点，表示它自己不能被获取了
理解节点这4种状态的转变是理解这个设计的关键。这个设计比较复杂，从篇幅考虑，这篇只介绍Lock和UnLock操作，下一篇说tryLock限时操作
1. 创建枚举类型State来表示状态
2. 创建QNode表示节点，使用一个AtomicReference原子变量指向它的State，以便于支持CAS操作。节点维护一个PreNode引用，只有节点被Aborted的时候才设置这个引用的值，表示跳过这个节点
3. 一个有界的QNode队列，使用数组表示
4. MIN_BACKOFF和MAX_BACKOFF支持回退操作，单位是毫秒。这两个值依赖于硬件性能，需要通过不断测试来获取最优值
5. 一个Random随机数，来产生随即的数组下标，非公平性需要
6. 一个AtomicStampedReference类型的原子变量作为队尾指针tail。AtomicStampedReference采用了版本号来避免CAS操作的ABA问题。这很重要，因为有界等待队列的节点会多次进出工作队列，所以可能发生同一个节点被前一个线程准备CAS操作时，已经被后几个线程进出了工作队列，导致第一个线程拿到的QNode的状态不正确。
7. lock实现分为三步，上文已经说过了
8. unlock操作就是两步，第一修改状态通知其他线程获取锁。第二是设置自己的节点引用，以便下次可再次获得锁而不影响其他线程的状态。这里是把线程指向的节点状态设置为RELEASED，同时设置线程的节点引用为空，这样其他线程可以继续使用这个节点。
package com.zc.lock;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicStampedReference;
/**
* 限时有界队列锁，并且直接不限数量的线程
* 由于是有界的队列，所以争用激烈，可以复合回退锁的概念，减少高争用
* 分为三步:
* 第一步是取得一个State为FREE的节点，设置为WAITING
* 第二步是把这个节点加入队列，获取前一个节点
* 第三步是在前一个节点上自旋
*
* 优点是L个锁的空间复杂度是O(L)，而限时无界队列锁的空间复杂度为O(Ln)
* **/
public class CompositeLock implements TryLock{
enum State {FREE, WAITING, RELEASED, ABORTED}
class QNode{
AtomicReference<State> state = new AtomicReference<CompositeLock.State>(State.FREE);
volatile QNode preNode;
}
private final int SIZE = 10;
private final int MIN_BACKOFF = 1;
private final int MAX_BACKOFF = 10;
private Random random = new Random();
// 有界的QNode数组，表示队列总共可以使用的节点数
private QNode[] waitings = new QNode[10];
// 指向队尾节点，使用AtomicStampedReference带版本号的原子引用变量，可以防止ABA问题，因为这个算法实现需要对同一个Node多次进出队列
private AtomicStampedReference<QNode> tail = new AtomicStampedReference<CompositeLock.QNode>(null, 0);
// 每个线程维护一个QNode引用
private ThreadLocal<QNode> myNode = new ThreadLocal<CompositeLock.QNode>(){
public QNode initialValue(){
return null;
}
};
public CompositeLock(){
for(int i = 0; i < SIZE; i ++){
waitings[i] = new QNode();
}
}
@Override
public void lock() {
Backoff backoff = new Backoff(MIN_BACKOFF, MAX_BACKOFF);
QNode node = waitings[random.nextInt(SIZE)];
// 第一步: 先获得数组里的一个Node，并把它的状态设置为WAITING，否则就自旋
GETNODE:
while(true){
while(node.state.get() != State.FREE){
// 因为释放锁时只是设置了State为RELEASED,由后继的线程来设置RELEASED为FREE
// 如果该节点已经是队尾节点了并且是RELEASED，那么可以直接可以被使用
// 获取当前原子引用变量的版本号
int[] currentStamp = new int[1];
QNode tailNode = tail.get(currentStamp);
if(tailNode == node && tailNode.state.get() == State.RELEASED){
if(tail.compareAndSet(tailNode, null, currentStamp[0], currentStamp[0] + 1)){
node.state.set(State.WAITING);
break GETNODE;
}
}
}
if(node.state.compareAndSet(State.FREE, State.WAITING)){
break;
}
try {
backoff.backoff();
} catch (InterruptedException e) {
throw new RuntimeException("Thread interrupted, stop to get the lock");
}
}
// 第二步加入队列
int[] currentStamp = new int[1];
QNode preTailNode = null;
do{
preTailNode = tail.get(currentStamp);
}
// 如果没加入队列，就一直自旋
while(!tail.compareAndSet(preTailNode, node, currentStamp[0], currentStamp[0] + 1));
// 第三步在前一个节点自旋，如果前一个节点为null，证明是第一个加入队列的节点
if(preTailNode != null){
// 在前一个节点的状态自旋
while(preTailNode.state.get() != State.RELEASED){}
// 设置前一个节点的状态为FREE,可以被其他线程使用
preTailNode.state.set(State.FREE);
}
// 将线程的myNode指向获得锁的node
myNode.set(node);
return;
}
@Override
public void unlock() {
QNode node = myNode.get();
node.state.set(State.RELEASED);
myNode.set(null);
}
@Override
public boolean trylock(long time, TimeUnit unit)
throws InterruptedException {
// TODO Auto-generated method stub
return false;
}
}
采用我们之前的验证锁正确性的测试用例来测试lock, unlock操作。
package com.zc.lock;
public class Main {
//private static Lock lock = new TimeCost(new ArrayLock(150));
private static Lock lock = new CompositeLock();
//private static TimeCost timeCost = new TimeCost(new TTASLock());
private static volatile int value = 0;
public static void method(){
lock.lock();
System.out.println("Value: " + ++value);
lock.unlock();
}
public static void main(String[] args) {
for(int i = 0; i < 50; i ++){
Thread t = new Thread(new Runnable(){
@Override
public void run() {
method();
}
});
t.start();
}
}
}
结果是顺序打印的，证明锁是正确的，每次只有一个线程获得了锁
Value: 1 Value: 2 Value: 3 Value: 4 Value: 5 Value: 6 Value: 7 Value: 8 Value: 9 Value: 10 Value: 11 Value: 12 Value: 13 Value: 14 Value: 15 Value: 16 Value: 17 Value: 18 Value: 19 Value: 20 Value: 21 Value: 22 Value: 23 Value: 24 Value: 25 Value: 26 Value: 27 Value: 28 Value: 29 Value: 30 Value: 31 Value: 32 Value: 33 Value: 34 Value: 35 Value: 36 Value: 37 Value: 38 Value: 39 Value: 40 Value: 41 Value: 42 Value: 43 Value: 44 Value: 45 Value: 46 Value: 47 Value: 48 Value: 49