锁 | GRYFFONDOR

在很多的多线程编程场景下都会遇到多个线程对一个资源进行操作访问的情况，这种场景一旦发生就会牵扯到线程安全问题。为了保证程序的正确性，我们不得不花很大的力气去解决这些线程安全问题。在Java中解决线程安全问题的办法被分为了三种，其一是互斥同步，其二是非阻塞同步，其三是无同步方案。前两种的实现形式都是锁，第三种是通过设计模式的转变来将代码转变为不共享变量的形式，这不在这篇博客的讨论范围中。

线程不安全

首先我们来分析一下线程安全问题产生的底层原因到底是什么？先看下面的代码：

package io.talkwithkeyboard.code;

public class MultiThreadIncrease {

    public static int race = 0;

    public static void increase() {
        race = race + 1;
    }

    public static void main(String[] args) {
        final int THREAD_COUNT = 20;
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < THREAD_COUNT; i++) {
            threads[i] = new Thread(() -> {
                System.out.format("This is %d thread.\n", Thread.currentThread().getId());
                for (int j = 0; j < 10000; j++) {
                    increase();
                }
            });
            threads[i].start();
        }
        while (Thread.activeCount() > 2) {
            Thread.yield();
        }
        System.out.println(race);
    }
}

首先解释一个细节，因为我的代码是在Idea里跑的，所以不仅是有Main thread，还有一个Monitor Ctrl-Break的守护线程。所以代码中是Thread.activeCount() > 2。如果直接使用java执行的话，这里是1即可。然后上面的代码做了一个很简单的事情，开了20个线程，每个线程做一件事情，对race这个变量累加10000次，最后输出。最后的结果显然不是200000，会小很多并且每次都不一样。这是为什么呢？

是因为race = race + 1这一行代码其实做了三件事情：

1. 取出race现有的值
1. 给race现有的值加上1
1. 将更新后的值再附给race

我们理想的状态是，每个线程顺序的做完这三件事：

thread1.1  // race=0
thread1.2  // race=0
thread1.3  // race=1
thread2.1  // race=1
thread2.2  // race=1
thread2.3  // race=2
...

但实际是：

thread1.1  // race=0
thread2.1  // race=0
thread2.2  // race=0
thread1.2  // race=0
thread1.3  // race=1
thread2.3  // race=1
...

甚至更加的混乱，这就造成代码运行结果错误的现象，也就是出现了线程不安全行为。那么为了规避这样的行为，就需要引出锁的概念，悲观锁就是互斥同步的实现，乐观锁是非阻塞同步的实现。

互斥同步悲观锁

首先我们看看《深入理解JVM》中对互斥同步的定义，”互斥同步是常见的一种并发正确性保障手段。同步是指在多个线程并发访问共享数据时，保证共享数据在同一个时刻只被一个（或者一些，使用信号量的时候）线程使用。而互斥是实现同步的一种手段，临界区、互斥量和信号量都是主要的互斥实现方式。因此，在这4个字里面，互斥是因，同步是果；互斥是方法，同步是目的。“

synchronized

而如何保证共享数据在同一个时刻只被一个线程使用？那么就需要在这个数据被使用之前就为期加上锁，只有获得锁的线程能够对其进行操作，而这样的锁就被称为悲观锁。在Java中，最基本的实现就是synchronized关键字。其实现的原理是在编译后会在同步块的前后分别形成monitorenter和monitorexit这两个字节码指令，这两个字节码指令都需要一个reference类型的参数来指明要锁定和解锁的对象。如果指明的是对象参数，那就是这个对象的reference；如果没有明确指定，那就根据synchronized的是实例还是类方法，去取对应的对象实例或Class对象来作为锁对象。先看下面的例子，对比一下添加synchronized关键字前后的字节码：

package io.talkwithkeyboard.code;

public class NoSynchronized {

    public static int race = 0;

    public void increase() {
        race = race + 1;
    }

    public static void main(String args[]) {
        new NoSynchronized().increase();
    }
}

通过javap工具来获取字节码：

1	$ javap -verbose -p io.talkwithkeyboard.code.NoSynchronized

我们只关注increase方法：

public void increase();
    descriptor: ()V
    flags: ACC_PUBLIC
    Code:
      stack=2, locals=1, args_size=1
         0: getstatic     #2                  // Field race:I
         3: iconst_1
         4: iadd
         5: putstatic     #2                  // Field race:I
         8: return

那么在添加synchronized关键字后：

package io.talkwithkeyboard.code;

public class WithSynchronized {
    public static int race = 0;

    public void increase() {
        synchronized (this) {
            race = race + 1;
        }
    }

    public static void main(String args[]) {
        new WithSynchronized().increase();
    }
}

还是只关注increase方法：

public void increase();
    descriptor: ()V
    flags: ACC_PUBLIC
    Code:
      stack=2, locals=3, args_size=1
         0: aload_0
         1: dup
         2: astore_1
         3: monitorenter
         4: getstatic     #2                  // Field race:I
         7: iconst_1
         8: iadd
         9: putstatic     #2                  // Field race:I
        12: aload_1
        13: monitorexit
        14: goto          22
        17: astore_2
        18: aload_1
        19: monitorexit
        20: aload_2
        21: athrow
        22: return

字节码描述的过程是：

1. 将类对象入栈
1. 复制栈顶元素（即类对象的引用）
1. 将栈顶元素（类对象）存储到局部变量表Slot 1中
1. 以栈顶元素做为锁开始同步
1. 取获取类的静态字段（race），将其值压入栈顶
1. int型常量1进栈
1. 对操作数栈上的两个数值进行加法，结果压入栈顶
1. 用栈顶元素给类的静态字段（race）赋值
1. 将局部变量表Slot 1中的类对象入栈
1. 退出同步
1. 方法正常结束，跳转到22返回
1. 从这步开始是异常路径，暂不赘述

在展示了整个synchronized关键字的代码流程以后，我们再深究一下monitorenter指令和monitorexit指令在机器码成面到底做了什么。为了阅读方便，我们先不展示机器码的内容，而是从虚拟机规范出发，在执行monitorenter指令的时，首先要尝试获取对象的锁。如果这个对象没有被锁定，或者当前线程已经拥有了那个对象的锁，把锁的计数器加1。相应的，在执行monitorexit指令的时候，把锁的计数器减1，当计数器为0的时候，锁就被释放掉。

ReentrantLock

以上就是锁的整个低层实现过程，在Java中其实还有更上层的锁封装能实现更多特性的锁，那就是ReentrantLock类，它和synchronized关键字一样都是悲观锁的实现。但是相比synchronized，增加了一些高级功能，主要是以下三点：

等待可中断：当持有锁的线程长期不释放锁的时候，正在等待的线程可以选择放弃等待，改为处理其他事情，可中断特性对处理执行时间非常长的同步块很有帮助。

// lock() 实现
final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
// lockInterruptibly() 实现
private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
	}

可以看到在lock()和lockInterruptibly()源码的实现中，唯一的区别是在一直等待锁的过程中，lock()会吞掉中断，近记录中断状态，而lockInterruptibly()会抛异常到上层，交给上面的业务逻辑进行处理。

公平锁：多个线程在等待同一个锁时，必须按照申请锁的时间顺序来依次获得锁，而非公平锁是不能保证这一点的。synchronized就是非公平锁，可以通过
1
final ReentrantLock lock = new ReentrantLock(true);

来创建公平锁。

可以绑定多个条件：主要是处理生产者消费者模型，由于篇幅这里暂不赘述。

volatile

volatile可以说是Java中最轻量化级的同步机制，在一定程度上也是可以当作对象的锁来进行使用的，但是在功能上还是不能完全替代被synchronized作用的对象。当一个变量定义为volatile之后，它将具备两种特性，第一是当一个线程修改了这个变量的值，新值对于其他线程来说是立即得知的，这就是可见性。第二个语义是禁止指令重排序优化。

首先介绍一下可见性是如何实现的？普通变量在被一个线程修改之后，会向主内存进行回写，只有等到主内存回写完成以后，其他的线程才能读到新的值。而被volatile修饰的变量在赋值后会产生一个lock addl $0x0,(%esp)向寄存器中加0的空操作，这个操作能使用本CPU的Cache写入内核，并使别的CPU或者别的内核无效化Cache。相当于将工作内存中的变量拿到了主内存当中，正是因为此让volatile修饰的变量马上对其他线程可见。

package io.talkwithkeyboard.code;

public class VolatileControl {

    private volatile static boolean shutdownRequested = false;

    public void shutdown() {
        shutdownRequested = true;
    }

    public void doWork() {
        long loopCount = 100000000;
        System.out.println(Thread.currentThread().getId() + ":" + loopCount);
        for (int i = 0; i < loopCount; i++) {
            if (shutdownRequested) {
                return;
            }
        }
        System.out.println(Thread.currentThread().getId() + ":" + "shutdown!");
        shutdown();
    }

    public static void main(String[] args) {
        final int THREAD_COUNT = 10;
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < THREAD_COUNT; i ++) {
            threads[i] = new Thread(() -> new VolatileControl().doWork());
            threads[i].start();
        }

        while (Thread.activeCount() > 2) {
            Thread.yield();
        }
    }
}

比如在这个例子当中，让每个线程都循环100000000次，在大多数情况下最后可以看到shutdown!只被打印了一次。但是一旦去掉volatile修饰以后，就会看到很多个shutdown!被打印出来，这就是因为很多线程在shutdownRequested被修改以后，都读到了老版本的值，出现了线程不安全的情况。而volatile从表现来看基本上达到了为shutdownRequested加锁的效果。但是刚才也提到了我们是在大多数情况下是只看到一次shutdown!，这是为什么呢？可以先看一个更加容易复现的例子：

package io.talkwithkeyboard.code;

public class VolatileIncrease {
    public static volatile int race = 0;

    public static void increase() {
        race = race + 1;
    }

    public static void main(String[] args) {
        final int THREAD_COUNT = 20;
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < THREAD_COUNT; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    increase();
                }
            });
            threads[i].start();
        }

        while (Thread.activeCount() > 2) {
            Thread.yield();
        }

        System.out.println(race);
    }
}

还是上面出现过的，每个线程都给race累加值的代码，只不过现在会用volatile进行修饰。volatile的特性又是值被修改后立即能被其他线程看见，那么这个例子就应该输出正确的结果200000，但是运行后会发现还是出现了上面提到的线程不安全的问题。那么这是不是和volatile的描述不符呢？我们还是输出字节码来看看：

public static void increase();
    descriptor: ()V
    flags: ACC_PUBLIC, ACC_STATIC
    Code:
      stack=2, locals=0, args_size=0
         0: getstatic     #2                  // Field race:I
         3: iconst_1
         4: iadd
         5: putstatic     #2                  // Field race:I
         8: return

可以看到问题是出在race = race + 1上面，到字节码层面上这个操作已经被拆分成了4条指令，并且不具备事务性了。volatile能保证的是getstatic能取到最新的值，但是在iadd操作的时候其他线程可能已经把这个值加大了。在上面的例子当中同理，在将shutdownRequested赋值为true的时候，可能其他线程已经赋值成功，但是当前线程不可见。所以volatile的作用是非常轻微的，只能够保证在取值的时候能取到最新值，当一个操作的事务性无法保证的时候，volatile也不能提供锁的性质。至于防止指令重拍和题目相关性不强，这里先不做赘述。

总结

在《深入理解JVM虚拟机》中，有对synchronized和ReentrantLock进行性能对比，通过对synchronized的优化，性能基本上持平。并且提供因为团队会更偏向于优化原生的synchronized关键字，所以当两个都能使用的时候可以优先使用synchronized关键字，需要更高阶的功能时，再选择ReentrantLock。但是因为阻塞的实现方式，这两种实现都会阻塞后面其他的线程进入，而Java的线程是映射到操作系统的原生线程之上的，如果一个线程要阻塞或唤醒，都是需要操作系统从用户态切换到核心态来进行帮忙的，所以需要非常谨慎的时候，在一定情况下是可以使用volatile关键字来进行替代的，以提高性能。

非阻塞同步乐观锁

由于悲观锁的实现中涉及到加锁、用户态核心态切换、维护锁计数器和检查是否有被阻塞线程需要被唤醒等复杂的操作，在执行效率上大打折扣。随着硬件指令集的发展，又多了一种锁实现方案，也就是乐观锁，其主要的思想是：先进行操作，如果没有竞争则操作成功，如果有竞争，那就再采取其他的补偿措施。这种实现方式下，不需要将线程挂起，因此也称为非阻塞同步。

Compare-and-Swap

乐观锁的一个实现关键是需要让“现在的值等于旧预期值时，将新预期值写入”这个操作原子化，而这也依赖于硬件指令集的发展，出现了CAS(Compare-and-Swap)指令来完成这个任务。CAS指令需要三个操作数，分别是内存位置V、旧的预期值A、新的预期值B。CAS指令执行时，当且仅当V符合旧的预期值A的时候，处理器用新值B更新V的值，否则它就不执行更新，但是无论是否更新了V的值，都会返回V的旧值，并且上述的过程是原子性的。在JDK1.5之后，sun.misc.Unsafe类的compareAndSwapInt()和compareAndSwapLong()等几个操作都依靠CAS指令执行，虚拟机内部对这些方法做了特殊处理，即时编译出来的结果就是一条平台相关的处理器CAS指令。在更上层中的接口中，AtomicInteger.incerementAndGet()等使用了Unsafe的低层接口。

package io.talkwithkeyboard.code;

import java.util.concurrent.atomic.AtomicInteger;

public class MultiThreadAtomicIncrease {

    public static AtomicInteger race = new AtomicInteger(0);

    public static void insert() {
        race.incrementAndGet();
    }

    public static void main(String[] args) {
        final int THREAD_COUNT = 20;
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < THREAD_COUNT; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    insert();
                }
            });
            threads[i].start();
        }

        while (Thread.activeCount() > 2) {
            Thread.yield();
        }
        System.out.println(race.get());
    }
}

使用乐观锁来对上面多线程累加的程序进行优化，运行程序可以看到正确的结果。

1
2
3

public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

incrementAndGet方法就是对Unsafe类的getAndAddInt方法进行了封装，而getAndAddInt在低层使用了和CAS类似的指令Fetch-and-Increment，将获取值和累加两个操作进行原子化封装。而在早期的JDK实现中，是使用CAS指令进行完成，不断尝试将比现在值大1的值写入。这个优化也是硬件指令集的进一步丰富带来的。

public final int getAndIncrement() {
    for (;;) {
        int current = get();
        int next = current + 1;
        if (compareAndSet(current, next))
            return current;
    }
}

锁优化

自旋锁

因为互斥同步对性能的消耗非常大，并且JVM团队发现大量的锁定状态只会持续很短的一段时间，这个时间远小于对CPU的用户态和内核态切换时间。所以就想出来一个办法不轻易的对线程进行阻塞，而使用忙循环（自旋）来替代。不过自旋也不是完全优于阻塞的，虽然省下了线程切换的开销，不过忙循环会占用处理器时间，所以如果锁定状态时间较短使用自旋是划算的，锁定状态时间较长就会浪费处理器资源，带来性能的消耗。因此现在的实现中，会规定一个自旋的上限，当达到上限以后就转为重锁的方式挂起线程。现在高版本的JDK中自旋是默认开启的，Java用户可以通过-XX:PreBlockSpin来修改自旋的次数。

并且为了进一步的提高自旋锁的性能，在JDK1.6提出了自适应的自旋锁，每次自旋的时间不固定，而是由前一次在同一个锁上的自旋时间以及锁的拥有者的状态决定。如果在一个锁对象上，通过自旋的方式经常成功获得过锁，并且持有锁的进程正在运行中，那么这次自旋有较大可能获得锁，就可以等待较多的自旋次数。如果在一个锁对象上从来没有成功通过自旋获得锁，那么就直接省去自旋步骤，直接进入重锁。

锁消除

锁消除是指开发人员虽然要求一段代码块需要上锁，同步执行。但是被JVM检测到存在不可能存在共享数据竞争的锁，就会自动将其消除掉。这个检测主要依赖逃逸分析的数据支持，如果判断在一段代码中，堆上的所有数据都不会逃逸出去被其他线程访问到，那就可以把它当作栈上数据对待，认为他们是线程私有的，就不用加锁。

锁粗化

很多时候我们都希望加锁的作用范围限制的尽可能的小，这样可以缩短锁状态的持续时间，让等待的线程尽快的获得锁。但是偶尔会出现一系列连续的操作都对同一个对象反复加锁和解锁，甚至加锁操作出现在循环体中的，这样频繁的进行互斥同步会极大的降低执行效率，这时候虚拟机探测到有这样一串零碎的操作都对一个对象加锁，就会把加锁的范围粗化到整个操作序列的外部，这样加锁一次就可以了。就还是用上面的例子举例，每次increase操作都有加锁解锁的步骤，这时就会把锁粗化到for循环的外部。

package io.talkwithkeyboard.code;

public class MultiThreadIncreaseSync {

    public static int race = 0;

    public static synchronized void increase() {
        race = race + 1;
    }

    public static void main(String[] args) {
        final int THREAD_COUNT = 20;
        Thread[] threads = new Thread[THREAD_COUNT];
        for (int i = 0; i < THREAD_COUNT; i++) {
            threads[i] = new Thread(() -> {
                for (int j = 0; j < 10000; j++) {
                    increase();
                }
            });
            threads[i].start();
        }

        while (Thread.activeCount() > 2) {
            Thread.yield();
        }

        System.out.println(race);
    }
}

轻量级锁

轻量级锁的优化方向是使用CAS代替互斥量的开销，并且依据经验“对于绝大多数的锁，在整个同步周期内都是不存在竞争的”，假设没有竞争那么CAS操作就避免了互斥量的开销，但是如果存在竞争，轻量锁最终会膨胀为重量锁，不仅有互斥量的开销，还多了CAS操作。在HotSpot虚拟机中，对象头由两部分组成，一部分是非固定数据结构的，用来储存对象自身的运行时数据，如哈希值，GC分带年龄等数据，官方称为”Mark word”；另一部分用于储存指向方法区对象类型数据的指针，这一部分先不关注。而“Mark word”就是锁实现的关键，我们以32位的HotSpot举例，32bit的”Mark word”中除了2bit用于存储锁标志位外，其他的30bit所保存的内容都根据锁状态发生变化：

当处于未锁定（标志位为01）时，29bit存储了对象哈希值、对象分代年龄等
需要加锁时，先检查是否处于未锁定状态，如果是，在当前线程的栈帧中建立一个名为锁记录（Lock Record）的空间，用于存储当前锁对象”Mark word”的拷贝
使用CAS操作将其余30bit更新为指向锁记录的指针
- 这些动作成功了，改变标志位（00是轻量级锁），这个线程就拥有了该对象的锁
- 如果失败了，虚拟机会检查对象的Mark word是否指向当前线程，如果是则说明当前线程已经获得锁，则直接进入同步块执行。否则这个锁对象已经被其他的线程抢占了。这时候轻量锁膨胀为重量锁，标志位改为10，Mark word指向重量锁，后面的线程进入阻塞状态。
当执行完同步块，使用CAS操作将对象当前的Mark word与之前存储的老Mark word拷贝进行交换，完成解锁。

偏向锁

偏向锁的优化方向是在不存在竞争时直接去掉同步原语，当锁对象第一次被线程获取的时候，虚拟机会将标志位改为01，即偏向模式，同时使用CAS操作把获取到的锁的线程ID记录在Mark word之中，如果操作成功，这个线程就拥有了该对象的锁。之后当持有偏向锁的线程进入同步块的时候，虚拟机不需要做任何操作，而在轻量锁中，还是需要尝试检查锁定状态，以及对象的Mark word是否指向自己。当有其他线程尝试获取锁的时候，偏向锁就膨胀为轻量锁。偏向锁可以提高带有同步但无竞争的程序性能，但是如果程序中大多数的锁总是被多个不同的线程访问，那偏向模式就是多余的。可以通过-XX:-UseBiasedLocking来进行禁止。

a-14-1

End

参考资料：《深入理解Java虚拟机》

Prev Home Next

锁