05.JUC核心篇：CLH锁和MCS锁

字数：2k字 | 预计阅读时长：7分钟

2021-04-06

无论是简单的非公平自旋锁还是公平的基于排队的自旋锁，由于执行线程均在同一个共享变量上自旋，申请和释放锁的时候必须对该共享变量进行修改，这将导致所有参与排队自旋锁操作的处理器的缓存变得无效。如果排队自旋锁竞争比较激烈的话，平凡的缓存同步操作会导致繁重的系统总线和内存的流量，从而大大降低了系统整体的性能。

所以需要有一种办法能够让执行线程不再在同一个共享变量上自旋，避免过高频率的缓存同步操作，于是CLH锁和MCS锁应用诞生了。

CLH锁

CLH锁其实就是一种基于逻辑队列非线程饥饿的一种自旋公平锁，由于是Craig、Landin 和 Hagersten三位大佬的发明，因此命名为CLH锁。

CLH锁原理

首先有一个尾节点指针，通过这个尾节点指针来构建等待线程的逻辑队列，因此能确保线程先到先服务的公平性，因此尾指针可以说是构建逻辑的桥梁；此外这个尾节点指针是原子引用类型，避免了多线程并发操作的线程安全性问题。
通过等待锁的每个线程在自己的某个变量上自旋等待，这个变量将由前一个线程写入。由于某个线程获取锁操作时总是通过尾节点指针获取到前一线程写入的变量，而尾节点指针又是原子引用类型，因此确保了这个变量获取出来总是线程安全的。

CLH锁数据结构

简单来说就是：tailNode一直指向最后一个线程的当前节点，而preNode指向前一个线程的curNode节点，获取锁的时候会判断前一个线程的locked标识是否为false，为false则自己获取锁，如果前一个节点的locked为true则自旋等待获取锁。

当由新节点插入的时候：新节点的preNode指向当前的tailNode指向的节点，而tailNode指向新插入的节点。

CLH锁代码实现

public class CLHLock {
    /**
     * CLH锁节点
     */
    private static class CLHNode {
        // 锁状态：默认为false，表示线程没有获取到锁；true表示线程获取到锁或正在等待
        // 为了保证locked状态是线程间可见的，因此用volatile关键字修饰
        volatile boolean locked = false;
    }
    // 尾结点，总是指向最后一个CLHNode节点
    // 【注意】这里用了java的原子系列之AtomicReference，能保证原子更新
    private final AtomicReference<CLHNode> tailNode;
    // 当前节点的前驱节点
    private final ThreadLocal<CLHNode> predNode;
    // 当前节点
    private final ThreadLocal<CLHNode> curNode;

    // CLHLock构造函数，用于新建CLH锁节点时做一些初始化逻辑
    public CLHLock() {
        // 初始化时尾结点指向一个空的CLH节点
        tailNode = new AtomicReference<>(new CLHNode());
        // 初始化当前的CLH节点
        curNode = new ThreadLocal() {
            @Override
            protected CLHNode initialValue() {
                return new CLHNode();
            }
        };
        // 初始化前继节点，注意此时前继节点没有存储CLHNode对象，存储的是null
        predNode = new ThreadLocal();
    }

    /**
     * 获取锁
     */
    public void lock() {
        // 取出当前线程ThreadLocal存储的当前节点，初始化值总是一个新建的CLHNode，locked状态为false。
        CLHNode currNode = curNode.get();
        // 此时把lock状态置为true，表示一个有效状态，
        // 即获取到了锁或正在等待锁的状态
        currNode.locked = true;
        // 当一个线程到来时，总是将尾结点取出来赋值给当前线程的前继节点；
        // 然后再把当前线程的当前节点赋值给尾节点
        // 【注意】在多线程并发情况下，这里通过AtomicReference类能防止并发问题
        // 【注意】哪个线程先执行到这里就会先执行predNode.set(preNode);语句，因此构建了一条逻辑线程等待链
        // 这条链避免了线程饥饿现象发生
        CLHNode preNode = tailNode.getAndSet(currNode);
        // 将刚获取的尾结点（前一线程的当前节点）付给当前线程的前继节点ThreadLocal
        // 【思考】这句代码也可以去掉吗，如果去掉有影响吗？
        predNode.set(preNode);
        // 【1】若前继节点的locked状态为false，则表示获取到了锁，不用自旋等待；
        // 【2】若前继节点的locked状态为true，则表示前一线程获取到了锁或者正在等待，自旋等待
        while (preNode.locked) {
            System.out.println("线程" + Thread.currentThread().getName() + "没能获取到锁，进行自旋等待。。。");
        }
        // 能执行到这里，说明当前线程获取到了锁
        System.out.println("线程" + Thread.currentThread().getName() + "获取到了锁！！！");
    }

    /**
     * 释放锁
     */
    public void unLock() {
        // 获取当前线程的当前节点
        CLHNode node = curNode.get();
        // 进行解锁操作
        // 这里将locked至为false，此时执行了lock方法正在自旋等待的后继节点将会获取到锁
        // 【注意】而不是所有正在自旋等待的线程去并发竞争锁
        node.locked = false;
        System.out.println("线程" + Thread.currentThread().getName() + "释放了锁！！！");
        // 小伙伴们可以思考下，下面两句代码的作用是什么？？
        CLHNode newCurNode = new CLHNode();
        curNode.set(newCurNode);

        // 【优化】能提高GC效率和节省内存空间，请思考：这是为什么？
        // curNode.set(predNode.get());
    }
}

MCS锁

MCS自旋锁是一种基于单向链表的高性能、公平的自旋锁，申请加锁的线程只需要在本地变量上自旋，直接前驱负责通知其结束自旋，从而极大的减少了不必要的处理器缓存同步的次数，降低了总线和内存的开销。

原理

MCS锁数据结构

每个线程持有一个自己的node，node有一个locked属性，true表示等待获取锁，false表示可以获取到锁，并且持有下一个node（后继者）的引用（可能存在）。
线程在轮询自己node的locked状态，true表示锁被其他线程占用，等待获取锁，自旋。
线程释放锁的时候，修改后继者（nextNode）的locked属性，通知后继结束自旋。

代码实现

public class MCSLock {
    public static class MCSNode {
        // 持有后继者的引用
        MCSNode next;
        // 默认是在等待锁
        boolean locked = true;
    }

    volatile MCSNode tail;// 指向最后一个申请锁的MCSNode
    private static final AtomicReferenceFieldUpdater<MCSLock, MCSNode> UPDATER = AtomicReferenceFieldUpdater
            .newUpdater(MCSLock.class, MCSNode.class, "tail");

    public void lock(MCSNode currentThreadMcsNode) {
        // 更新tial为最新加入的线程节点，并取出之前的节点（也就是前驱）
        MCSNode predecessor = UPDATER.getAndSet(this, currentThreadMcsNode);// step4
        // 前驱为空表示没有线程占用锁
        if (predecessor != null) {
            // 将当前节点设置为前驱节点的后继者
            predecessor.next = currentThreadMcsNode;//step5
            // 轮询自己的isLocked属性
            while (currentThreadMcsNode.locked) {

            }
        }
    }

    public void unlock(MCSNode currentThreadMcsNode) {
        // UPDATER.get(this) 获取最后加入的线程的node
        // 如果获取到的最后加入的node和当前node（currentThreadMcsNode）不相同，表示还有其他线程等待锁，直接修改后继者的isLocked属性。
        // 相同代表当前没其他有线程等待锁，进入下面的处理
        if (UPDATER.get(this) == currentThreadMcsNode) {//step1
            // 这个时候可能会有其他线程又加入了进来，检查时候有人排在自己后面，currentThreadMcsNode.next 表示依然没有染排在自己后面
            if (currentThreadMcsNode.next == null) { //step2
                // 将tail设置为空，如果返回true设置成功，如果返回false，表示设置失败（其他线程加入了进来，使得当前tail持有的节点不等于currentThreadMcsNode）
                if (UPDATER.compareAndSet(this, currentThreadMcsNode, null)) {// //step3
                    // 设置成功返回，没有其他线程等待锁
                    return;
                } else {
                    // 突然有其他线程加入，需要检测后继者是否有值，因为：step4执行完后，step5可能还没执行完
                    while (currentThreadMcsNode.next == null) {
                    }
                }
            }
            // 修改后继者的isLocked,通知后继者结束自旋
            currentThreadMcsNode.next.locked = false;
            currentThreadMcsNode.next = null;// for GC
        }
    }
}