ConcurrentHashMap源码阅读（JDK1.7+JDK1.8）

文章目录

一.JDK1.8实现关于UnSafe数据结构1.原子操作：tabAt2.原子操作：casTabAt3.原子操作：setTabAt4.get5.put6.clear7.remove 二.JDK1.7实现数据结构1.Segment2.HashEntry3.get4.put图示：1.ConcurrentHashMap.put2.Segment.put3.Segment.scanAndLockForPut 4.remove5.replace6.clear

一.JDK1.8实现

关于UnSafe

JDK中有一个类Unsafe，它提供了硬件级别的原子操作。JDK API文档也没有提供任何关于这个类的方法的解释。 从描述可以了解到Unsafe提供了硬件级别的操作，比如说获取某个属性在内存中的位置，比如说修改对象的字段值，即使它是私有的。

数据结构

和HashMap采用类似的数据结构，主要实现使用了Unsafe方法提供的原子操作和sychronized关键字实现

1.原子操作：tabAt

用来返回节点数组的指定位置的节点的原子操作

@SuppressWarnings("unchecked") static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); }

2.原子操作：casTabAt

cas原子操作，在指定位置设定值

static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); }

3.原子操作：setTabAt

原子操作，在指定位置设定值

static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); }

4.get

get操作支持并发

public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; //1.计算哈希值 int h = spread(key.hashCode()); //2.判断table不为空，长度不为0，第一个节点不为空 if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { //3.如果第一个节点的哈希值相等并且值也相等，直接返回. if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))){ return e.val; } } //4.如果哈希值小于0，从当前节点一直查找，找到返回对应值或者null else if (eh < 0){ return (p = e.find(h, key)) != null ? p.val : null; } //5.如果上述两种情况不满足，则遍历链表进行查找. while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; //这边看来，get操作是支持并发的. }

5.put

put方法这边实现

利用UnSafe的CAS方法添加第一个节点（如果为空）如果检测到哈希值为MOVED，会启动帮助扩容设值节点（第一个节点不为空）操作通过sychronized方法保证同步. 图示： public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { //1.不允许key或者value为null的情况 if (key == null || value == null) { throw new NullPointerException(); } //2.计算哈希值 int hash = spread(key.hashCode()); int binCount = 0; //3.进入循环 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //4.初始化table if (tab == null || (n = tab.length) == 0){ tab = initTable(); } //5.如果第一个节点为null，通过cas的方式去尝试设定值，成功则退出循环.这边没有加锁，失败则继续往下 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //6.如果检测出当前哈希值是MOVED，说明此时正在扩容，这边会帮助扩容 else if ((fh = f.hash) == MOVED){ tab = helpTransfer(tab, f); } //7.进入设值阶段，同步操作 else { V oldVal = null; synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; //8.链表的处理--这边直接进行赋值 for (Node<K,V> e = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node<K,V> pred = e; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } //9.红黑树结构的操作 else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } //10.判断长度是否超过，超过则转换为树结构，否则直接返回 if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //11.计数---不是很理解为什么这么做 addCount(1L, binCount); return null; }

6.clear

这边有一步，检测到MOVED，帮助扩容，这边我的理解是需要清空所有扩容的tab，不然会有内存隐患

public void clear() { long delta = 0L; // negative number of deletions int i = 0; Node<K,V>[] tab = table; //1.循环直到所有节点为空 while (tab != null && i < tab.length) { int fh; Node<K,V> f = tabAt(tab, i); //2.略过空桶 if (f == null) ++i; //3.帮助扩容？ else if ((fh = f.hash) == MOVED) { tab = helpTransfer(tab, f); i = 0; // restart } //4.同步块处理，清空当前桶所有节点 else { synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> p = (fh >= 0 ? f : (f instanceof TreeBin) ? ((TreeBin<K,V>)f).first : null); while (p != null) { --delta; p = p.next; } setTabAt(tab, i++, null); } } } } if (delta != 0L) addCount(delta, -1); }

7.remove

remove这边直接调用了replaceNode方法 这边也有一步是帮助扩容的操作

final V replaceNode(Object key, V value, Object cv) { int hash = spread(key.hashCode()); //1.进入循环 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //2.头结点为空，说明没有元素，直接退出循环，返回 if (tab == null || (n = tab.length) == 0 || (f = tabAt(tab, i = (n - 1) & hash)) == null) break; //3.如果在做扩容操作，则帮助扩容 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; boolean validated = false; //4.锁定该节点 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { validated = true; //5.循环链表查找，找到对应元素设值，否则继续往下 for (Node<K,V> e = f, pred = null;;) { K ek; //6.找到对应节点 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { V ev = e.val; //7.设值操作 if (cv == null || cv == ev || (ev != null && cv.equals(ev))) { oldVal = ev; if (value != null) e.val = value; else if (pred != null) pred.next = e.next; else setTabAt(tab, i, e.next); } break; } pred = e; //8.达到最后节点，退出 if ((e = e.next) == null) break; } } //9.红黑树的替换处理 else if (f instanceof TreeBin) { validated = true; TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> r, p; if ((r = t.root) != null && (p = r.findTreeNode(hash, key, null)) != null) { V pv = p.val; if (cv == null || cv == pv || (pv != null && cv.equals(pv))) { oldVal = pv; if (value != null) p.val = value; else if (t.removeTreeNode(p)) setTabAt(tab, i, untreeify(t.first)); } } } } } if (validated) { if (oldVal != null) { if (value == null) addCount(-1L, -1); return oldVal; } break; } } } return null; }

二.JDK1.7实现

该部分参考:ConcurrentHashMap1.7源码分析

数据结构

jdk1.7的concurrentHashMap使用一组segment对象存储对应的HashEntry数组，每个HashEntry对象上又挂着一组链表，代表真正的key-value数据对象。

1.Segment

下面可以看出segment是继承了ReentrantLock ，猜测这边的同步操作依赖于锁

static final class Segment<K,V> extends ReentrantLock implements Serializable { //table数组 transient volatile HashEntry<K,V>[] table; transient int count; transient int modCount; //临界值 transient int threshold; //负载因子 final float loadFactor; }

2.HashEntry

HashEntry节点和HashMap的Entry节点类似，不过这边对value和next节点加了volatitle关键字

static final class HashEntry<K,V> { final int hash; final K key; volatile V value; volatile HashEntry<K,V> next; }

3.get

public V get(Object key) { Segment<K,V> s; HashEntry<K,V>[] tab; //1.获取key对应hash值 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; //2.通过Unsafe中的getObjectVolatile方法进行volatile语义的读，获取到segments在偏移量为u位置的分段Segment， //并且分段Segment中对应table数组不为空 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); e != null; e = e.next) { K k; //3.直接拿到数据返回，但是可能这个数据被其他线程修改，弱一致性. if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; }

4.put

put方法直接使用了Segment的方法，在put之前尝试加锁，加锁成功则进入操作，加锁失败则循环cas尝试获取锁，当达到一定次数后，锁住等待其他线程释放锁.

图示：

这边主要涉及到三个方法如下:

1.ConcurrentHashMap.put

public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); //1.计算哈希值、找到对应的Segment. int hash = hash(key); int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment<K,V>)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE)) == null) s = ensureSegment(j); //2.直接调用Segment的put方法 return s.put(key, hash, value, false); }

2.Segment.put

final V put(K key, int hash, V value, boolean onlyIfAbsent) { //1.先尝试对segment加锁，如果直接加锁成功，那么node=null；如果加锁失败，则会调用scanAndLockForPut方法去获取锁 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { //2.table是volatile，因为锁定之后不会被修改，所以赋值给普通变量能减少volatile的消耗 HashEntry<K,V>[] tab = table; //3.计算哈希值，查找对应的HashEntry数组 int index = (tab.length - 1) & hash; HashEntry<K,V> first = entryAt(tab, index); //4.循环当前数组的所有HashEntry节点 for (HashEntry<K,V> e = first;;) { //5.当前节点不为空，进入匹配 if (e != null) { K k; //6.匹配成功，赋值和退出循环 if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { e.value = value; ++modCount; } break; } e = e.next; } //6.当前节点为空，这边的方式应该也是属于前插法 else { if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; //7.超过大小扩容 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { //8.解锁 unlock(); } return oldValue; }

3.Segment.scanAndLockForPut

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; //1.如果尝试加锁失败，那么就对segment[hash]对应的链表进行遍历找到需要put的这个entry所在的链表中的位置， //这里之所以进行一次遍历找到坑位，主要是为了通过遍历过程将遍历过的entry全部放到CPU高速缓存中， //这样在获取到锁了之后，再次进行定位的时候速度会十分快，这是在线程无法获取到锁前并等待的过程中的一种预热方式。 while (!tryLock()) { HashEntry<K,V> f; //2.获取锁失败，初始时retries=-1必然开始先进入第一个if if (retries < 0) { //3.第一次进来第一个节点为空，给他赋值，判断node为空，是因为后面判断其他线程改变链表之后， //避免node节点在之前的遍历中赋值，导致重复赋值 if (e == null) { if (node == null) { node = new HashEntry<K,V>(hash, key, value, null); } retries = 0; } //4.遍历过程发现链表中找到了我们需要的key的坑位 else if (key.equals(e.key)){ retries = 0; } //往下遍历 else{ e = e.next; } } //5.尝试获取锁次数超过设置的最大值，直接进入阻塞等待，这就是所谓的有限制的自旋获取锁， //之所以这样是因为如果持有锁的线程要过很久才释放锁，这期间如果一直无限制的自旋其实是对系统性能有消耗的， //这样无限制的自旋是不利的，所以加入最大自旋次数，超过这个次数则进入阻塞状态等待对方释放锁并获取锁。 //unlock释放锁，必定是在外面的方法结束之后去释放. else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } //6.遍历过程中，有可能其它线程改变了遍历的链表，这时就需要重新进行遍历了。 else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) { e = first = f; retries = -1; } } return node; }

4.remove

remove操作同样用到了有限次自旋CAS锁加锁，对整个的remove操作结束之后会释放锁

public V remove(Object key) { int hash = hash(key); Segment<K,V> s = segmentForHash(hash); //查找segment，如果segment不为空，则调用它的remove方法 return s == null ? null : s.remove(key, hash, null); } final V remove(Object key, int hash, Object value) { //1.没有拿到锁，就进入循环cas锁获取、达到一定次数挂起 if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try { HashEntry<K,V>[] tab = table; int index = (tab.length - 1) & hash; HashEntry<K,V> e = entryAt(tab, index); HashEntry<K,V> pred = null; //2.通过哈希值获取的HashEntry不为空 while (e != null) { K k; HashEntry<K,V> next = e.next; //3.获取到对应的key if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { V v = e.value; if (value == null || value == v || value.equals(v)) { //4.当前值为空，直接设置next值在当前位置 if (pred == null) setEntryAt(tab, index, next); //5.否则直接把指针指向e节点的下一个节点（1.e = next;2.next=e.next），也就是略过该节点（删除） else pred.setNext(next); ++modCount; --count; oldValue = v; } break; } pred = e; e = next; } } finally { //解锁 unlock(); } return oldValue; }

5.replace

replace和remove道理差不多类似

public V replace(K key, V value) { //1.找到对应hash的segment int hash = hash(key); if (value == null) throw new NullPointerException(); Segment<K,V> s = segmentForHash(hash); //2.进行replace操作或者直接返回null return s == null ? null : s.replace(key, hash, value); } final V replace(K key, int hash, V value) { //1.获取锁和有限次cas获取锁 if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try { HashEntry<K,V> e; //2.遍历对应的HashEntry节点 for (e = entryForHash(this, hash); e != null; e = e.next) { K k; //3.找到匹配的进行换值操作 if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; e.value = value; ++modCount; break; } } } finally { //4.解锁 unlock(); } return oldValue; }

6.clear

clear会循环所有segment进行清除操作，并且清除时锁住每个segment，防止其他线程进入

public void clear() { final Segment<K,V>[] segments = this.segments; //1.循环所有的segment进行移除操作 for (int j = 0; j < segments.length; ++j) { Segment<K,V> s = segmentAt(segments, j); if (s != null) s.clear(); } } final void clear() { //1.锁定当前segment，这样应该也是为了get、put操作不能直接进来 lock(); try { //2.清除所有元素 HashEntry<K,V>[] tab = table; for (int i = 0; i < tab.length ; i++) setEntryAt(tab, i, null); ++modCount; count = 0; } finally { //3.解锁 unlock(); } } }