ConcurrentHashMap1.7 源码浅析

分析过HashMap的1.7的版本的结构，但是HashMap是线程不安全的，多线程触发扩容还会发生死循环问题，那么ConcurrentHashMap 就是解决这个问题的，这是一个线程安全的Map，那么对应的内部实现是怎么样的，简单分析下，和HashMap相同的位置就不多做重复分析了

构造方法

这是个最基础的构造方法，需要的参数有容量，扩容因子，这是和HashMap相同的地方，但是多了一个并发水平选项，这里默认值是16，也就是并发粒度的控制，最多可以16个线程同时加锁对Map处理，我们看下具体怎么做的。

    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;// Find power-of-two sizes best matching argumentsint sshift = 0;int ssize = 1;// 小于并发等级就左移动一位，*2倍在得到一个大于等于这个数的2的n次幂while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// sshift 记录的是对应的多少位// ssize记录的就是处理过的并发水平this.segmentShift = 32 - sshift;this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;// 这里在计算分段之后，每段的长度int c = initialCapacity / ssize;// 这里向上取下，因为上面int处理的直接向下取整了if (c * ssize < initialCapacity)++c;// 最小的分段长度为2int cap = MIN_SEGMENT_TABLE_CAPACITY;// 如果小于需要的值，就左移动一位，扩大二倍，知道大于等于需要的while (cap < c)cap <<= 1;// create segments and segments[0]// 对应的段，段的数组长度就是cap，长度的HashEntrySegment<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];// cas 赋值UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]// 赋值给Map的成员变量this.segments = ss;}

从构造方法来看，就是对HashMap进行分段了，控制实际的容量大于等于需要的容量，并发粒度也是大于等于需要的粒度，这样可以对每一个段进行加锁，保证并发安全，又能保证一定的并发粒度，后面看下是怎么进行插入和获取的，是怎么进行加锁的，怎么保证数据安全的？

Segment 的结构

看下主要的结构，继承了ReentrantLock，同样维护了Map的扩容因子，扩容阈值，元素数量，Entry数组这些

    static final class Segment<K,V> extends ReentrantLock implements Serializable {// 在预扫描之前尝试锁定的最大次数static final int MAX_SCAN_RETRIES =Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;// 对应的Entry数组，volatile修饰，修改数据对其他线程可见，及时刷新回主存// transient修饰，序列化忽略这个transient volatile HashEntry<K,V>[] table;// 分段内的元素的数量transient int count;// 修改相关的计数transient int modCount;// 扩容阈值transient int threshold;// 扩容因子final float loadFactor;}

put 方法分析

老规矩先从主要方法put开始看，put里面一般能看到存储结构，查找顺序等关键信息

public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();int hash = hash(key);// 计算位于哪个segmentint j = (hash >>> segmentShift) & segmentMask;// segment未初始化化的时候，进行初始化if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck(segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegments = ensureSegment(j);// 元素插入return s.put(key, hash, value, false);
}final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试获取锁，获取到了，第一步获取不到就是后面还是在获取，// 存在尝试次数，尝试次数完了之后死等，如果直接得到锁，返回的node就是null// 如果开始没得到锁，并且开始数组没数据的时候，会得到一个nodeHashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);// 上一步走完就是获取到锁了V oldValue;try {HashEntry<K,V>[] tab = table;int index = (tab.length - 1) & hash;// 获取对应下标的值，valitale保证及时可见HashEntry<K,V> first = entryAt(tab, index);for (HashEntry<K,V> e = first;;) {// 数组上有值if (e != null) {K k;// 遇见相同keyif ((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)// 头插法，上面获取到了nodenode.setNext(first);elsenode = new HashEntry<K,V>(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;
}

put方法就是先检查对应的segment是不是初始化了，未初始化的先进性初始化，初始化的时候按照第一个进行复制，然后cas赋值到对应位置，然后执行实际的插入逻辑，插入的时候先进行获取lock，因为Segnment继承的ReentrantLock，直接使用tryLock(),和lock()进行获取的非公平锁，获取之后对对应的值进行添加

ensureSegment 初始化segment分析

private Segment<K,V> ensureSegment(int k) {final Segment<K,V>[] ss = this.segments;long u = (k << SSHIFT) + SBASE; // raw offsetSegment<K,V> seg;if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {// 用构造方法时创建的第一个segment作为原型，复制一个segment出来Segment<K,V> proto = ss[0]; // use segment 0 as prototypeint 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) { // recheckSegment<K,V> s = new Segment<K,V>(lf, threshold, tab);// cas赋值segmentwhile ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}}return seg;
}

scanAndLockForPut� 这里也是在获取锁，能抽空的话就会返回一个HashEntry

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {// 寻找hash对应的entryHashEntry<K,V> first = entryForHash(this, hash);HashEntry<K,V> e = first;HashEntry<K,V> node = null;int retries = -1; // negative while locating node// 获取不到锁，就一直获取while (!tryLock()) {HashEntry<K,V> f; // to recheck first below// 尝试次数if (retries < 0) {// 这里的e是遍历的当前节点if (e == null) {if (node == null) // speculatively create node// 这里的创建Entry应该就是简单的没别的事情做？充分利用下node = new HashEntry<K,V>(hash, key, value, null);retries = 0;}// 直接遇到相同的了else if (key.equals(e.key))retries = 0;// 遍历下一个elsee = e.next;}// 超过获取锁的最大尝试次数了else if (++retries > MAX_SCAN_RETRIES) {lock();break;}// 这里在判断尝试次数是不是为0，为0的上面两种，一种是对应数组位置无值，一种是有相同// key，可以直接替换的，//然后 后面的&& 判断了下是不是自己，如果不是自己插入的，更换下头节点，可能是别的线程// 插入的else if ((retries & 1) == 0 &&(f = entryForHash(this, hash)) != first) {e = first = f; // re-traverse if entry changedretries = -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(node) 扩容方法分析

private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;// 原来的容量int oldCapacity = oldTable.length;// 新的容量，扩大为2倍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;if (next == null)   //  Single node on list// 只有一个newTable[idx] = e;else { // Reuse consecutive sequence at same slot// 记录的当前遍历的头节点eHashEntry<K,V> lastRun = e;int lastIdx = idx;// 这里在遍历链表，last就是当前遍历到的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;// Clone remaining nodes// 然后重新遍历了一遍，遇到之前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);}}}}// 扩容之后把node插入进去int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;table = newTable;
}

可以看到resize基本等同HashMap，不过在resize里面把node进行插入的

get 方法

可以看到get方法比较简单，不需要加锁，通过volitale修饰的key，然后UNSAFE.getObjectVolatile 拿到对应的值

    public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;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;if ((k = e.key) == key || (e.hash == h && key.equals(k)))return e.value;}}return null;}

总结

整体总结下，ConcurrentHashMap 通过引入Segment概念来对HashMap进行分段，等于有了很多个小的HashMap，然后又继承了ReentrantLock，通过lock来保证put的数据安全性，并发粒度通过设置segment的数组长度来控制，默认16，可以自定义，不过也是2的n次幂，类似于原来hashmap的数组大小取值算法，对于segment里面，数组的最小长度为2，这个数组长度和segment的长度决定了容量大小，这里会大于等于设置的值，segment数组是不可变的，也就是map构造完成的时候，并发粒度就确定了，segment的长度大小不可变，扩容是在扩容的内部的HashMap，也就是HashEntry数组，整体就是hash计算了两次，第一次确认在那个segment，然后再计算落到那个段里面的哪个位置。