一、介绍

二、 原理与实现

2.1 Java7 中的实现

ConcurrentHashMap 采用了分段锁技术，其中 Segment 继承于 ReentrantLock（可重入锁）。不会像 HashTable 那样不管是 put 还是 get 操作都需要做同步处理，理论上 ConcurrentHashMap 支持 CurrencyLevel (Segment 数组数量)的线程并发。每当一个线程占用锁访问一个 Segment 时，不会影响到其他的 Segment。

  • 数据结构图示

image.png

  • 成员变量
  1. //定义的常量
  2. //初始时默认容量
  3. static final int DEFAULT_INITIAL_CAPACITY = 16;
  4. //负载因子
  5. static final float DEFAULT_LOAD_FACTOR = 0.75f;
  6. //默认的并发等级，
  7. static final int DEFAULT_CONCURRENCY_LEVEL = 16;
  8. //最大容量
  9.  static final int MAXIMUM_CAPACITY = 1 << 30;
  10. //最小每个Segment持有table数量，必须是2的倍数
  11. static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
  12. //Segment 数组最大容量　65536
  13. static final int MAX_SEGMENTS = 1 << 16;
  14. //不加锁进行检索的数量
  15. static final int RETRIES_BEFORE_LOCK = 2;
  17. //Segment 数组, 数组中的每个元素都持有HashEntry 桶
  18. final Segment<K,V>[] segments;
  20. transient Set<K> keySet;
  21. transient Set<Map.Entry<K,V>> entrySet;
  22. transient Collection<V> values;
  • Segment 的源码实现
  1. static final class Segment<K,V> extends ReentrantLock implements Serializable {
  2.         private static final long serialVersionUID = 2249069246763182397L;
  4.         static final int MAX_SCAN_RETRIES =
  5.             Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
  6.         //存放数据的hash 桶
  7.         transient volatile HashEntry<K,V>[] table;
  9.         transient int count;
  11.         transient int modCount;
  13.         transient int threshold;
  15.         final float loadFactor;
  17.         Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
  18.             this.loadFactor = lf;
  19.             this.threshold = threshold;
  20.             this.table = tab;
  21.         }
  22.     }
  • Entry 实现
  1. static final class HashEntry<K,V> {
  2.         final int hash; //hahs值
  3.         final K key;　//键
  4.         volatile V value;　//值
  5.         volatile HashEntry<K,V> next;　　//后继指针
  6. 　    HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
  7.             this.hash = hash;
  8.             this.key = key;
  9.             this.value = value;
  10.             this.next = next;
  11.         }
  12. }
  • 构造函数
  1.  public ConcurrentHashMap(int initialCapacity,
  2.                              float loadFactor, int concurrencyLevel) {
  3.         if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
  4.             throw new IllegalArgumentException();
  5.         if (concurrencyLevel > MAX_SEGMENTS)
  6.             concurrencyLevel = MAX_SEGMENTS;
  7.         // Find power-of-two sizes best matching arguments
  8.         int sshift = 0; //sshift等于ssize从1向左移位的次数
  9.         int ssize = 1; //Segment 数组的大小
  10.      //为了能通过按位与的散列算法来定位segments数组的索引,必须保证segments数组的长度是2的N次方
  11. //(power-of-two size),所以必须计算出一个大于或等于concurrencyLevel的最小的2的N次方值
  12. //来作为segments数组的长度。
  13.         while (ssize < concurrencyLevel) {
  14.             ++sshift;
  15.             ssize <<= 1;
  16.         }
  17.      //segmentShift用于定位参与散列运算的位数
  18.         this.segmentShift = 32 - sshift;
  19.      //segmentMask是散列运算的掩码,等于ssize减1,即15,掩码的二进制各个位的值都是1
  20.         this.segmentMask = ssize - 1;
  21.         if (initialCapacity > MAXIMUM_CAPACITY)
  22.             initialCapacity = MAXIMUM_CAPACITY;
  23.         int c = initialCapacity / ssize;
  24.         if (c * ssize < initialCapacity)
  25.             ++c;
  26.         int cap = MIN_SEGMENT_TABLE_CAPACITY;
  27.         while (cap < c)
  28.             cap <<= 1;
  29.         //创建 Segment,并放入Segment数组        
  30.         Segment<K,V> s0 =
  31.             new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
  32.                              (HashEntry<K,V>[])new HashEntry[cap]);
  33.         Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
  34.         UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
  35.         this.segments = ss;
  36.     }
  • get 方法
  1. public V get(Object key) {
  2.         Segment<K,V> s; // manually integrate access methods to reduce overhead
  3.         HashEntry<K,V>[] tab;
  4.         //对key 进行散列，得到hash值
  5.         int h = hash(key);
  6.         //计算出key 所在的segments数组下标
  7.         long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
  8.         if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
  9.             (tab = s.table) != null) {
  10.             //遍历桶中的元素，找到key对应的的元素
  11.             for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
  12.                      (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
  13.                  e != null; e = e.next) {
  14.                 K k;
  15.                 if ((k = e.key) == key || (e.hash == h && key.equals(k)))
  16.                     return e.value;
  17.             }
  18.         }
  19.         return null;
  20.     }

get操作的高效之处在于整个get过程不需要加锁

  • put 方法了解
  1. public V put(K key, V value) {
  2.         Segment<K,V> s;
  3.         if (value == null)
  4.             throw new NullPointerException();
  5.         //对key 进行散列
  6.         int hash = hash(key);
  7.         //计算存放到哪个Segment
  8.         int j = (hash >>> segmentShift) & segmentMask;
  9.         if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
  10.              (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
  11.             s = ensureSegment(j);
  12.         return s.put(key, hash, value, false);
  13.     }
  1. //如果不存在，创建Segment,并返回
  2. private Segment<K,V> ensureSegment(int k) {
  3.         final Segment<K,V>[] ss = this.segments;
  4.         long u = (k << SSHIFT) + SBASE; // raw offset
  5.         Segment<K,V> seg;
  6.         if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
  7.             Segment<K,V> proto = ss[0]; // use segment 0 as prototype
  8.             int cap = proto.table.length;
  9.             float lf = proto.loadFactor;
  10.             int threshold = (int)(cap * lf);
  11.             HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
  12.             if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
  13.                 == null) { // recheck
  14.                 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
  15.                 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
  16.                        == null) {
  17.                     if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
  18.                         break;
  19.                 }
  20.             }
  21.         }
  22.         return seg;
  23.     }

找到对应的Segment，执行put 方法

  1. final V put(K key, int hash, V value, boolean onlyIfAbsent) {
  2.             HashEntry<K,V> node = tryLock() ? null : //1 
  3.                 scanAndLockForPut(key, hash, value); //2
  4.             V oldValue;
  5.             try {
  6.                 HashEntry<K,V>[] tab = table;
  7.                 int index = (tab.length - 1) & hash;
  8.                 HashEntry<K,V> first = entryAt(tab, index); //3
  9.                 for (HashEntry<K,V> e = first;;) {
  10.                     if (e != null) {
  11.                         K k;
  12.                         if ((k = e.key) == key ||
  13.                             (e.hash == hash && key.equals(k))) {//4
  14.                             oldValue = e.value;
  15.                             if (!onlyIfAbsent) {
  16.                                 e.value = value;
  17.                                 ++modCount;
  18.                             }
  19.                             break;
  20.                         }
  21.                         e = e.next;
  22.                     }
  23.                     else {//5
  24.                         if (node != null)
  25.                             node.setNext(first);
  26.                         else
  27.                             node = new HashEntry<K,V>(hash, key, value, first);
  28.                         int c = count + 1;
  29.                         if (c > threshold && tab.length < MAXIMUM_CAPACITY)
  30.                             rehash(node);
  31.                         else
  32.                             setEntryAt(tab, index, node);
  33.                         ++modCount;
  34.                         count = c;
  35.                         oldValue = null;
  36.                         break;
  37.                     }
  38.                 }
  39.             } finally {
  40.                 unlock(); //6
  41.             }
  42.             return oldValue;
  43.         }
  1. 首先第一步的时候会尝试获取锁: tryLock()
  2. 如果获取失败肯定就有其他线程存在竞争，则利用 scanAndLockForPut() 自旋获取锁。
  3. 将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry。
  4. 遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value。
  5. 不为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。
  6. 最后会解除在 1 中所获取当前 Segment 的锁。
  • scanAndLockForPut方法
  1.    private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
  2.             HashEntry<K,V> first = entryForHash(this, hash);
  3.             HashEntry<K,V> e = first;
  4.             HashEntry<K,V> node = null;
  5.             int retries = -1; // negative while locating node
  6.             while (!tryLock()) { //1 
  7.                 HashEntry<K,V> f; // to recheck first below
  8.                 if (retries < 0) {
  9.                     if (e == null) {
  10.                         if (node == null) // speculatively create node
  11.                             node = new HashEntry<K,V>(hash, key, value, null);
  12.                         retries = 0;
  13.                     }
  14.                     else if (key.equals(e.key))
  15.                         retries = 0;
  16.                     else
  17.                         e = e.next;
  18.                 }
  19.                 else if (++retries > MAX_SCAN_RETRIES) {//2 
  20.                     lock();
  21.                     break;
  22.                 }
  23.                 else if ((retries & 1) == 0 &&
  24.                          (f = entryForHash(this, hash)) != first) {
  25.                     e = first = f; // re-traverse if entry changed
  26.                     retries = -1;
  27.                 }
  28.             }
  29.             return node;
  30.         }
  1. 尝试自旋获取锁。
  2. 如果重试的次数达到了 MAX_SCAN_RETRIES 则改为阻塞锁获取，保证能获取成功。
  • rehash方法
  1. private void rehash(HashEntry<K,V> node) {
  3.             HashEntry<K,V>[] oldTable = table;
  4.             int oldCapacity = oldTable.length;
  5.             int newCapacity = oldCapacity << 1;　//1
  6.             threshold = (int)(newCapacity * loadFactor);
  7.             HashEntry<K,V>[] newTable =
  8.                 (HashEntry<K,V>[]) new HashEntry[newCapacity];
  9.             int sizeMask = newCapacity - 1;
  10.             for (int i = 0; i < oldCapacity ; i++) {
  11.                 HashEntry<K,V> e = oldTable[i];
  12.                 if (e != null) {
  13.                     HashEntry<K,V> next = e.next;
  14.                     int idx = e.hash & sizeMask;
  15.                     if (next == null)   //  Single node on list  //2
  16.                         newTable[idx] = e;
  17.                     else { // Reuse consecutive sequence at same slot //3
  18.                         HashEntry<K,V> lastRun = e;
  19.                         int lastIdx = idx;
  20.                         for (HashEntry<K,V> last = next;
  21.                              last != null;
  22.                              last = last.next) {
  23.                             int k = last.hash & sizeMask;
  24.                             if (k != lastIdx) {
  25.                                 lastIdx = k;
  26.                                 lastRun = last;
  27.                             }
  28.                         }
  29.                         newTable[lastIdx] = lastRun;
  30.                         // Clone remaining nodes 
  31.                         for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
  32.                             V v = p.value;
  33.                             int h = p.hash;
  34.                             int k = h & sizeMask;
  35.                             HashEntry<K,V> n = newTable[k];
  36.                             newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
  37.                         }
  38.                     }
  39.                 }
  40.             }
  41.             int nodeIndex = node.hash & sizeMask; // add the new node
  42.             node.setNext(newTable[nodeIndex]);
  43.             newTable[nodeIndex] = node;
  44.             table = newTable;
  45.         }
  1. 计算新的容量为旧容量的2倍
  2. 遍历旧HashEntry桶，如果当前HashEntry只用一个节点，直接放到新的HashEntry桶中
  3. 如果当前HashEntry是链表，则遍历链表，重新计算下标放到新的HashEntry桶中

2.2、 Java8 中的实现

  • 数据结构图示

抛弃了原有的 Segment 分段锁，而采用了 CAS + synchronized 来保证并发安全性。结构上也引入了红黑树，防止查询效率退化为O(N)
image.png

  • Node类与Java7　HashEntry类似
  1. static class Node<K,V> implements Map.Entry<K,V> {
  2.         final int hash;
  3.         final K key;
  4.         volatile V val;　//volatile保证可见性
  5.         volatile Node<K,V> next;
  7.         Node(int hash, K key, V val, Node<K,V> next) {
  8.             this.hash = hash;
  9.             this.key = key;
  10.             this.val = val;
  11.             this.next = next;
  12.         }
  13. }
  • get方法
  1.  public V get(Object key) {
  3.         Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
  4.         int h = spread(key.hashCode());
  5.         if ((tab = table) != null && (n = tab.length) > 0 &&
  6.             (e = tabAt(tab, (n - 1) & h)) != null) {
  7.              //根据计算出来的 hashcode 寻址，如果就在桶上那么直接返回值。
  8.             if ((eh = e.hash) == h) {
  9.                 if ((ek = e.key) == key || (ek != null && key.equals(ek)))
  10.                     return e.val;
  11.             }
  12.             else if (eh < 0)//如果是红黑树那就按照树的方式获取值。
  13.                 return (p = e.find(h, key)) != null ? p.val : null;
  14.             while ((e = e.next) != null) { 就不满足那就按照链表的方式遍历获取值。
  15.                 if (e.hash == h &&
  16.                     ((ek = e.key) == key || (ek != null && key.equals(ek))))
  17.                     return e.val;
  18.             }
  19.         }
  20.         return null;
  21.     }
  • put 方法
  1. final V putVal(K key, V value, boolean onlyIfAbsent) {
  2.         if (key == null || value == null) throw new NullPointerException();
  3.         int hash = spread(key.hashCode());
  4.         int binCount = 0;
  5.         for (Node<K,V>[] tab = table;;) {
  6.             Node<K,V> f; int n, i, fh;
  7.             //如果桶为空，初始化
  8.             if (tab == null || (n = tab.length) == 0)
  9.                 tab = initTable();
  10.             else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  11.                 //采用CAS无锁put入新的元素，成功返回
  12.                 //失败自旋
  13.                 if (casTabAt(tab, i, null,
  14.                              new Node<K,V>(hash, key, value, null)))
  15.                     break;                   // no lock when adding to empty bin
  16.             }
  17.             //如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。
  18.             else if ((fh = f.hash) == MOVED)
  19.                 tab = helpTransfer(tab, f);
  20.             else {
  21.                 //如果都不满足，则利用 synchronized 锁写入数据。
  22.                 V oldVal = null;
  23.                 synchronized (f) {
  24.                     if (tabAt(tab, i) == f) {
  25.                         if (fh >= 0) {
  26.                             binCount = 1;
  27.                             for (Node<K,V> e = f;; ++binCount) {
  28.                                 K ek;
  29.                                 if (e.hash == hash &&
  30.                                     ((ek = e.key) == key ||
  31.                                      (ek != null && key.equals(ek)))) {
  32.                                     oldVal = e.val;
  33.                                     if (!onlyIfAbsent)
  34.                                         e.val = value;
  35.                                     break;
  36.                                 }
  37.                                 Node<K,V> pred = e;
  38.                                 if ((e = e.next) == null) {
  39.                                     pred.next = new Node<K,V>(hash, key,
  40.                                                               value, null);
  41.                                     break;
  42.                                 }
  43.                             }
  44.                         }
  45.                         else if (f instanceof TreeBin) {
  46.                             Node<K,V> p;
  47.                             binCount = 2;
  48.                             if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  49.                                                            value)) != null) {
  50.                                 oldVal = p.val;
  51.                                 if (!onlyIfAbsent)
  52.                                     p.val = value;
  53.                             }
  54.                         }
  55.                     }
  56.                 }
  57.                 if (binCount != 0) {
  58.                     //如果达到需要转换为红黑树的阀值 TREEIFY_THRESHOLD = 8
  59.                     if (binCount >= TREEIFY_THRESHOLD)
  60.                         treeifyBin(tab, i);//将链表转换为红黑树
  61.                     if (oldVal != null)
  62.                         return oldVal;
  63.                     break;
  64.                 }
  65.             }
  66.         }
  67.         addCount(1L, binCount);
  68.         return null;
  69.     }

三、 示例

四、参考