ConcurrentHashMap、synchronized與線程安全

原文：https://blog.csdn.net/sadfishsc/article/details/42394955

最近作的項目中遇到一個問題：明明用了ConcurrentHashMap，但是始終線程不安全

除去項目中的業務邏輯，簡化後的代碼以下：

public class Test40 {

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            System.out.println(test());
        }
    }

    private static int test() throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 0; i < 8; i++) {
            pool.execute(new MyTask(map));
        }
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.DAYS);

        return map.get(MyTask.KEY);
    }

}

class MyTask implements Runnable {

    public static final String KEY = "key";

    private ConcurrentHashMap<String, Integer> map;

    public MyTask(ConcurrentHashMap<String, Integer> map) {
        this.map = map;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            this.addup();
        }
    }

    private void addup() {
        if (!map.containsKey(KEY)) {
            map.put(KEY, 1);
        } else {
            map.put(KEY, map.get(KEY) + 1);
        }    
    }

}

測試代碼跑了10次，每次都不是800。這就很讓人疑惑了，難道ConcurrentHashMap的線程安全性失效了？
查了一些資料後發現，原來ConcurrentHashMap的線程安全指的是，它的每一個方法單獨調用（即原子操做）都是線程安全的，可是代碼整體的互斥性並不受控制。以上面的代碼爲例，最後一行中的：

map.put(KEY, map.get(KEY) + 1);

實際上並非原子操做，它包含了三步：

1. map.get
2. 加1
3. map.put

其中第1和第3步，單獨來講都是線程安全的，由ConcurrentHashMap保證。可是因爲在上面的代碼中，map自己是一個共享變量。當線程A執行map.get的時候，其它線程可能正在執行map.put，這樣一來當線程A執行到map.put的時候，線程A的值就已是髒數據了，而後髒數據覆蓋了真值，致使線程不安全。

簡單地說，ConcurrentHashMap的get方法獲取到的是此時的真值，但它並不保證當你調用put方法的時候，當時獲取到的值仍然是真值。

爲了使上面的代碼變得線程安全，我引入了synchronized關鍵字來修飾目標方法，以下：

public class Test40 {
    
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            System.out.println(test());
        }
    }

    private static int test() throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 0; i < 8; i++) {
            pool.execute(new MyTask(map));
        }
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.DAYS);

        return map.get(MyTask.KEY);
    }
    
}

class MyTask implements Runnable {
    
    public static final String KEY = "key";

    private ConcurrentHashMap<String, Integer> map;

    public MyTask(ConcurrentHashMap<String, Integer> map) {
        this.map = map;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            this.addup();
        }
    }

    private synchronized void addup() { // 用關鍵字synchronized修飾addup方法
        if (!map.containsKey(KEY)) {
            map.put(KEY, 1);
        } else {
            map.put(KEY, map.get(KEY) + 1);
        }
    }
}

運行以後仍然是線程不安全的，難道synchronized也失效了？
查閱了synchronized的資料後，原來，無論synchronized是用來修飾方法，仍是修飾代碼塊，其本質都是鎖定某一個對象。修飾方法時，鎖上的是調用這個方法的對象，即this；修飾代碼塊時，鎖上的是括號裏的那個對象。

在上面的代碼中，很明顯就是鎖定的MyTask對象自己。可是因爲在每個線程中，MyTask對象都是獨立的，這就致使實際上每一個線程都對本身的MyTask進行鎖定，而並不會干涉其它線程的MyTask對象。換言之，上鎖壓根沒有意義。

理解到這點以後，對上面的代碼又作了一次修改：

public class Test40 {
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            System.out.println(test());
        }
    }

    private static int test() throws InterruptedException {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 0; i < 8; i++) {
            pool.execute(new MyTask(map));
        }
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.DAYS);

        return map.get(MyTask.KEY);
    }
}

class MyTask implements Runnable {
    public static final String KEY = "key";

    private ConcurrentHashMap<String, Integer> map;

    public MyTask(ConcurrentHashMap<String, Integer> map) {
        this.map = map;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            synchronized (map) { // 對共享對象map上鎖
                this.addup();
            }
        }
    }

    private void addup() {
        if (!map.containsKey(KEY)) {
            map.put(KEY, 1);
        } else {
            map.put(KEY, map.get(KEY) + 1);
        }
    }
}

此時在調用addup時直接鎖定map，因爲map是被全部線程共享的，於是達到了讓全部線程互斥的目的，線程安全達成。
修改後，ConcurrentHashMap的做用就不大了，能夠直接將代碼中的map換成普通的HashMap，以減小由ConcurrentHashMap帶來的鎖開銷。

最後特別補充的是，synchronized關鍵字判斷對象是不是它屬於鎖定的對象，本質上是經過 == 運算符來判斷的。換句話說，上面的代碼中，能夠採用任何一個常量，或者每一個線程都共享的變量，或者MyTask類的靜態變量，來代替map。只要該變量與synchronized鎖定的目標變量相同（==），就可使synchronized生效。

綜上，代碼最終能夠修改成：

public class Test40 {
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 100; i++) {
            System.out.println(test());
        }
    }

    private static int test() throws InterruptedException {
        Map<String, Integer> map = new HashMap<String, Integer>();
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 0; i < 8; i++) {
            pool.execute(new MyTask(map));
        }
        pool.shutdown();
        pool.awaitTermination(1, TimeUnit.DAYS);

        return map.get(MyTask.KEY);
    }
}

class MyTask implements Runnable {
    public static Object lock = new Object();

    public static final String KEY = "key";

    private Map<String, Integer> map;

    public MyTask(Map<String, Integer> map) {
        this.map = map;
    }

    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            synchronized (lock) {
                this.addup();
            }
        }
    }

    private void addup() {
        if (!map.containsKey(KEY)) {
            map.put(KEY, 1);
        } else {
            map.put(KEY, map.get(KEY) + 1);
        }
    }
}