自旋锁 | 枫秀学习日志

自旋锁是什么

自旋锁与互斥量(注意这里不是互斥锁)类似，用于多线程同步的一种锁。但他不是休眠使线程阻塞，而是在获取锁之前一直处于忙等（自旋）阻塞状态。基本的实现是通过线程反复检查锁特定变量是否可用。由于线程在这一过程中保持执行，因此是一种忙等待。一旦获取了自旋锁，线程会一直保持该锁，直至显式释放自旋锁。

自旋锁通常被作为底层原语用于实现其他类型的锁。根据它们所基于与的系统体系结构，可以通过使用测试并设置指令（CAS）有效的实现。当然这里说的有效也还是会导致CPU资源的浪费：当线程自旋等待锁变为可用时，CPU不能做其它的事情，这也是自旋锁只能够持有一段时间的原因。

有了这一层了解，自旋锁的优势和劣势，以及其适用场景也就一目了然了。

优势：

没有线程阻塞，也就没有了线程上下文切换带来的开销
自旋操作更加直观，无需分析什么情况下会导致线程阻塞

劣势：

最大的问题就是由于需要一直循环检测锁的状态，因此会浪费CPU Cycles

适用场景：

结合上述的优劣，自旋锁在锁的临界区很小并且锁争抢排队不是非常严重的情况下是非常合适的：

临界区小，因此每个使用锁的线程占用锁的时间不会很长，自旋等待的线程能够快速地获取到锁。
所争抢排队不严重，因此锁的自旋时间也是可控的，不会有大量线程处于自旋等待的状态中，从而增加浪费CPU Cycles。

2. 自旋锁实现

首先需要明确的一点是，对于加锁和释放锁的操作，需要是原子性的。这是能够继续讨论的基石。对于现代处理器，一般通过CAS(Compare And Set)操作来保证原子性。它的原理其实很简单，就是将“对比-设置”这一个流程原子化，保证在符合某种预期的前提下，完成一次写操作。

对应到Java语言层面，就是那一大票的AtomicXXX类型。比如在下面的非公平自旋锁的实现中，会借助AtomicReference类型提供的CAS操作来完成加锁和释放锁的操作。

1. 简单的自旋锁

直接上代码

package JUC.spinLock;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;

/**************************************
 *      Author : zhangke
 *      Date   : 2018-12-12 19:38
 *      email  : [email protected]
 *      Desc   : 简单的自旋锁，哪个线程竞争到，
 *      则获取锁，有可能产生饥饿现象
 ***************************************/
public class SpinLock {

    //维护当前拥有锁的线程
    private AtomicReference<Thread> owner = new AtomicReference<>();


    public void lock() {
        Thread currentThread = Thread.currentThread();

        // 如果锁未被占用，则设置当前线程为锁的拥有者
        while (!owner.compareAndSet(null, currentThread)) {
            // 自旋
        }
    }


    public void unLock() {
        Thread currentThread = Thread.currentThread();
        // 只有锁的拥有者才能释放锁，其它的线程因为无法满足Compare，因此不会Set成功
        owner.compareAndSet(currentThread, null);
    }


    //一下代码属于测试
    //用于等待所有线程都已结束
    static CountDownLatch countDownLatch = new CountDownLatch(10);

    //所有线程共享的变量，用于测试，如果加锁和释放锁是正确的操作
    //则在操作此集合时不会产生ConcurrentModificationException
    static List<String> list = new ArrayList<>();


    public static void main(String[] args) throws InterruptedException {
        final SpinLock lock = new SpinLock();

        for (int i = 0; i < 10; i++) {
            new Thread(generateTask(lock, String.valueOf(i), list)).start();
        }
        countDownLatch.await();
        System.out.println(list);
    }


    
    private static Runnable generateTask(final SpinLock lock, 
                                        final String taskId, final List<String> list) {
        return () -> {
            lock.lock();
            try {
                Thread.sleep(300);
                list.add(taskId);

            } catch (Exception e) {

            }
            String s = list.toString();
            System.out.println(String.format("Thread %s Completed %s", taskId, s));
            lock.unLock();
            countDownLatch.countDown();
        };
    }
}

这里的关键就是加锁和释放锁中的两个CAS操作：

加锁过程:将CAS操作置于一个while循环中，来实现自旋的语义。由于CAS操作成功与否是成功取决于它的boolean返回值，因此当CAS操作失败的情况下，while循环将不会退出，会一直尝试CAS操作直到成功为止，此即所谓的自旋(忙等待)。
释放锁过程:此时不需要循环操作，但是仍然会考虑到只有当前拥有锁的线程才有资格释放锁。这一点还是通过CAS操作来保证。

这里用AtomicReference是为了使用它的原子性的compareAndSet方法（CAS操作），解决了多线程并发操作导致数据不一致的问题，确保其他线程可以看到锁的真实状态。

运行结果:

Thread 0 Completed [0]
Thread 9 Completed [0, 9]
Thread 3 Completed [0, 9, 3]
Thread 8 Completed [0, 9, 3, 8]
Thread 4 Completed [0, 9, 3, 8, 4]
Thread 6 Completed [0, 9, 3, 8, 4, 6]
Thread 5 Completed [0, 9, 3, 8, 4, 6, 5]
Thread 7 Completed [0, 9, 3, 8, 4, 6, 5, 7]
Thread 1 Completed [0, 9, 3, 8, 4, 6, 5, 7, 1]
Thread 2 Completed [0, 9, 3, 8, 4, 6, 5, 7, 1, 2]
[0, 9, 3, 8, 4, 6, 5, 7, 1, 2]

从上可以看出，加锁和释放锁正确的实现，和预期的也是一样的，这是一个非公平的自旋锁，不是按照先来先获取锁的方式实现。

缺点

CAS操作需要硬件的配合；
保证各个CPU的缓存（L1、L2、L3、跨CPU Socket、主存）的数据一致性，通讯开销很大，在多处理器系统上更严重；
没法保证公平性，不保证等待进程/线程按照FIFO顺序获得锁。

2. Ticket Lock

Ticket Lock 是为了解决上面的公平性问题，类似于现实中银行柜台的排队叫号：锁拥有一个服务号，表示正在服务的线程，还有一个排队号；每个线程尝试获取锁之前先拿一个排队号，然后不断轮询锁的当前服务号是否是自己的排队号，如果是，则表示自己拥有了锁，不是则继续轮询。

当线程释放锁时，将服务号加1，这样下一个线程看到这个变化，就退出自旋。

具体代码如下：


/**************************************
 *      Author : zhangke
 *      Date   : 2018-12-12 20:23
 *      email  : [email protected]
 *      Desc   : Ticket Lock 简单实现，
 *      此自旋锁保证了FIFI，不会产生饥饿现象
 ***************************************/
public class TicketLock {
    private AtomicInteger serviceNum = new AtomicInteger(); //服务号

    private AtomicInteger ticketNum = new AtomicInteger(); //排队号


    public int lock() {
        //首先原子性地获得一个排队号
        int myTicketNum = ticketNum.getAndIncrement();

        //只要当前服务号不是自己的就不断轮询
        while (serviceNum.get() != myTicketNum) {

        }
        return myTicketNum;
    }


    public void unLock(int myTicketNum) {
        //只有当前线程拥有者才能释放锁
        int next = myTicketNum + 1;
        serviceNum.compareAndSet(myTicketNum, next);
    }


    static CountDownLatch countDownLatch = new CountDownLatch(10);

    static List<String> list = new ArrayList<>();
    
    public static void main(String[] args) throws InterruptedException {
        final TicketLock lock = new TicketLock();

        for (int i = 0; i < 10; i++) {
            new Thread(generateTask(lock, String.valueOf(i), list)).start();
        }
        countDownLatch.await();
        System.out.println(list);
    }
    
    private static Runnable generateTask(final TicketLock lock, 
    		final String taskId, final List<String> list) {
        return () -> {
            int myTicketNum = lock.lock();
            try {
                Thread.sleep(300);
                list.add(taskId);

            } catch (Exception e) {

            }
            String s = list.toString();
            System.out.println(String.format("Thread %s Completed %s", taskId, s));
            lock.unLock(myTicketNum);
            countDownLatch.countDown();
        };
    }
}

加锁和释放锁两个操作的过程如下：

加锁过程。获取一个排队号，当排队号和当前的服务号不相等时自旋等待。
释放锁过程。当前正被服务的线程释放锁，计算下一个服务号并设置。

这里的AtomicInteger是为了保证服务号和等待号的原子性

运行结果如下

Thread 0 Completed [0]
Thread 1 Completed [0, 1]
Thread 2 Completed [0, 1, 2]
Thread 3 Completed [0, 1, 2, 3]
Thread 4 Completed [0, 1, 2, 3, 4]
Thread 5 Completed [0, 1, 2, 3, 4, 5]
Thread 6 Completed [0, 1, 2, 3, 4, 5, 6]
Thread 7 Completed [0, 1, 2, 3, 4, 5, 6, 7]
Thread 8 Completed [0, 1, 2, 3, 4, 5, 6, 7, 8]
Thread 9 Completed [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

从运行结果可以看出，确实是按照FIFO的顺序来获取锁，实现了公平性。

不过上面的代码有一个问题，在释放锁的时候需要外部传进来排队号，这样就会带来一个隐患，外部传进来的排队号是可以修改的，也就是线程A获取的排队号是5，但是释放锁的时候却传进来6。这样就有可能导致死锁等问题。下面这个是通过ThreadLocal来改进上面的代码

public class TicketLockImprove {
    private AtomicInteger serviceNum = new AtomicInteger(); //服务号

    private AtomicInteger ticketNum = new AtomicInteger(); //排队号

    //用于保存当前
    private ThreadLocal<Integer> threadLocalTicketNum = new ThreadLocal<>();


    public void lock() {
        //首先原子性地获得一个排队号
        int myTicketNum = ticketNum.getAndIncrement();
        //设置当前线程持有的排队号
        threadLocalTicketNum.set(myTicketNum);
        //只要当前服务号不是自己的就不断轮询
        while (serviceNum.get() != myTicketNum) {

        }

    }


    public void unLock() {
        //获取当前线程持有的排队号
        int myTicketNum = threadLocalTicketNum.get();
        //只有当前线程拥有者才能释放锁
        int next = myTicketNum + 1;
        serviceNum.compareAndSet(myTicketNum, next);
    }


    static CountDownLatch countDownLatch = new CountDownLatch(10);

    static List<String> list = new ArrayList<>();


    public static void main(String[] args) throws InterruptedException {
        final TicketLockImprove lock = new TicketLockImprove();

        for (int i = 0; i < 10; i++) {
            new Thread(generateTask(lock, String.valueOf(i), list)).start();
        }
        countDownLatch.await();
        System.out.println(list);
    }


    private static Runnable generateTask(final TicketLockImprove lock, final String taskId, final List<String> list) {
        return () -> {
            lock.lock();
            try {
                Thread.sleep(300);
                list.add(taskId);

            } catch (Exception e) {

            }
            String s = list.toString();
            System.out.println(String.format("Thread %s Completed %s", taskId, s));
            lock.unLock();
            countDownLatch.countDown();
        };
    }
}

这里是通过ThreadLocal来保存每个线程的排队号，这样就不会出现排队号被私自修改问题。其他的和上面的代码一致。

缺点

Ticket Lock 虽然解决了公平性的问题，但是多处理器系统上，每个进程/线程占用的处理器都在读写同一个变量serviceNum,每次读写操作都必须在多个处理器缓存之间进行缓存同步，这会导致繁重的系统总线和内存的流量，大大降低系统整体的性能。

所以，需要有一种办法能够让执行线程不再在同一个共享变量上自旋，避免过高频率的缓存同步操作。下面介绍的CLH锁和MCS锁都是为了解决这个问题的。

MCS来自于其发明人名字的首字母： John Mellor-Crummey和Michael Scott。

CLH的发明人是：Craig，Landin and Hagersten。

3. MCS 锁

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

先上实现代码，然后在分析重点：

/**************************************
 *      Author : zhangke
 *      Date   : 2018-12-12 20:34
 *      email  : [email protected]
 *      Desc   : MCS Lock
 ***************************************/
public class MCSLock {

    //MCS锁节点
    public static class MCSNode {
        //指向后继节点
        volatile MCSNode next;

        //默认是在等待锁
        volatile boolean isBlock = true;
    }

    volatile MCSNode tail; //指向最后一个申请锁的MCSNode

    // 原子更新器
    private static final AtomicReferenceFieldUpdater<MCSLock, MCSNode> UPDATER
            = AtomicReferenceFieldUpdater.newUpdater(MCSLock.class, MCSNode.class, "tail");

    //用于保存当前节点对应的MCSNode对象
    ThreadLocal<MCSNode> currentThreadNode = new ThreadLocal<>();


    public void lock() {

        //获得当前线程对应的节点
        MCSNode mcsNode = currentThreadNode.get();
        if (mcsNode == null) {
            //初始化节点对象
            mcsNode = new MCSNode();
            currentThreadNode.set(mcsNode);
        }
        //设置当前队列节点到队尾并获取前置节点
        MCSNode predecsessor = UPDATER.getAndSet(this, mcsNode); // step 1

        //如果前继节点不为空，则设置前置节点的后继为当前节点，并等待获取锁
        if (predecsessor != null) {
            predecsessor.next = mcsNode;  // step 2
            // 当前线程处于等待状态时自旋(MCSNode的isBlock初始化为true)
            // 等待前驱节点主动通知，即将isBlock设置为false，表示当前线程可以获取到锁
            while (mcsNode.isBlock) {

            }
        } else {
            // 只有一个线程在使用锁，没有前驱来通知它，所以得自己标记自己为非阻塞 - 表示已经加锁成功
            mcsNode.isBlock = false;
        }
    }


    public void unLock() {
        MCSNode mcsNode = currentThreadNode.get();
        // 当前线程对应存在节点并且
        // 锁拥有者进行释放锁才有意义 - 当blocked为true时，表示此线程处于等待状态中，
        //                          并没有获取到锁，因此没有权利释放锁
        if (mcsNode == null && mcsNode.isBlock) {
            return;
        }

        if (mcsNode.next == null) {
            if (UPDATER.compareAndSet(this, mcsNode, null)) {
                // compareAndSet返回true表示确实没有人排在自己后面
                return;
            } else {
                // 突然有人排在自己后面了，可能还不知道是谁，下面是等待后续者
                // 这里之所以要忙等是因为：step 1执行完后，step 2可能还没执行完
                while (mcsNode.next == null) {
                }
            }
        }
        // 通知后继节点可以获取锁
        mcsNode.next.isBlock = false;

        // 将当前节点从链表中断开，方便对当前节点进行GC
        mcsNode.next = null;// for GC
        //for GC
        currentThreadNode.remove();
    }


    public static void main(String[] args) throws InterruptedException {
        final MCSLock lock = new MCSLock();

        for (int i = 0; i < 10; i++) {
            new Thread(generateTask(lock, String.valueOf(i), list)).start();
        }
        countDownLatch.await();
        System.out.println(list);
    }


    static CountDownLatch countDownLatch = new CountDownLatch(10);

    static List<String> list = new ArrayList<>();


    private static Runnable generateTask(final MCSLock lock, 
    					final String taskId, final List<String> list) {
        return () -> {
            lock.lock();
            try {
                Thread.sleep(300);
                list.add(taskId);

            } catch (Exception e) {

            }
            String s = list.toString();
            System.out.println(String.format("Thread %s Completed %s", taskId, s));
            lock.unLock();
            countDownLatch.countDown();
        };
    }
}

实现的代码量虽然不多，但是lock和unlock的设计思想还是有些微妙之处，想要实现正确也并不容易。

需要把握的几个重点：

MCS锁的节点对象需要有两个状态，next用来维护单向链表的结构，blocked用来表示节点的状态，true表示处于自旋中；false表示加锁成功
MCS锁的节点状态blocked的改变是由其前驱节点触发改变的
加锁时会更新链表的末节点并完成链表结构的维护
释放锁的时候由于链表结构建立的时滞(getAndSet原子方法和链表建立整体而言并非原子性)，可能存在多线程的干扰，需要使用忙等待保证链表结构就绪

4. CLH锁

同MCS自旋锁一样，CLH也是一种基于单向链表(隐式创建)的高性能、公平的自旋锁，申请加锁的线程只需要在其前驱节点的变量上自旋，从而极大地减少了不必要的处理器缓存同步的次数，降低了总线和内存的开销。

实现代码

public class CLHLock {

    public static class CLHNode {
        private boolean isLocked = true; // 默认状态为true - 即处于等待状态或者加锁成功(换言之，即此节点处于有效的一种状态)
    }

    @SuppressWarnings("unused")
    private volatile CLHNode tail;

    //线程对应CLH节点映射
    private ThreadLocal<CLHNode> currentThreadNode = new ThreadLocal<>();


    /**
     * CLH 锁获取
     */
    public void lock() {
        CLHNode clhNode = currentThreadNode.get();
        if (clhNode == null) {
            clhNode = new CLHNode();
            currentThreadNode.set(clhNode);
        }
        //通过这个操作完成隐式链表的维护，后继节点只需要在前驱节点的locked状态上自旋
        CLHNode preNode = UPDATER.getAndSet(this, clhNode);
        if (preNode != null) {  //已有线程占用了锁，进入自旋
            while (preNode.isLocked) {  //自旋等待前驱节点状态变更 - unlock中进行变更

                //这里这样写，或者isLocked加上voliate，则能观察到变量的变化
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        // 没有前驱节点表示可以直接获取到锁，由于默认获取锁状态为true，此时可以什么操作都不执行
        // 能够执行到这里表示已经成功获取到了锁

    }


    /**
     * CLH 锁释放
     */
    public void unLock() {
        CLHNode clhNode = currentThreadNode.get();
        //只有持有锁的线程才能释放
        if (clhNode == null || !clhNode.isLocked) {
            return;
        }

        //从映射关系中移除当前线程对应的节点
        currentThreadNode.remove();

        //如果队列里只有当前线程，则释放对当前线程的引用(for GC)
        // 尝试将tail从currentThread变更为null，因此当tail不为currentThread时表示还有线程在等待加锁
        if (!UPDATER.compareAndSet(this, clhNode, null)) {
            //还有后续线程
            clhNode.isLocked = false; //改变状态，让后续线程结束自旋
        }
    }


    private static final AtomicReferenceFieldUpdater<CLHLock, CLHNode> UPDATER =
            AtomicReferenceFieldUpdater.newUpdater(CLHLock.class, CLHNode.class, "tail");

    static CountDownLatch countDownLatch = new CountDownLatch(10);


    public static void main(String[] args) throws InterruptedException {
        final CLHLock lock = new CLHLock();
        List<String> list = new ArrayList<>();

        for (int i = 0; i < 10; i++) {
            new Thread(generateTask(lock, String.valueOf(i), list)).start();
        }
        countDownLatch.await();
        System.out.println(list);
    }


    private static Runnable generateTask(final CLHLock lock, 
    					final String taskId, final List<String> list) {
        return () -> {
            lock.lock();
            try {
                Thread.sleep(300);
                list.add(taskId);
            } catch (Exception e) {

            }
            System.out.println(String.format("Thread %s Completed %s", taskId, list));
//            System.out.println(String.format("Thread %s Completed ", taskId));
            lock.unLock();
            countDownLatch.countDown();
        };
    }
}

实现的代码量相比MCS锁少了很多，也简洁了不少。

需要把握的几个重点：

CLH锁的节点对象只有一个isLocked属性，关于其含义前面已经详细讨论过
CLH锁的节点属性isLocked的改变是由其自身触发的
CLH锁是在前驱节点的isLocked属性上进行自旋

众所周知，AbstractQueuedSynchronizer是Java并发包的基石之一，而CLH锁的原理和思想则是AbstractQueuedSynchronizer的基石之一。理解清楚CLH锁的原理和实现对后面学习和理解AbstractQueuedSynchronizer是非常必要的。

在Doug Lea 写的The java.util.concurrent Synchronizer Framework 文章中有这么一句话 CLH锁显然比MCS锁更合适。因为CLH锁可以更容易地去实现“取消（cancellation）”和“超时”功能，因此我们选择了CLH锁作为实现的基础。其实只要实现了取消功能，那么超时就比较容易实现，因为超时功能就是在取消的功能基础上加了一个时间的设置。为什么CLH更容易实现取消功能呢。首先在CLH各节点之间只有一条隐形的链表存在，而排队的节点是观察前面节点的信息来判断是否可以获取锁，因此在取消排队节点时，是不需要修改前继节点，而后继节点只需要简单的进行一次自旋看是否可以获取锁，此时因为是取消节点，而不是释放锁，所以后继节点是获取不了，接着后继节点就继续自旋，因此对取消节点，既不需要更改后继节点，也不需要修改前继节点，所以相对来说需要操作的少，因此就相对与MCS自旋锁要简单点。如果MCS自旋锁要取消，则需要改变此节点的自旋状态，改变后继节点在链表中的位置，也就是取代此节点，变成此节点前继节点的后继节点，如果此时前继节点正在拿到锁运行，那此时更改锁就不是那么容易的，而且链表又是单链表，因此更改起来更加麻烦。

总结

下面我们来比较一下MCS和CLH锁

首先我们先补充一点知识SMP和NUMA架构

SMP(Symmetric Multi-Processor)：对称多处理器结构，指服务器中多个 CPU 对称工作，每个 CPU 访问内存地址所需时间相同。其主要特征是共享，包含对 CPU，内存，I/O 等进行共享。SMP 能够保证内存一致性，但这些共享的资源很可能成为性能瓶颈，随着 CPU 数量的增加，每个 CPU 都要访问相同的内存资源，可能导致内存访问冲突，可能会导致 CPU 资源的浪费。常用的 PC 机就属于这种。
NUMA(Non-Uniform Memory Access)：非一致存储访问，将 CPU 分为 CPU 模块，每个 CPU 模块由多个 CPU 组成，并且具有独立的本地内存、I/O 槽口等，模块之间可以通过互联模块相互访问，访问本地内存的速度将远远高于访问远地内存 (系统内其它节点的内存) 的速度，这也是非一致存储访问的由来。NUMA 较好地解决 SMP 的扩展问题，当 CPU 数量增加时，因为访问远地内存的延时远远超过本地内存，系统性能无法线性增加。

CLH 队列锁的优点是空间复杂度低（如果有 n 个线程，L 个锁，每个线程每次只获取一个锁，那么需要的存储空间是 O（L+n），n 个线程有 n 个myNode，L 个锁有 L 个 tail），CLH 的一种变体被应用在了 JAVA 并发框架中 (AbstractQueuedSynchronizer.Node)。CLH 在 SMP 系统结构下该法是非常有效的。但在 NUMA 系统结构下，每个线程有自己的内存，如果前趋结点的内存位置比较远，自旋判断前趋结点的 locked 域，性能将大打折扣，一种解决 NUMA 系统结构的思路是 MCS 队列锁。

MSC 与 CLH 最大的不同并不是链表是显示还是隐式，而是线程自旋的规则不同:CLH 是在前趋结点的 locked 域上自旋等待，而 MCS 是在自己的结点的 locked 域上自旋等待。正因为如此，它解决了 CLH 在 NUMA 系统架构中获取 locked 域状态内存过远的问题。