ConcurrentHasMap
Posted saiqsai
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ConcurrentHasMap相关的知识，希望对你有一定的参考价值。
结构图：
segment的数据结构：
static final class Segment<K, V> extends ReentrantLock implements Serializable { /* * 当前Segment里实际包含HashEntry元素的数量 * 变量申明volatile ，它的写立马对后续的读可见 */ transient volatile int count; /* * 当Segment进行put、clear、remove操作时，modCount就+1 * 用来标记Segment容器是否变化 */ transient int modCount; /* * Segment容器中包含HashEntry的元素数量的阈值 * 当容器中元素数量超过threshold时，table将rehash重新散列 * 一般threshold=loadFactor * 容器容量 */ transient int threshold; /* * Segment容器中包含的HashEntry数组 * 变量申明volatile ，它的写立马对后续的读可见 */ transient volatile HashEntry<K, V>[] table; /* * 负载因子，通常等于0.75,最优 * 负载因子较小时，降低了冲突的可能性，链表长度减少，加快查询速度，但是所占空间加大，频繁rehash性能会降低; * 负载因子较大时，元素分布比较紧凑，容易导致hash冲突，链表加长，访问更新速度较慢，但是空间利用率高。 final float loadFactor; }
实际存放元素的HashEntry数组：
static final class HashEntry<K, V> { final K key; volatile V value; /* * key的hash值, 采用Wang/Jenkins算法, 生成的hash值是32位的 */ final int hash; /* * 当key的hash冲突时(hash值相同)，通过next组成链表 */ volatile HashEntry<K, V> next; }
初始化：
public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS; /* * 求出用于定位参与散列运算的位数sshift * 求出初始化Segment的长度ssize ，即并发度 */ int sshift = 0; int ssize = 1; while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } /* * 求出参与散列运算的的掩码=散列槽的长度-1 * 初始化ConcurrentHashMap中包含的Segment数组,大小为ssize */ this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; /* * 计算每个Segment中容纳HashEntry数组的大小 * 规则：将Map设定的初始容量initialCapacity除以Segment分段锁的数量， * 求出每个Segment应该容纳多少HashEntry */ int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; /* * 创建Segment数组ss以及Segment元素s0, 并把s0作为ss的第一个元素 */ Segment<K, V> s0 = new Segment<K, V>(loadFactor, (int) (cap * loadFactor), (HashEntry<K, V>[]) new HashEntry[cap]); Segment<K, V>[] ss = (Segment<K, V>[]) new Segment[ssize]; UNSAFE.putOrderedObject(ss, SBASE, s0); // 直接操作内存 this.segments = ss; }
CurrentHashMap的初始化一共有三个参数：
initialCapacity：决定CurrentHashMap的初始容量
loadFactor：表示负载参数，决定每个Segment中元素的阈值threshold，Segment中元素的数量超过threshold时就会对该Segment进行扩容
concurrentLevel：决定Segment的数量，Segment的数量一经指定，不可改变，后续如果需要扩容，ConcurrentHashMap不会增加Segment的数量，而只会增加Segment中链表数组的容量大小，这样的好处是扩容过程不需要对整个ConcurrentHashMap做rehash，而只需要对Segment里面的元素做一次rehash就可以了。
整个ConcurrentHashMap的初始化方法还是非常简单的，先是根据concurrentLevel来算出Segment数组的大小ssize，这里ssize是大于concurrentLevel的最小的2的指数，这样的好处是方便采用位操作来进行hash，加快hash的速度。接下来就是根据intialCapacity确定每个Segment的容量的大小cap，cap也是2的指数，同样使为了加快hash的过程。
这边需要特别注意一下两个变量，分别是segmentShift和segmentMask，假设构造函数确定了Segment数组的大小是2的n次方，那么segmentShift就等于32减去n，而segmentMask就等于2的n次方减一。
UNSAFE.putOrderedObject(ss, SBASE, s0)
SBASE = UNSAFE.arrayBaseOffset(Segment[].class);
UNSAFE是java提供的直接操作内存的类。SBASE是对象头的大小。putOrderedObject方法就是吧s0放在相对于ss的偏移量为SBASE大小的地方。其实就是等价于放在ss数组的第一个元素的位置。
put操作：
public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); int hash = hash(key); //计算key的hash值，采用Wang/Jenkins算法，生成的hash值是32位的 /* * 取出hash值的低4位，定位到在Segment数组中的位置 */ int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); return s.put(key, hash, value, false); }
假设Segment数组的大小是2的4次方，即16，那么segmentShift就等于28，而segmentMask就等于15。
int j = (hash >>> segmentShift) & segmentMask;
那么j = hash值的最后四位。
private Segment<K,V> ensureSegment(int k) { final Segment<K,V>[] ss = this.segments; /* * 定位元素在内存的位置 */ long u = (k << SSHIFT) + SBASE; Segment<K,V> seg; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { /* * 把ss[0]作为原型 */ Segment<K,V> proto = ss[0]; // use segment 0 as prototype int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); /* * 如果元素已经存在（别的线程添加），则返回该元素 * 负责采用CAS替换该元素 */ while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }
final V put(K key, int hash, V value, boolean onlyIfAbsent) { /* * 尝试获取锁，如果没有成功，调用scanAndLockForPut方法，在该方法里面再进行尝试 */ HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry<K,V>[] tab = table; int index = (tab.length - 1) & hash; /* * 找到在table中的插入点 */ HashEntry<K,V> first = entryAt(tab, index); for (HashEntry<K,V> e = first;;) { if (e != null) { K k; /* * 如果存在hash值一样的，则用新的value值替换掉老的value值 */ 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<K,V>(hash, key, value, first); int c = count + 1; /* * 如果容量超过该segment的阈值，则rehash */ if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { unlock(); } return oldValue; }
如put源码所示，当put操作尝试加锁没成功时，它不是直接进入等待状态，而是调用了scanAndLockForPut()操作，该操作持续查找key对应的节点链中是否已存在该hash值对应的节点，如果没有找到已存在的节点，则预创建一个新节点，并且尝试加锁，直到尝试次数大于MAX_SCAN_RETRIES，才真正进入等待状态，即所谓的自旋等待。对最大尝试次数，目前的实现单核次数为1，多核为64。
在这段逻辑中，它先获取key对应的节点链的头，然后持续遍历该链，如果节点链中不存在要插入的节点，则预创建一个节点，否则retries值自增，直到操作最大尝试次数而进入等待状态。这里需要注意最后一个else if中的逻辑：当在自旋过程中发现节点链的链头发生了变化，则更新节点链的链头，并重置retries值为－1，重新为尝试获取锁而自旋遍历。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { /* * 获取该hash值对应的HashEntry数组中的元素 */ HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; while (!tryLock()) { HashEntry<K,V> f; if (retries < 0) { /* * 没有相应的节点，则创建一个 */ if (e == null) { if (node == null) node = new HashEntry<K,V>(hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else e = e.next; } /* * 自旋次数超过MAX_SCAN_RETRIES，则执行lock()，等待锁 */ else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } /* * 如果节点被别的线程修改了，则重新进行自旋 */ else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) { e = first = f; retries = -1; } } return node; }
static final <K,V> HashEntry<K,V> entryForHash(Segment<K,V> seg, int h) { HashEntry<K,V>[] tab; return (seg == null || (tab = seg.table) == null) ? null : (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); }
rehash的逻辑比较简单，它创建一个原来两倍容量的数组，然后遍历原来数组以及数组项中的每条链，对每个节点重新计算它的数组索引，然后创建一个新的节点插入到新数组中，这里需要重新创建一个新节点而不是修改原有节点的next指针是为了在做rehash时可以保证其他线程的get遍历操作可以正常在原有的链上正常工作。为了减少重新创建新节点的开销，这里做了两点优化：1，对只有一个节点的链，直接将该节点赋值给新数组对应项即可（之所以能这么做是因为Segment中数组的长度也永远是2的倍数，而将数组长度扩大成原来的2倍，那么新节点在新数组中的位置只能是相同的索引号或者原来索引号加原来数组的长度，因而可以保证每条链在rehash是不会相互干扰）；2，对有多个节点的链，先遍历该链找到第一个后面所有节点的索引值一样的节点p，然后只重新创建节点p以前的节点即可，此时新节点链和旧节点链同时存在，在p节点相遇，这样即使有其他线程在当前链做遍历也能正常工作
private void rehash(HashEntry<K,V> node) { HashEntry<K,V>[] oldTable = table; int oldCapacity = oldTable.length; //把table容量变为原来的两倍 int newCapacity = oldCapacity << 1; threshold = (int)(newCapacity * loadFactor); HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; int sizeMask = newCapacity - 1; for (int i = 0; i < oldCapacity ; i++) { HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; int idx = e.hash & sizeMask; /* * table上的该节点只有一个元素，直接用新的节点指向他 * 否则需要重新计算 */ if (next == null) newTable[idx] = e; else { HashEntry<K,V> lastRun = e; int lastIdx = idx; for (HashEntry<K,V> last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } newTable[lastIdx] = lastRun; for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { V v = p.value; int h = p.hash; int k = h & sizeMask; HashEntry<K,V> n = newTable[k]; newTable[k] = new HashEntry<K,V>(h, p.key, v, n); } } } } int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }
rehash示意图：
以上是关于ConcurrentHasMap的主要内容，如果未能解决你的问题，请参考以下文章