ConcurrentHashMap源码解析
前言
concurrentMap 是 hashMap 在并发情况下的替代map， 是线程安全
本文针对1.8的源码分析 问题如何保证线程安全? 读加锁么，写加锁么？ 在jdk1.8与jdk1.7的差别? 为什么key不能为null, value不能为null? 原理概况
通过Node[]数组来保存添加到Map中的键值对 keyhash不冲突，放在数组 Key hash冲突，以链表形式放在数组 链表长度超过8个，又小于64，扩容数组，如果数组长度大于等于64，则转换成树 扩容后，某个链表的数组长度小于6个，则会将该树转为链表 源码结构源码
元素的基本节点 static class Node implements Map.Entry { final int hash; final K key; volatile V val; volatile Node next; Node(int hash, K key, V val, Node next) { this.hash = hash; this.key = key; this.val = val; this.next = next; } // 略
树的节点 static final class TreeNode extends Node { TreeNode parent; // red-black tree links TreeNode left; TreeNode right; TreeNode prev; // needed to unlink next upon deletion boolean red; TreeNode(int hash, K key, V val, Node next, TreeNode parent) { super(hash, key, val, next); this.parent = parent; } // 略 } 重要方法
在ConcurrentHashMap中使用了unSafe方法，通过直接操作内存的方式来保证并发处理的安全性，使用的是硬件的安全机制 /* * 用来返回节点数组的指定位置的节点的原子操作 */ @SuppressWarnings(＂unchecked＂) static final Node tabAt(Node[] tab, int i) { return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); } /* * cas原子操作，在指定位置设定值 */ static final boolean casTabAt(Node[] tab, int i, Node c, Node v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); } put源码 /* * 当添加一对键值对的时候，首先会去判断保存这些键值对的数组是不是初始化了， * 如果没有的话就初始化数组 * 然后通过计算hash值来确定放在数组的哪个位置 * 如果这个位置为空则直接添加，如果不为空的话，则取出这个节点来 * 如果取出来的节点的hash值是MOVED(-1)的话，则表示当前正在对这个数组进行扩容，复制到新的数组，则当前线程也去帮助复制 * 最后一种情况就是，如果这个节点，不为空，也不在扩容，则通过synchronized来加锁，进行添加操作 * 然后判断当前取出的节点位置存放的是链表还是树 * 如果是链表的话，则遍历整个链表，直到取出来的节点的key来个要放的key进行比较，如果key相等，并且key的hash值也相等的话， * 则说明是同一个key，则覆盖掉value，否则的话则添加到链表的末尾 * 如果是树的话，则调用putTreeVal方法把这个元素添加到树中去 * 最后在添加完成之后，会判断在该节点处共有多少个节点（注意是添加前的个数），如果达到8个以上了的话， * 则调用treeifyBin方法来尝试将处的链表转为树，或者扩容数组 */ final V putVal(K key, V value, boolean onlyIfAbsent) { // k,v都不能为空，否则会报空指针异常 if (key == null || value == null) throw new NullPointerException(); // 取得Key的hash值 int hash = spread(key.hashCode()); // 用来计算在这个节点总共有多少个元素，用来控制扩容或者转移树 int binCount = 0; for (Node[] tab = table;;) { Node f; int n, i, fh; // 第一次Put，没有初始化，则初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); // 通过哈希计算出一个表中的位置，因为n是数组的长度，所以(n-1)&hash肯定不会出现数组越界 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果这个位置没有元素的话，则通过cas的方式尝试添加，注意这个时候是没有加锁的 if (casTabAt(tab, i, null, // 创建一个Node添加到数组中去，null表示的是下一个节点为空 new Node(hash, key, value, null))) break; // no lock when adding to empty bin } // 如果检测到某个节点的hash值是MOVED,则表示正在进行数组扩张的数据复制阶段 // 则当前线程也会参与去复制，通过允许多线程复制的功能，一次来减少数组的复制所带来的性能损失 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { /* * 如果这个位置有元素的话，就采用synchronized的方式加锁 * 如果是链表的话（hash大于0），就对这个链表的所有元素进行遍历 * 如果找到了key和key的hash值都一样的节点，则把它的值替换 * 如果没找到的话，则添加在链表的最后面 * 如果是树的话，会对该节点上的关联数目进行判断 * 如果在8个以上的话，则会调用treeifyBin方法，来尝试转化为树，或者扩容 */ V oldVal = null; // 再次取出要存储的位置的元素，跟前面取出来的作比较 synchronized (f) { if (tabAt(tab, i) == f) { // 取出来的元素的hash值大于0 if (fh >= 0) { binCount = 1; // 遍历这个链表 for (Node e = f;; ++binCount) { K ek; // 如果hash,key相同，则替换value if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; // 当使用putIfAbsent的时候，只有在这个key没有设置值的时候才设置 if (!onlyIfAbsent) e.val = value; break; } Node pred = e; // 如果不是同样的hash，同样的Key的时候，则判断该节点的下一个节点是否为空， 为空吧这个要加入的节点设置为当前节点的下一个节点 if ((e = e.next) == null) { pred.next = new Node(hash, key, value, null); break; } } } // 如果已经转成红黑树类型了 else if (f instanceof TreeBin) { Node p; binCount = 2; // 将元素添加到树中去 if ((p = ((TreeBin)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { // 当在同一个节点的数目达到8个的时候，则扩张数组或将给节点的数据转为tree if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 计数 addCount(1L, binCount); return null; } 扩容源码
节点个数 超过8，调用 treeifyBin 扩容。
节点数组长度小于64，扩张数组长度一倍，否则的话把链表转为树 /** * Replaces all linked nodes in bin at given index unless table is * too small, in which case resizes instead. * 当数组长度小于64的时候，扩张数组长度一倍，否则的话把链表转为树 */ private final void treeifyBin(Node[] tab, int index) { Node b; int n, sc; if (tab != null) { if ((n = tab.length) < MIN_TREEIFY_CAPACITY) //MIN_TREEIFY_CAPACITY 64 tryPresize(n << 1); // 数组扩容 else if ((b = tabAt(tab, index)) != null && b.hash >= 0) { synchronized (b) { //使用synchronized同步器，将该节点出的链表转为树 if (tabAt(tab, index) == b) { TreeNode hd = null, tl = null; //hd：树的头(head) for (Node e = b; e != null; e = e.next) { TreeNode p = new TreeNode(e.hash, e.key, e.val, null, null); if ((p.prev = tl) == null) //把Node组成的链表，转化为TreeNode的链表，头结点任然放在相同的位置 hd = p; //设置head else tl.next = p; tl = p; } setTabAt(tab, index, new TreeBin(hd));//把TreeNode的链表放入容器TreeBin中 } } } } } 读源码
读支持并发操作 /* * 相比put方法，get就很单纯了，支持并发操作， * 当key为null的时候回抛出NullPointerException的异常 * get操作通过首先计算key的hash值来确定该元素放在数组的哪个位置 * 然后遍历该位置的所有节点 * 如果不存在的话返回null */ public V get(Object key) { Node[] tab; Node e, p; int n, eh; K ek; int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } 1.7及以下版本原理
ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构， 一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素， 每个Segment守护着一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁 分段锁结构
static final class Segment extends ReentrantLock implements Serializable { /** * The number of elements in this segment＂s region. */ transient volatileint count; /** * Number of updates that alter the size of the table. This is * used during bulk-read methods to make sure they see a * consistent snapshot: If modCounts change during a traversal * of segments computing size or checking containsValue, then * we might have an inconsistent view of state so (usually) * must retry. */ transient int modCount; /** * The table is rehashed when its size exceeds this threshold. * (The value of this field is always (int)(capacity * * loadFactor).) */ transient int threshold; /** * The per-segment table. */ transient volatile HashEntry[] table; /** * The load factor for the hash table. Even though this value * is same for all segments, it is replicated to avoid needing * links to outer object. * @serial */ final float loadFactor; Segment(float lf, int threshold, HashEntry[] tab) { this.loadFactor = lf; this.threshold = threshold; this.table = tab; } } 插入源码
先加锁，再操作 public V put(K key, V value) { Segment s; if (value == null) throw new NullPointerException(); int hash = hash(key.hashCode()); int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); return s.put(key, hash, value, false); } final V put(K key, int hash, V value, boolean onlyIfAbsent) { HashEntry node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry[] tab = table; int index = (tab.length - 1) & hash; HashEntry first = entryAt(tab, index); for (HashEntry e = first;;) { if (e != null) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { e.value = value; ++modCount; } break; } e = e.next; } else { if (node != null) node.setNext(first); else node = new HashEntry(hash, key, value, first); int c = count + 1; if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { unlock(); } return oldValue; } 读取源码 /** * Returns the value to which the specified key is mapped, * or {@code null} if this map contains no mapping for the key. * * More formally, if this map contains a mapping from a key * {@code k} to a value {@code v} such that {@code key.equals(k)}, * then this method returns {@code v}; otherwise it returns * {@code null}. (There can be at most one such mapping.) * * @throws NullPointerException if the specified key is null */ public V get(Object key) { Segment s; // manually integrate access methods to reduce overhead HashEntry[] tab; int h = hash(key.hashCode()); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 原子操作 if ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { for (HashEntry e = (HashEntry) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); e != null; e = e.next) { K k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; } 删除操作
先上锁，再操作 final V remove(Object key, int hash, Object value) { if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try { HashEntry[] tab = table; int index = (tab.length - 1) & hash; HashEntry e = entryAt(tab, index); HashEntry pred = null; while (e != null) { K k; HashEntry next = e.next; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { V v = e.value; if (value == null || value == v || value.equals(v)) { if (pred == null) setEntryAt(tab, index, next); else pred.setNext(next); ++modCount; --count; oldValue = v; } break; } pred = e; e = next; } } finally { unlock(); } return oldValue; } 扩容
如何扩容。扩容的时候首先会创建一个两倍于原容量的数组，然后将原数组里的元素进行再hash后插入到新的数组里。为了高效ConcurrentHashMap不会对整个容器进行扩容，而只对某个segment进行扩容。 private void rehash(HashEntry node) { /* * Reclassify nodes in each list to new table. Because we * are using power-of-two expansion, the elements from * each bin must either stay at same index, or move with a * power of two offset. We eliminate unnecessary node * creation by catching cases where old nodes can be * reused because their next fields won＂t change. * Statistically, at the default threshold, only about * one-sixth of them need cloning when a table * doubles. The nodes they replace will be garbage * collectable as soon as they are no longer referenced by * any reader thread that may be in the midst of * concurrently traversing table. Entry accesses use plain * array indexing because they are followed by volatile * table write. */ HashEntry[] oldTable = table; int oldCapacity = oldTable.length; int newCapacity = oldCapacity << 1; threshold = (int)(newCapacity * loadFactor); HashEntry[] newTable = (HashEntry[]) new HashEntry[newCapacity]; int sizeMask = newCapacity - 1; for (int i = 0; i < oldCapacity ; i++) { HashEntry e = oldTable[i]; if (e != null) { HashEntry next = e.next; int idx = e.hash & sizeMask; if (next == null) // Single node on list newTable[idx] = e; else { // Reuse consecutive sequence at same slot HashEntry lastRun = e; int lastIdx = idx; for (HashEntry last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } newTable[lastIdx] = lastRun; // Clone remaining nodes for (HashEntry p = e; p != lastRun; p = p.next) { V v = p.value; int h = p.hash; int k = h & sizeMask; HashEntry n = newTable[k]; newTable[k] = new HashEntry(h, p.key, v, n); } } } } int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; } 结论
Q：1.8版本如何保证线程安全？ 写操作：同步处理主要是通过Synchronized和unsafe两种方式来完成的在取得某个位置的Node的时候，使用的都是unsafe的方法，来达到并发安全的目的当需要在某个位置设置节点的时候，则会通过Synchronized的同步机制来锁定该位置的节点。在数组扩容的时候，则通过处理的步长和fwd节点来达到并发安全的目的，通过设置hash值为MOVED当把某个位置的节点复制到扩张后的table的时候，也通过Synchronized的同步机制来保证线程安全 读操作：根本没有使用同步机制，也没有使用unsafe方法，所以读操作是支持并发操作的
Q：读加锁么，写加锁么？ 在1.7版本及以前版本写使用分段锁（可重入锁）保证安全读使用UNSAFE.getObjectVolatile，保证数据一致性 在1.8版本写时，取位置使用unsafe， 设置节点用Synchronized，保证数据一致性读操作是直接支持并发操作
Q： 在jdk1.8与jdk1.7的差别? 1.7 使用分段锁保证线程安全，1.8 使用Synchronized和unsafe两种方式保证线程安全 1.7扩容时，分段锁局部扩容，hash算好后，重新分配，1.8整个扩容，然后数据重新放入 1.8引入红黑树，在链表太长时，转换，提高查询效率
Q：为什么key不能为null, value不能为null
看源码put入口判断 if (key == null || value == null) throw new NullPointerException();
设计意图：二义性，防止歧义
如果判断value为null， 不确定这个Null是插入进去的，还是本身真的没值, 导致歧义，为了线程安全性，所以都不能为null
