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

博客原文 一条面试题引发的思考--浅谈 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 行的操作做了这三件事，如下:

1. 线程读取 index 的值
2. 比较 index 的值是否大于 30
3. 如果小于 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时输出, 不加锁, 不符合条件的线程作自旋。根据思路整理代码如下:

pubic 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(); } } }); } } 

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