首页 > 代码库 > 聊聊高并发(十一)实现几种自旋锁(五)

聊聊高并发(十一)实现几种自旋锁(五)

在聊聊高并发(九)实现几种自旋锁(四)中实现的限时队列锁是一个基于链表的限时无界队列锁,它的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 = http://www.mamicode.com/0;>

结果是顺序打印的,证明锁是正确的,每次只有一个线程获得了锁


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






聊聊高并发(十一)实现几种自旋锁(五)