一条面试题引发的思考--浅谈 Java 公平锁与内存模型

前言

春天来了，春招还会远么? 又到了春招的季节，随之而来的是各种的面试题。今天就看到组内大佬面试实习生的一道 Java 题目:

编写一个程序，开启 3 个线程 A,B,C，这三个线程的输出分别为 A、B、C，每个线程将自己的输出在屏幕上打印 10 遍，要求输出的结果必须按顺序显示。如：ABCABCABC....

掉进坑里

出于好奇的心态，我花了点时间来尝试解决这个问题, 主要的难点是让线程顺序地如何顺序地输出，线程之间如何交换。很快就按着思路写出了一个版本，用 Lock 来控制线程的顺序，A,B,C 线程依次启动，因为 A 线程先启动，所以 A 线程会最先拿到锁，B,C 阻塞；但是 A 输出完字符串，释放锁，B 线程获得锁，C,A 线程阻塞; 依此循环:

public void Test(){
    private static Integer index = 0;

    Lock lock = new ReentrantLock();
	
	@Test
    public void testLock(){
        Thread threadA = work(i -> i % 3 == 0, () -> System.out.println("A"));
        Thread threadB = work(i -> i % 3 == 1, () -> System.out.println("B"));
        Thread threadC = work(i -> i % 3 == 2, () -> System.out.println("C"));
        threadA.start();
        threadB.start();
        threadC.start();
    }

    private Thread work(Predicate<Integer> condition, Runnable function) {
        return new Thread(() -> {
            while (index < 30) {
                lock.lock();
                if (condition.test(index)) {
                    function.run();
                    index++;
                }
                lock.unlock();
            }
        });
    }
}

输入结果如我预期那般，ABCABC 交替输出，也成功输出了 10 次，奇怪的是 A,B 却多输出了一次？

为什么会多输出一次，不是应该恰好是输出 30 次么, 为什么会多输出一次 A,B 真的百思不得其解. 所以我把index 也打印出来查看, 结果相当奇怪:

...
function.run();
System.out.println(index);
....

为什么 A 会是 30, B 会是 31, 不是有(index.intvalue<30) 的条件判断么, 为什么还会出现这样的数据？灵异事件?

解惑

灵异事件自然是不存在的，仔细分析了一番代码之后，发现了问题：

while (index.intValue() < 30) {  // 1
    lock.lock(); // 2
    if (condition.test(index.intValue())) {
        function.run();
        index++;
    }
    lock.unlock();
}

将 1，2 行的操作做了这三件事，如下:

线程读取 index 的值
比较 index 的值是否大于 30
如果小于 30, 尝试获取锁

换言之，当 index=29 时，线程 C 持有锁，但是锁只能阻止线程 A,线程 B 修改 index 的值，并不能阻止线程 A, 线程 B 在获取锁之前读取 index 的值，所以线程 A 读取 index=29, 并把值保持到线程的内部，如下图:

当线程 C 执行完，还没释放锁的时候，线程 A 的 index 值为 29 ；当线程 C 释放锁，线程 A 获取锁，进入同步块的时候，因为 Java 内存模型有内存可见性的要求, 兼之 Lock 的实现类实现了内存可见，所以线程 A 的 index 值会变成 30, 这就解析了为什么线程 A index=30 的时候能跳过(index.intValue<30)的判断条件，因为执行这个判断条件的时候线程 A index=29, 进入同步块之后变成了 30:

CPU 缓存

为什么每个线程都会持有一个 index 值呢？来看看下面的分级缓存图:

 _______________    ______________  
 |     CPU 1     |  |     CPU 2    |  
 |   _________   |  |   _________  |  
 |  | Level 1 |  |  |  | Level 1 | |  
 |  |   Cache |  |  |  |  Cache  | |  
 |  |         |  |  |  |         | |
 |  |_________|  |  |  |_________| |  
 |_______________|  |______________|
           | |              | |
           | |              | |
          _|_|______________|_|__
         |                       |
         |      MAIN MEMORY      | 
         |_______________________|

众所周知，在计算机的存储体系中，分布式存储系统比硬盘慢，硬盘比内存跑得慢，内存比 Cpu L3 level Cache 跑得慢，L3 Cache 比 L2 Cache 等等。空间越大，速度越慢，速度越快，空间越小，本质上就是空间与时间的取舍。例如，为了提高效率，就会把预计即将用到的变量index从主存缓存到 CPU 缓存，而 CPU 有多个核心，每个核心都缓存变量index。前文的问题本质就是 CPU 缓存的index与主存index不一致，而内存可见性说的就是强制 CPU 从主存获取变量index, 从而规避缓存不一致的问题

解决方案

把问题剖析清楚之后，解决方案就呼之欲出了:

while (index.intValue() < 30) {  // 1
    lock.lock(); // 2
	if(index>=30){
		continue;
	}
    if (condition.test(index.intValue())) {
        function.run();
        index++;
    }
    lock.unlock();
}

这种解决方法不禁让我想起单例模式里面的双重校验:

public static Singleton getSingleton() {
    if (instance == null) {                         //Single Checked
        synchronized (Singleton.class) {
            if (instance == null) {                 //Double Checked
                instance = new Singleton();
            }
        }
    }
    return instance ;
}

只是当时并不清楚 Double Checked 的作用，究竟解决了什么问题？只是知道不加这条语句就会造成初始化多个示例，的确是需要知其然知其所以然.

公平锁与非公平锁

前文说到，

这个程序是用 Lock 来控制线程的顺序，A,B,C 线程依次启动，因为 A 线程先启动，所以 A 线程会最先拿到锁，B,C 阻塞；但是 A 输出完字符串，释放锁，B 线程获得锁，C,A 线程阻塞; 依此循环。

粗看似乎没什么问题, 但是这里是存在着一个问题: 当线程 A 释放锁的时候，获取锁的是否一定是线程 B, 而不是线程 C, 线程 C 是否能够"插队"抢占锁? 这个就涉及到了公平锁和非公平锁的定义了: 公平锁: 线程 C 不能抢占，只能排队等待线程 B 获取并释放锁非公平锁：线程 C 能抢占，抢到锁之后线程 B 只能继续等(有点惨!)

而 ReentrantLock 默认恰好是非公平锁, 查看源码可知:

/**
     * Creates an instance of {@code ReentrantLock}.
     * This is equivalent to using {@code ReentrantLock(false)}.
     */
    public ReentrantLock() {
        sync = new NonfairSync();
    }

因此为了规避非公平锁抢占的问题, 上述的代码在同步块增加了判断条件:

 if (condition.test(index.intValue())) {
   ....
 }

只有符合条件的线程才能进行操作，否则就是线程自旋.(但是加锁+自旋实现起来，效率不会太高效!)

使用公平锁

如果使用公平锁，也可以不需要上述这样的判断条件，直接让线程顺序排队和唤醒: 通过让ReentrantLock成为公平锁. 方法也很简单, 只需要构造参数加上一个 boolean 值:

/**
     * Creates an instance of {@code ReentrantLock} with the
     * given fairness policy.
     *
     * @param fair {@code true} if this lock should use a fair ordering policy
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

公平锁改造之后的代码如下:

public void Test(){
    private static Integer index = 0;
    Lock lock = new ReentrantLock(true);
  @Test
    public void testLock(){
        Thread threadA = work(() -> System.out.println("A"));
        Thread threadB = work(() -> System.out.println("B"));
        Thread threadC = work(() -> System.out.println("C"));
        threadA.setName("threadA");
        threadA.start();
        threadB.setName("threadB");
        threadB.start();
        threadC.setName("threadC");
        threadC.start();
        System.out.println();
    }

    private Thread work(Runnable function) {
        return new Thread(() -> {
            while (index.intValue() < 30) {
                lock.lock();
                if (index >= 30) {
                    continue;
                }
                function.run();
                index++;
                lock.unlock();
            }
        });
    }
}

后记

组内路过的 @碳素大佬看到我在研究这道问题的时候，给出了不一样的解决方案: 使用 AtomicInteger 作为控制手段，循环三十次，线程 A 在index%3==0时输出，线程 B 在index%3==1时输出, 线程 C 在index%3==2时输出, 不加锁, 不符合条件的线程作自旋。根据思路整理代码如下:

public void Test(){
      private static AtomicInteger index = new AtomicInteger(0);
   @Test
    public void testLock(){
        Thread threadA = work(i -> i % 3 == 0, () -> System.out.println("A"));
        Thread threadB = work(i -> i % 3 == 1, () -> System.out.println("B"));
        Thread threadC = work(i -> i % 3 == 2, () -> System.out.println("C"));
        threadA.setName("threadA");
        threadA.start();
        threadB.setName("threadB");
        threadB.start();
        threadC.setName("threadC");
        threadC.start();
        System.out.println();
    }

    private Thread work(Predicate<Integer> condition, Runnable function) {
        return new Thread(() -> {
            while (index.intValue() < 30) {
                if (condition.test(index.intValue())) {
                    function.run();
                    index.incrementAndGet();
                }
            }
        });
    }
}

就这样，把前文所有存在的问题都完美规避，实现还很优雅，效率目测也比加锁的方式高，碳总牛🍺!!!