聊聊高并发（八）实现几种自旋锁（三）

在聊聊高并发（七）实现几种自旋锁（二） 这篇中介绍了两种队列锁，一种是有界队列锁，一种是无界队列锁。其中无界队列锁CLHLock采用了链表的方式来组织多线程，使用了两个ThreadLocal做指针指向自身的node和前一个node。它的特点是在前一个node的lock状态自旋，当前一个node的锁释放时，会自动通知下一个线程去获得锁。

CLHLock是无饥饿的，保证先来先服务公平性，只有少量的缓存一致性流量，在SMP系统结构中，是一种比较完善的锁。但是在没有cache的NUMA系统架构中，由于在前一个节点的lock状态上自旋，NUMA架构中处理器访问本地内存的速度高于通过网络访问其他节点的内存，所以CLHLock在NUMA架构上不是最优的自旋锁。

这篇介绍一种适合在无cache的NUMA系统架构中比较完善的队列锁MCSLock。它的特点是:

1. 使用1个ThreadLocal指针来做链表，由QNode自身维护下一个节点的指针

2. 线程在自身节点自旋，而不是CLHLock那样在前一个节点自旋

3. 在释放锁时需要判断是否是唯一节点，需要做一次CAS操作，如果不是唯一节点，要稍微等待链表关系的建立

package com.zc.lock;
 
import java.util.concurrent.atomic.AtomicReference;
 
/**
 * 无界队列锁，使用一个链表来组织线程
 * 假设L把锁，n个线程，那么锁的空间复杂度为O(L+n)
 * **/
public class MCSLock implements Lock{
    // 原子变量指向队尾
    private AtomicReference<QNode> tail;
    // 两个指针，一个指向自己的Node,一个指向前一个Node
    ThreadLocal<QNode> myNode;
    
    public MCSLock(){
        tail = new AtomicReference<QNode>(null);
        myNode = new ThreadLocal<QNode>(){
            protected QNode initialValue(){
                return new QNode();
            }
        };
    }
    
    @Override
    public void lock() {
        QNode node = myNode.get();
        // CAS原子操作，保证原子性
        QNode preNode = tail.getAndSet(node);
        // 如果preNode等于空，证明是第一个获取锁的
        if(preNode != null){
            node.lock = true;
            preNode.next = node;
            // 对线程自己的node进行自旋，对无cache的NUMA系统架构来说，访问本地内存速度优于其他节点的内存
            while(node.lock){
                
            }
        }
    }
 
    @Override
    public void unlock() {
        QNode node = myNode.get();
        if(node.next == null){
            // CAS操作，判断是否没有新加入的节点
            if(tail.compareAndSet(node, null)){
                // 没有新加入的节点,直接返回
                return;
            }
            // 有新加入的节点，等待设置链关系
            while(node.next == null){
                
            }
        }
        // 通知下一个节点获取锁
        node.next.lock = false;
        // 设置next节点为空，为下次获取锁清理状态
        node.next = null;
    }
    
    public static class QNode {
        volatile boolean lock;
        volatile QNode next;
    }
    
    public String toString(){
        return "MCSLock";
    }
}
 

下面采用和上一篇同样的测试用例来测试MCSLock的正确性

package com.zc.lock;
 
public class Main {
    //private static Lock lock = new TimeCost(new ArrayLock(150));
    
    private static Lock lock = new MCSLock();
    
    //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();
        }
    }
 
}

测试结果：顺序地打印出volatile变量++的结果，证明同一时刻只有一个线程在做volatile++操作，证明加锁成功。

Value: 1Value: 2Value: 3Value: 4Value: 5Value: 6Value: 7Value: 8Value: 9Value: 10Value: 11Value: 12Value: 13Value: 14Value: 15Value: 16Value: 17Value: 18Value: 19Value: 20Value: 21Value: 22Value: 23Value: 24Value: 25Value: 26Value: 27Value: 28Value: 29Value: 30Value: 31Value: 32Value: 33Value: 34Value: 35Value: 36Value: 37Value: 38Value: 39Value: 40Value: 41Value: 42Value: 43Value: 44Value: 45Value: 46Value: 47Value: 48Value: 49Value: 50