ConcurrentHashMap的putIfAbsent方法可能忽视的问题

背景

ConcurrentHashMap是一个线程安全的Map，正因为它是线程安全的Map所以在使用时不注意也很可能带来问题。在业务上我们经常会遇到一种情况就是通过一个线程安全的Map来存储一个唯一的实例，如果对象本身是单例模式并不会有什么问题，但是如果对象并不是单例模式，则需要保证在多线程下只能创建一个。下面这段示例代码能保证A实例只创建一个吗？

public class App {

    private static Map<String,Object> map = new ConcurrentHashMap<>();

    public static void main(String[] args) throws InterruptedException {
        Runnable r = new Runnable() {
            @Override
            public void run() {
                Object a = map.putIfAbsent("a", new A());
                System.out.printf("线程[%s],A实例[%s]%n",Thread.currentThread().getName(),a);
            }
        };

        Thread t1 = new Thread(r);
        Thread t2 = new Thread(r);
        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println(map.get("a"));

    }
}

class A{

    public A() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.printf("线程[%s]创建A实例对象[%s]%n",Thread.currentThread().getName(),this);
    }
}

运行上面的代码结果如下:

线程[Thread-0]创建A实例对象[com.buydeem.cs.A@5fed5d65]
线程[Thread-0],A实例[null]
线程[Thread-1]创建A实例对象[com.buydeem.cs.A@70bea8b8]
线程[Thread-1],A实例[com.buydeem.cs.A@5fed5d65]
com.buydeem.cs.A@5fed5d65

从结果可以看出实际上A实例创建了2个，并没有达到我们预想中的结果。其实问题的关键点就出现在putIfAbsent方法上。

原因

Map的putIfAbsent方法它的效果如下所示:

if (!map.containsKey(key))
  return map.put(key, value);
else
  return map.get(key);

该方法是一个原子操作，具体可以查看ConcurrentMap中该方法的定义。但是Object a = map.putIfAbsent("a", new A());并不是一个原子操作。它实际上是两个操作：

A a = new A();
map.put("a",a);

如何解决

问题找到了，那么该如何解决呢？最简单的方式直接使用synchronized来解决，但是这并不是最好的一种解决方式。其实我们可以使用一个轻量级的自旋锁来解决这个问题。

public class App {

    private static Map<String,Object> map = new ConcurrentHashMap<>();

    private static Map<String,SpinStatus> spinStatusMap = new ConcurrentHashMap<>();

    public static void main(String[] args) throws InterruptedException {
        Runnable r = new Runnable() {
            @Override
            public void run() {
                //获取A实例
                Object a = map.get("a");
                if (a == null){
                    //为当前线程创建一个自旋状态
                    SpinStatus status = new SpinStatus();
                    //将自旋状态放入到map中，如果放入成功则代表获取到锁，未放入成功则说明没有拿到锁
                    SpinStatus oldStatus = spinStatusMap.putIfAbsent("a", status);
                    if (oldStatus == null){
                        //说明当前线程拿到了锁
                        a = new A();
                        map.put("a",a);
                        //将自旋锁状态设置为释放
                        status.released = true;
                    }else {
                        //没有拿到自旋锁,则一直自旋等待锁释放
                        while (!oldStatus.released){}
                    }
                }
            }
        };

        Thread t1 = new Thread(r);
        Thread t2 = new Thread(r);
        t1.start();
        t2.start();

        t1.join();
        t2.join();

        System.out.println(map.get("a"));

    }

    /**
     * 记录自旋状态对象
     */
    static class SpinStatus{
        /**
         * 是否释放。注意使用volatile来修饰
         */
        volatile boolean released;
    }
}

class A{

    public A() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.printf("线程[%s]创建A实例对象[%s]%n",Thread.currentThread().getName(),this);
    }
}

再次运行代码，运行结果如下:

线程[Thread-1]创建A实例对象[com.buydeem.cs.A@31ec4de7]
com.buydeem.cs.A@31ec4de7

从结果可以看出，只有一个线程创建了A实例。该方式的主要逻辑就是通过自旋来保证实例A只创建一个。在创建A实例之前我们先创建一个自旋锁对象放入到ConcurrentHashMap中，如果放入成功则代表了拿到了锁，而放入失败则代表没有拿到锁。成功或失败的判断标志就是通过putIfAbsent的结果来判断，如果返回null则代表放入成功，如果有返回值则说明没有放入成功。而没有放入成功的线程就根据自旋锁的自旋状态自旋直到自旋锁释放。