Java ConcurrentHashMap的小测试

今天正式开始自己的分布式学习，在第一章介绍多线程工作模式时，作者抛出了一段关于ConcurrentHashMap代码让我很是疑惑，代码如下：

public class TestClass {
    
    private ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<String, Integer>();
    
    public void add(String key){
        Integer value = map.get(key);
        if(value == null){
            map.put(key, 1);
        }else{
            map.put(key, value + 1);
        }
    }
    
}

作者的结论是这样婶的：即使使用线程安全的ConcurrentHashMap来统计信息的总数，依然存在线程不安全的情况。

笔者的结论是这样婶的：ConcurrentHashMap本来就是线程安全的呀，读虽然不加锁，写是会加锁的呀，讲道理的话上面的代码应该没啥问题啊。

既然持怀疑态度，那笔者只有写个测试程序咯，因为伟大的毛主席曾说过：“实践是检验真理的唯一标准” ＝_＝ 

/**
 * @Title: TestConcurrentHashMap.java
 * @Describe:测试ConcurrentHashMap
 * @author: Mr.Yanphet
 * @Email: mr_yanphet@163.com
 * @date: 2016年8月1日 下午4:50:18
 * @version: 1.0
 */
public class TestConcurrentHashMap {

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

    public static void main(String[] args) {
        DoWork dw = new DoWork(map);
        ExecutorService pool = Executors.newFixedThreadPool(8);
        try {
            for (int i = 0; i < 20; i++) {
                pool.execute(new Thread(dw));// 开启20个线程
            }
            Thread.sleep(5000);// 主线程睡眠5s 等待子线程完成任务
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            pool.shutdown();// 关闭线程池
        }
        System.out.println("统计的数量：" + map.get("count"));
    }

    static class DoWork implements Runnable {

        private ConcurrentHashMap<String, Integer> map = null;

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

        @Override
        public void run() {
            add("count");
        }

        public void add(String key) {
            Integer value = map.get(key);// 获取map中的数值
            System.out.println("当前数量" + value);
            if (null == value) {
                map.put(key, 1);// 第一次存放
            } else {
                map.put(key, value + 1);// 以后次存放
            }
        }

    }

}

debug输出一下：

当前数量null
当前数量null
当前数量null
当前数量1
当前数量null
当前数量1
当前数量2
当前数量3
当前数量4
当前数量5
当前数量6
当前数量7
当前数量7
当前数量5
当前数量6
当前数量7
当前数量8
当前数量8
当前数量8
当前数量6
统计的数量：7

这结果并不是20呀，瞬间被打脸有木有啊？满满的心塞有木有啊？

秉承着打破砂锅问到底的精神，必须找到原因，既然打了脸，那就别下次还打脸啊.....

翻开JDK1.6的源码：

public V get(Object key) {
        int hash = hash(key.hashCode());
        return segmentFor(hash).get(key, hash);// segmentFor(hash)用于精确到某个段
    }

map的put方法调用了segment(类似hashtable的结构)的put方法，此外该方法涉及的另外两个方法，笔者一并放在一起分析。

 V get(Object key, int hash) {
            if (count != 0) { // segment存在值 继续往下查找
                HashEntry<K,V> e = getFirst(hash);// 根据hash值定位 相应的链表
                while (e != null) {
                    if (e.hash == hash && key.equals(e.key)) {
                        V v = e.value;
                        if (v != null)
                            return v;// 值不为null 立即返回
                        return readValueUnderLock(e); // 值为null 重新读取该值
                    }
                    e = e.next;// 循环查找链表中的下一个值
                }
            }
            return null;// 如果该segment没有值 直接返回null
        }
// 定位链表
HashEntry<K,V> getFirst(int hash) {
            HashEntry<K,V>[] tab = table;
            return tab[hash & (tab.length - 1)];
        }
// 再次读取为null的值
V readValueUnderLock(HashEntry<K,V> e) {
            lock();
            try {
                return e.value;
            } finally {
                unlock();
            }
        }

这里需要总结一下了：

1 ConcurrentHashMap读取数据不加锁的结论是不正确的，当读取的值为null时，这时候ConcurrentHashMap是会加锁再次读取该值的(上面粗体部分)。至于读到null就加锁再读的原因如下：

　 ConcurrentHashMap的put方法value是不能为null的(稍后代码展示)，现在get值为null，那么可能有另外一个线程正在改变该值(比如remove)，为了读取到正确的值，所以采取加锁再读的方法。在此对Doug Lee大师的逻辑严密性佩服得五体投地啊有木有......

2 读者大概也知道为啥不是20了吧，虽然put加锁控制了线程的执行顺序，但是get没有锁，也就是多个线程可能拿到相同的值，然后相同的值＋1，结果就不是预期的20了。

既然知道了原因，那么修改一下add(String key)这个方法，加锁控制它get的顺序即可。

public void add(String key) {
    lock.lock();
    try {
        Integer value = map.get(key);
        System.out.println("当前数量" + value);
        if (null == value) {
            map.put(key, 1);
        } else {
            map.put(key, value + 1);
        }
    } finally {
        lock.unlock();
    }
}

再次debug输出一下：

当前数量null
当前数量1
当前数量2
当前数量3
当前数量4
当前数量5
当前数量6
当前数量7
当前数量8
当前数量9
当前数量10
当前数量11
当前数量12
当前数量13
当前数量14
当前数量15
当前数量16
当前数量17
当前数量18
当前数量19
统计的数量：20

得到正确的结果。

 附上put方法的源码：

public V put(K key, V value) {
     if (value == null)
           throw new NullPointerException();// 值为空抛出NullPointerException异常
     int hash = hash(key.hashCode());// 根据key的hashcode 然后获取hash值
     return segmentFor(hash).put(key, hash, value, false); //定位到某个segment
}

V put(K key, int hash, V value, boolean onlyIfAbsent) {
      lock();
      try {
            int c = count;
            if (c++ > threshold) // 该segment总的key-value数量＋ 大于threshold阀值 
                 rehash(); // segment扩容
            HashEntry<K,V>[] tab = table;
            int index = hash & (tab.length - 1);// hash值与数组长度－1取&运算
            HashEntry<K,V> first = tab[index]; // 定位到某个数组元素(头节点)
            HashEntry<K,V> e = first;// 头节点
            while (e != null && (e.hash != hash || !key.equals(e.key)))
                  e = e.next;

            V oldValue;
            if (e != null) {// 找到key 替换旧值
                oldValue = e.value;
                if (!onlyIfAbsent)
                    e.value = value;
            }else {// 未找到key 生成节点
                oldValue = null;
                ++modCount;
                tab[index] = new HashEntry<K,V>(key, hash, first, value);
                count = c; // write-volatile
            }
            return oldValue;
        } finally {
          unlock();
      }
 }

程序员最怕对技术似懂非懂，在此与君共勉，慎之戒之！！！