你的位置:首页 > Java教程

[Java教程]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();   } }

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