自旋锁、排队自旋锁、MCS锁、CLH锁

自旋锁(Spin lock)

自旋锁是指当一个线程尝试获取某个锁时,如果该锁已被其他线程占用,就一直循环检测锁是否被释放,而不是进入线程挂起或睡眠状态。

自旋锁适用于锁保护的临界区很小的情况,临界区很小的话,锁占用的时间就很短。

简单的实现

import java.util.concurrent.atomic.AtomicReference;

public class SpinLock {
   private AtomicReference<Thread> owner = new AtomicReference<Thread>();

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

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

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

              // 只有锁的拥有者才能释放锁
       owner.compareAndSet(currentThread, null);
   }
}

SimpleSpinLock里有一个owner属性持有锁当前拥有者的线程的引用,如果该引用为null,则表示锁未被占用,不为null则被占用。

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

缺点

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

Ticket Lock

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

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

简单的实现

import java.util.concurrent.atomic.AtomicInteger;

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 myTicket) {
        // 只有当前线程拥有者才能释放锁
        int next = myTicket + 1;
        serviceNum.compareAndSet(myTicket, next);
    }
}

缺点

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

下面介绍的CLH锁和MCS锁都是为了解决这个问题的。

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

CLH的发明人是:Craig,Landin and Hagersten。

MCS锁

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

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

public class MCSLock {
    public static class MCSNode {
        volatile MCSNode next;
        volatile boolean isBlock = true; // 默认是在等待锁
    }

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

    public void lock(MCSNode currentThread) {
        MCSNode predecessor = UPDATER.getAndSet(this, currentThread);// step 1
        if (predecessor != null) {
            predecessor.next = currentThread;// step 2

            while (currentThread.isBlock) {// step 3
            }
        }else { // 只有一个线程在使用锁,没有前驱来通知它,所以得自己标记自己为非阻塞
               currentThread. isBlock = false;
          }
    }

    public void unlock(MCSNode currentThread) {
        if (currentThread.isBlock) {// 锁拥有者进行释放锁才有意义
            return;
        }

        if (currentThread.next == null) {// 检查是否有人排在自己后面
            if (UPDATER.compareAndSet(this, currentThread, null)) {// step 4
                // compareAndSet返回true表示确实没有人排在自己后面
                return;
            } else {
                // 突然有人排在自己后面了,可能还不知道是谁,下面是等待后续者
                // 这里之所以要忙等是因为:step 1执行完后,step 2可能还没执行完
                while (currentThread.next == null) { // step 5
                }
            }
        }

        currentThread.next.isBlock = false;
        currentThread.next = null;// for GC
    }
}

CLH锁

CLH锁也是一种基于链表的可扩展、高性能、公平的自旋锁,申请线程只在本地变量上自旋,它不断轮询前驱的状态,如果发现前驱释放了锁就结束自旋。

import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

public class CLHLock {
    public static class CLHNode {
        private volatile boolean isLocked = true; // 默认是在等待锁
    }

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

    public void lock(CLHNode currentThread) {
        CLHNode preNode = UPDATER.getAndSet( this, currentThread);
        if(preNode != null) {//已有线程占用了锁,进入自旋
            while(preNode.isLocked ) {
            }
        }
    }

    public void unlock(CLHNode currentThread) {
        // 如果队列里只有当前线程,则释放对当前线程的引用(for GC)。
        if (!UPDATER .compareAndSet(this, currentThread, null)) {
            // 还有后续线程
            currentThread. isLocked = false ;// 改变状态,让后续线程结束自旋
        }
    }
}

CLH锁 与 MCS锁 的比较

下图是CLH锁和MCS锁队列图示:
CLH-MCS-SpinLock

差异:

  1. 从代码实现来看,CLH比MCS要简单得多。
  2. 从自旋的条件来看,CLH是在前驱节点的属性上自旋,而MCS是在本地属性变量上自旋。
  3. 从链表队列来看,CLH的队列是隐式的,CLHNode并不实际持有下一个节点;MCS的队列是物理存在的。
  4. CLH锁释放时只需要改变自己的属性,MCS锁释放则需要改变后继节点的属性。

注意:这里实现的锁都是独占的,且不能重入的。


欢迎关注我的微信公众号: coderbee笔记,可以更及时回复你的讨论。

自旋锁、排队自旋锁、MCS锁、CLH锁》上有46个想法

  1. 我觉得MCS锁unlock这里有问题、
    if (currentThread.isBlock) {// 锁拥有者进行释放锁才有意义
    return;
    }

    thread 1 lock成功、调用unlock放锁、isBlock等于true直接返回了、
    thread 2 lock会卡死在while循环中、因为thread 2的isBlock一直是true、thread 1已经释放了、不会再设置thread 2的isBlock了

    我觉得讲这个if 去掉或者if(!currentThread.isBlock) 都可以

    • 你理解反了 currentThread.isBlock == true 是说当前线程没有获得锁 处于阻塞状态 所以它无权限去释放锁,只有获得了锁 才有权限去释放锁

  2. 还有简单实现的while循环我觉得也有问题

    while (owner.compareAndSet(null, currentThread)) {
    }
    假如有一个thread 加锁成功后、
    后面的thread compareAndSet失败 直接lock返回

    所以我觉得应该是while (!owner.compareAndSet(null, currentThread))

    第一个上锁的compareAndSet返回true、lock函数退出、后面compareAndSet返回false,while旋转

  3. 收到测试了下,MSCLock的代码,,发现有的时候,并发获取和解锁很快,有的时候很慢,,不如内置锁稳定。。

    • 你可以粘贴一下测试的代码吗?我这边试了一下,和预期不一样

  4. 从链表队列来看,CLH的队列是隐式的,CLHNode并不实际持有下一个节点;MCS的队列是物理存在的。
    请教下这个结论是怎么在代码上看出来的?

    • CLH的队列依靠的是unsafe的getAndSet来伪造的,而MSC的Node确实有next属性指向(MCS释放就是释放Node的next)

  5. 从链表队列来看,CLH的队列是隐式的,CLHNode并不实际持有下一个节点;MCS的队列是物理存在的。
    请教下这个结论是怎么在代码上提现的?

    • MCSNode 有 next 属性指向下一个等待节点,只要找到头节点就可以遍历完整的链表。

      CLH 是线程持有前一个等待线程的节点,CLHNode 里并没有前驱的指针,所以说是隐式的。

  6. 锁:
    public class MCSLock_1 {

    static class Node {
    volatile Node next;
    volatile boolean isBlock = true;
    }
    volatile Node queue;
    private static final AtomicReferenceFieldUpdater updater = AtomicReferenceFieldUpdater.newUpdater(MCSLock_1.class, Node.class, “queue”);
    public void lock(Node node){
    //线程蜂拥而至,只有一个线程predecessor为null,其余进行自旋
    Node predecessor = updater.getAndSet(this, node);
    if(predecessor!=null){
    predecessor.next = node;
    while(node.isBlock);
    } else {
    node.isBlock = false;
    }
    System.out.println(Thread.currentThread().getName()+”获取到了锁!”);
    }
    public void unlock(Node node){
    //获得锁的线程执行完毕后,进行释放锁操作,即将isBlock
    if(node.isBlock) return;
    if(node.next==null){
    if(updater.compareAndSet(this, node, null)){
    return;
    } else {
    while(node.next==null);
    }
    }
    node.next.isBlock = false;
    }
    }
    测试:
    public class MCSLock_2 {

    private MCSLock_1 lock = new MCSLock_1();
    Node node = new MCSLock_1.Node();
    public void print(){
    lock.lock(node);
    System.out.println(“\t”+Thread.currentThread().getName()+”打印中…”);
    try {
    Thread.currentThread().sleep(10);
    } catch (InterruptedException e) {
    e.printStackTrace();
    }
    System.out.println(“\t”+Thread.currentThread().getName()+”打印完毕!”);
    lock.unlock(node);
    }

    public static void main(String[] args) {
    MCSLock_2 obj = new MCSLock_2();
    for(int i=0; i<5; i++){
    new Thread(new Runnable() {
    @Override
    public void run() {
    obj.print();
    }
    }, "线程"+i).start();;
    }
    }
    }
    结果:
    线程2获取到了锁!
    线程3获取到了锁!
    线程4获取到了锁!
    线程2打印中…
    线程1获取到了锁!
    线程1打印中…
    线程0获取到了锁!
    线程0打印中…
    线程4打印中…
    线程3打印中…
    线程3打印完毕!
    线程1打印完毕!
    线程4打印完毕!
    线程0打印完毕!
    线程2打印完毕!
    求教:
    自己不太理解,就做了一个例子,但是不能实现MCS锁,求大佬指点

    • 你的例子的问题在于:申请锁时总是用了同一个节点,MCS 锁每次申请时需要使用不同的节点来排队。

      采用类似资源贷出的模式来使用MCS锁可能比较合适,下面是我写的例子:

      public class MCSLock_2 {
      private MCSLock_1 lock = new MCSLock_1();

      public void runInLock(Runnable action) {
      MCSLock_1.Node node = new MCSLock_1.Node();
      lock.lock(node);

      try {
      action.run();
      } finally {
      lock.unlock(node);
      }
      }

      public static void main(String[] args) {
      MCSLock_2 obj = new MCSLock_2();
      for (int i = 0; i < 5; i++) { new Thread(() -> {
      obj.runInLock(() -> {
      System.out.println("\t" + Thread.currentThread().getName() + "打印中…");
      try {
      Thread.currentThread().sleep(10);
      } catch (InterruptedException e) {
      e.printStackTrace();
      }
      System.out.println("\t" + Thread.currentThread().getName() + "打印完毕!");
      });
      }, "线程" + i).start();
      }
      }
      }

      • 谢谢你!后面我按照你的思路,把我的改了一下,即:每次lock都需要将后继Node设置为当前的Node,代码如下:
        public class MCSLock_2 {

        private MCSLock_1 lock = new MCSLock_1();
        public void print(){
        Node node = new MCSLock_1.Node();
        lock.lock(node);
        System.out.println(“\t”+Thread.currentThread().getName()+”打印中…”);
        try {
        Thread.currentThread().sleep(10);
        } catch (InterruptedException e) {
        e.printStackTrace();
        }
        System.out.println(“\t”+Thread.currentThread().getName()+”打印完毕!”);
        lock.unlock(node);
        }

        public static void main(String[] args) {
        MCSLock_2 obj = new MCSLock_2();
        for(int i=0; i<5; i++){
        new Thread(new Runnable() {
        @Override
        public void run() {
        obj.print();
        }
        }, "线程"+i).start();;
        }
        }
        }
        这下就好了!但是多个线程在new Node()时,应该会出现(内存可见性)问题吧?

        • 我又改进了一下,为了降低业务代码的耦合,把Node节点即当前线程对应的节点的创建封装到锁内部
          MCS锁:
          public class MCSLock_1 {

          static class Node {
          volatile Node next;
          volatile boolean isBlock = true;
          }
          volatile Node prev;
          private Node current;
          private AtomicReferenceFieldUpdater updater = AtomicReferenceFieldUpdater.newUpdater(MCSLock_1.class, Node.class, “prev”);

          public void lock(){
          //1.为每一个线程创建一个Node
          Node node = new Node();
          //2.获取线程前任节点
          Node predecessor = updater.getAndSet(this, node);
          //3.如果前任节点为null,说明锁无人使用,isBlock置为false
          if(predecessor!=null){
          predecessor.next = node;
          while(node.isBlock);
          } else {
          node.isBlock = false;
          }
          //设置current(当前线程)
          if(!node.isBlock) current = node;
          System.out.println(Thread.currentThread().getName()+”获取到了锁!”);
          }
          public void unlock(){
          if(current.next!=null){
          current.next.isBlock = false;
          } else {
          //忙等:等待current.next赋值;缺陷:如果赋值失败,会一直循环,占用大量CPU资源
          while(current.next==null);
          }
          }
          }
          测试类:
          public class MCSLock_2 {

          private MCSLock_1 lock = new MCSLock_1();
          public void print(){
          lock.lock();
          System.out.println(“\t”+Thread.currentThread().getName()+”打印中…”);
          try {
          Thread.currentThread().sleep(1000);
          } catch (InterruptedException e) {
          e.printStackTrace();
          }
          System.out.println(“\t”+Thread.currentThread().getName()+”打印完毕!”);
          lock.unlock();
          }

          public static void main(String[] args) {
          MCSLock_2 obj = new MCSLock_2();
          for(int i=0; i<5; i++){
          new Thread(new Runnable() {
          @Override
          public void run() {
          obj.print();
          }
          }, "线程"+i).start();
          }
          }
          }
          但是现在这样,有个隐患就是:在执行unlock时,如果当前Node的next赋值失败,则会一直循环,造成CPU大量占用

          • 你看下 MCSLock.unlock 方法,step 1 成功了,step 2 是不会失败的。这里不存在循环什么,是很快就执行完的,除非 step 1 执行后、step 2 执行完成前线程被操作系统调度出去了。

        • 每个线程在方法里创建的、在未加入队列之前对其他线程都是不可见,,,线程间只能通过锁才能感知到其他线程的等待节点。

  7. 不太理解什么叫本地变量自旋, 是针对前2种来说的吗 , 前2种是 堆变量, 后面是线程栈变量,是这意思吗?

    • 是的,前两种的变量是共享、所有线程都需要访问。CLH、MCS 的变量几乎是独占。

  8. 没办法保证打印出的是1000,请问是啥原因

    public class MCSLockTest {
    private MCSLock mcsLock = new MCSLock();

    @Test
    public void test() {
    MyLong myLong = new MyLong();
    ExecutorService executorService = Executors.newFixedThreadPool(100);
    for (int i = 0; i < 1000; ++i) {
    executorService.execute(new Task(myLong));
    }
    System.out.println(myLong.l);
    }

    private class Task implements Runnable {
    private final MyLong myLong;
    private final MCSNode mcsNode;

    public Task(MyLong myLong) {
    this.myLong = myLong;
    this.mcsNode = new MCSNode();
    }

    @Override
    public void run() {
    mcsLock.lock(mcsNode);
    myLong.increment();
    mcsLock.unlock(mcsNode);
    }
    }

    private class MyLong {
    private long l = 0L;
    public void increment() {
    l++;
    }
    }
    }

    • executorService.execute 只是把任务提交到线程池,并不表示任务执行完成了。

  9. 博主您好!我想请教个问题,您文章里说MCS锁和CLH锁申请锁时都是在本地变量上自旋,但是自旋的变量isLocked是用volatile修饰的,读取volatile变量时不是会在读操作前插入LoadLoad屏障吗?这样会清空无效化队列,也会带来一定的开销。这个开销与Ticket Lock里原子变量的get操作的开销相比如何呢?

    • 同问,CLH中需要轮询前驱状态,但是前驱状态的获取难道不需要缓存同步吗?需要同步的话,这个效率有什么区别?

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据