jdk1.7的ConcurrentHashMap
jdk1.7采用的是Segment+HashEntry数组+链表来实现的
ConcurrentHashMap初始化时有16个Segment，每个Segment中又包含了HashEntry数组，每个HashEntry又是一个链表，可以看出，Segment的结构和HashMap类似，你可以这样理解：ConcurrentHashMap由多个HashMap组成。
线程安全的实现原理
ConcurrentHashMap中的Segment在实现上继承了ReentrantLock，这样就实现了分段锁的功能。下面我们通过put()方法来看看它是如何保证线程安全的。
public V put(K key, V value) { Segment<K,V> s; // value不能为null if (value == null) throw new NullPointerException(); // 计算key的hash值 int hash = hash(key); // 定位到哪个Segment int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment<K,V>)UNSAFE.getObject (segments, (j << SSHIFT) + SBASE)) == null) // 如果此Segment为null，则初始化Segment s = ensureSegment(j); // 往Segment中put值 return s.put(key, hash, value, false);}final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 尝试获取锁，如果获取不到，则进入scanAndLockForPut方法 // scanAndLockForPut中会再次tryLock，如果还是获取不到，则lock()挂起线程 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { // table通过volatile修饰，保证拿到的都是最新的table HashEntry<K,V>[] tab = table; // 得到数组下标，&取模作用，详细介绍可以看我写的jdk1.8HashMap文章 int index = (tab.length - 1) & hash; // 得到此下标下的HashEntry HashEntry<K,V> first = entryAt(tab, index); // 遍历链表 for (HashEntry<K,V> e = first;;) { if (e != null) { K k; // 如果key相同，则覆盖原值 if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { e.value = value; // 修改次数+1，用于限制遍历中删除 ++modCount; } break; } // 继续遍历下个值 e = e.next; } else { if (node != null) // 头插法（头插法的问题看我写的jdk1.7HashMap文章） node.setNext(first); else // new新的HashEntry，里面也是采用头插法 node = 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;}通过上面源码可以看出jdk1.7的ConcurrentHashMap通过Segment的ReentrantLock锁来实现线程安全。要注意的是上锁操作是this.lock()，也就是每个Segment的锁是分开的，其中一个上锁并不影响其他Segment操作，说明可以同时16个线程同时进来对不同Segment操作，这样可以有效的提高并发度。
这种Segment+ReentrantLock看起来可以很好的解决并发问题了，那为什么jdk1.8又不继续使用此方法了呢？下面我们看看jdk1.8的ConcurrentHashMap
jdk1.8的ConcurrentHashMap
jdk1.8采用的是Node数组+链表/红黑树实现
jdk1.8的ConcurrentHashMap去除了外层的Segment，直接使用数组加链表或红黑树，数据结构和1.8的HashMap一样，不同之处在于使用了CAS+synchronized保证了线程安全。
线程安全的实现原理
我们还是通过put()方法看看它是如何保证线程安全的
public V put(K key, V value) { return putVal(key, value, false);}final V putVal(K key, V value, boolean onlyIfAbsent) { // 同样value不能为null if (key == null || value == null) throw new NullPointerException(); // 新的hash算法，和1.7的不同 int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0) // 如果table为空，初始化 tab = initTable(); // f=此下标的头节点，i=下标 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果此下标节点为空，new新节点放在此节点下 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } // fh=头节点的hash值 else if ((fh = f.hash) == MOVED) // 协助扩容 tab = helpTransfer(tab, f); else { V oldVal = null; // 重点来了，synchronized锁当前node节点 synchronized (f) { if (tabAt(tab, i) == f) { // 头节点hash值大于0，说明是链表 if (fh >= 0) { // 链表长度 binCount = 1; // 遍历链表 for (Node<K,V> e = f;; ++binCount) { K ek; // key相同，覆盖原值 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; // 到链表末端，new新node挂在末端 if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key,value, null); break; } } } // 红黑树逻辑 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; } } } } if (binCount != 0) { // 如果链表长度超过阈值，转红黑树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null;}通过上面源码可以看出，1.8的ConcurrentHashMap采用的是synchronized来实现线程安全。synchronized (f)可以看出锁的是每一个Node对象，这样可以将锁细化，除非有多个线程同时操作同一个Node，否则不会出现争抢锁，这样锁也不会升级，就算出现争抢锁，也可以通过自旋来获取锁，达到一定次数才会升级为重量级锁，这种情况是很少的。那为什么1.7的ReentrantLock不细化到Node上呢？
如果1.7的ReentrantLock细化到Node上，它只会在线程没有抢到锁的时候，新建节点再尝试一次，如果还没抢到，不会自旋，而是直接挂起，这样就增加了线程上下文开销的代价。当然，ReentrantLock的tryLock()也是可以自旋的，但是确定不了tryLock的时间。
所以，在锁细化到如此程度的情况下，使用synchronized是最好的选择。
<center>扫一扫，关注我</center>