聊聊高并发（十一）实现几种自旋锁（五）

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

聊聊高并发（十一）实现几种自旋锁（五）

2024-07-30 05:07:06 223人阅读

在聊聊高并发（九）实现几种自旋锁（四）中实现的限时队列锁是一个基于链表的限时无界队列锁，它的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






聊聊高并发（十一）实现几种自旋锁（五）

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

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

聊聊高并发（十一）实现几种自旋锁（五）

看完仍有疑问？有类似问题直接问程序猿