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

2019-03-21 15:52:51 +08:00
 samray

博客原文 一条面试题引发的思考--浅谈 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时输出, 不加锁, 不符合条件的线程作自旋。根据思路整理代码如下:

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

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

5862 次点击
所在节点    程序员
55 条回复
ToDyZHu
2019-03-22 00:08:06 +08:00
大三的我仍不会做这题怎么办。。。
scnace
2019-03-22 01:47:03 +08:00
@xrlin 终于看到一个 channel 版本的了🌚👍
scnace
2019-03-22 01:56:16 +08:00
这难道就是传说中的睡排序升级版?
NSVitus
2019-03-22 10:12:21 +08:00
回复的代码不排版下吗?
ydc886
2019-03-22 11:26:26 +08:00
public class Key implements Runnable {

public final String s;
public volatile static String flag = "A";
public static volatile int count = 0;
public ReentrantLock lock;
public Condition condition;

public static void main(String[] args) {

final ReentrantLock lock = new ReentrantLock();
new Thread(new Key("A", lock), "Thread-A").start();
new Thread(new Key("B", lock), "Thread-B").start();
new Thread(new Key("C", lock), "Thread-C").start();
}

public Key(final String s, final ReentrantLock lock) {
this.s = s;
this.lock = lock;
this.condition = lock.newCondition();
}

@Override
public void run() {
while (true) {
lock.lock();
try {

while (!flag.equalsIgnoreCase(s)) {
try {
condition.await(10, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
if (count > 29) {
System.exit(0);
}

System.out.print(s + "-" + (count++) + "; ");

switch (flag) {
case "A":
flag = "B";
break;
case "B":
flag = "C";
break;
case "C":
flag = "A";
break;
default:
throw new RuntimeException("");
}
condition.signalAll();
}
finally {
lock.unlock();
}
}
}
}
---

这种实现可行吗?
Graypn
2019-03-22 11:42:33 +08:00
二次判断 >= 30 的时候,是不是要先 unlock() ,再 break,释放资源,要不然进程没办法结束
rwdy2008
2019-03-22 12:45:02 +08:00
@xrlin 你说的很对,感谢
ydc886
2019-03-22 14:15:45 +08:00
cyhulk
2019-03-22 14:37:15 +08:00
稍微修改下,加上 volatile 和 lock 进入的位置,应该就可以了
public class ThreadTest {
private static volatile 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 (true) {
lock.lock();
if (index >= 30) {
break;
}
if (condition.test(index)) {
function.run();
index++;
}
lock.unlock();
}
});
}
}
ZiLong
2019-03-22 14:40:12 +08:00
@ydc886 目测,大佬喊过不要再 while 里面 lock,提到 while 外面去
wdmx007
2019-03-22 15:22:03 +08:00
来一发 java 升级版的,求大佬们指教

https://gist.github.com/woodyDM/fa1f453f0564e873f561fbb3b09159b0
bwangel
2019-03-25 10:10:01 +08:00
https://gist.github.com/bwangelme/b44b427621b28f1a17b0cd7a3e6e94ae

楼主,是这样的么,检查条件,i++ 都要上锁
yehen
2019-03-29 13:10:24 +08:00
package com.ljw.HelloJava;

import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;

public class ABCThreads {
private static Integer index = 0;
private static Integer max = 6;
private static Object lock = new Object();

public static void main(String[] args) {

Thread a = getThread(i -> i % 3 == 0, "A");
Thread b = getThread(i -> i % 3 == 1, "B");
Thread c = getThread(i -> i % 3 == 2, "C");
a.start();
b.start();
c.start();

}

private static Thread getThread(Predicate<Integer> condition, String value) {
return new Thread(() -> {
while (true) {
synchronized (lock) {
while (!condition.test(index)) {
try {
lock.wait();
} catch (InterruptedException e) {
System.out.println(e.getMessage());
}
}
if (index >= max) {
return;
}
System.out.printf("index:%s,value:%s\n", index, value);
index++;
lock.notifyAll();
}
}
});
}
}
oaix
2019-04-07 12:02:07 +08:00
coffeSlider
2019-04-07 23:28:26 +08:00
第一反应就是用 AtomicInteger 控制。

这是一个专为移动设备优化的页面(即为了让你能够在 Google 搜索结果里秒开这个页面),如果你希望参与 V2EX 社区的讨论,你可以继续到 V2EX 上打开本讨论主题的完整版本。

https://tanronggui.xyz/t/547045

V2EX 是创意工作者们的社区,是一个分享自己正在做的有趣事物、交流想法,可以遇见新朋友甚至新机会的地方。

V2EX is a community of developers, designers and creative people.

© 2021 V2EX