简单总结ConcurrentHashMap
1、HashTablehtml
hashTable是一个线程安全的容器，是线程安全版本的HashMap。但它的底层是和HashMap同样的，只是在方法上都加上了synchronized关键字。node
这样子有什么后果呢：算法
- 效率及低，意味着每一个线程在执行HashTable的方法的时候，或者说操纵HashTable的时候，都要锁住整个对象。也就是让并行并发的访问，变成了串行。
- 复合操做会有线程安全问题。由于它是每一个方法都加锁了，这意味着在执行单个方法像put，contains方法的时候，是能够保证原子性的，但若是是执行一个复合操做的时候，就不保证了。
if(!table.contains("key")) { map.put("key", object); }
相似于这样的方法，当线程1在执行if里面的判断的时候，线程1会得到table实例的所，其余线程没法访问table的其余同步方法。但当线程1判断完if后，锁会放掉，这个时候若是线程2进来，得到table实例的锁，而后put了一个”key“进来，而后再放锁；那么线程1再执行put方法就不对了。（它原本是觉得没有这个key再put的）数组
2、concurrentHashMap1.7安全
并发思路多线程
concurrenthashMap是采用一个叫作分段所的机制。并发
它能够看做是一个二重hashMap，首先concurrentHashMap是一个segment数组，每一个segment都是一个继承了ReentrantLock的类，这样就能够方便地在各个segment里面加锁因此每次须要加锁的操做锁住的是一个 segment，这样只要保证每一个 Segment 是线程安全的，也就实现了全局的线程安全。ssh
哦哦还要注意，这个最外面的Segment[]数组，是不能够扩容的！ide
而后进到Segment内部，会发现，每一个Segment能够看做一个HashMap。也就是在一个Segment里面，有个HashEntry[]数组，而后这个数组是一个个桶，桶里面是单向链表。函数
（图片来自：http://www.importnew.com/28263.html）
构造函数
而后咱们经过构造函数进入，顺便了解ConcurrentHashMap中重要的field吧。
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 arguments int sshift = 0; int ssize = 1; // 计算并行级别 ssize，由于要保持并行级别是 2 的 n 次方 while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } // 咱们这里先不要那么烧脑，用默认值，concurrencyLevel 为 16，sshift 为 4 // 那么计算出 segmentShift 为 28，segmentMask 为 15，后面会用到这两个值 this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // initialCapacity 是设置整个 map 初始的大小， // 这里根据 initialCapacity 计算 Segment 数组中每一个位置能够分到的大小 // 如 initialCapacity 为 64，那么每一个 Segment 或称之为"槽"能够分到 4 个 int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，由于这样的话，对于具体的槽上， // 插入一个元素不至于扩容，插入第二个的时候才会扩容 int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; // 建立 Segment 数组， // 并建立数组的第一个元素 segment[0] 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]; // 往数组写入 segment[0] UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] this.segments = ss; }
initialCapacity和之前同样，指的是这个ConcurrenthashMap的初始容量，或者说是理解成初始桶的数量。但咱们这个hashmap是有两重表的嘛，因此在实际操做的时候会把这个值分配给各个Segment，也就至关于间接指定了每一个Segment中应该有几个桶。
loadFactor和通常的hashTable同样，负载因子，size/capacity。但上面说了Segment数组是不能够扩容的，因此这个也是给Segment里面的数组用的。
concurrencyLevel：concurrencyLevel：并行级别、并发数、Segment 数，怎么翻译不重要，理解它。默认是 16，也就是说 ConcurrentHashMap 有 16 个 Segments，因此理论上，这个时候，最多能够同时支持 16 个线程并发写，只要它们的操做分别分布在不一样的 Segment 上。这个值能够在初始化的时候设置为其余值，可是一旦初始化之后，它是不能够扩容的。
segmentShift：
这个值=32 - shift，shift是你>=你传进来的concurrentLevel的一个2次幂数的左移位数。而二次幂的数字，都是10000这样的嘛，因此shift就是10000中0的个数。
因此field segmentShift我以为能够理解成000000100000（32位数字），而后是前面的0加上1的位数就是segmentShift吧。
SegmetnMask：
掩码嘛，就二次幂处理后的concurrentLevel的长度 - 1，获得的就相似0111111这样咯，因此等等用来作与操做用的。
而后最后那个Unsafe的putOrderObject一个不安全的直接操纵内存的方法，应该是由于这样会快点吧。这个order应该是防止指令重排序的意思。
要了解Unsafe能够看这篇文章：https://www.cnblogs.com/throwable/p/9139947.html
若是咱们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的，那么初始化完成后：
Segment 数组长度为 16，不能够扩容
Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是之后插入第一个元素不会触发扩容，插入第二个会进行第一次扩容
这里初始化了 segment[0]，其余位置仍是 null，至于为何要初始化 segment[0]，后面的代码会介绍
当前 segmentShift 的值为 32 – 4 = 28，segmentMask 为 16 – 1 = 15，姑且把它们简单翻译为移位数和掩码，这两个值立刻就会用到
put方法
而后来看重要的put方法。
先看put的主流程：
public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); // 1. 计算 key 的 hash 值 int hash = hash(key); // 2. 根据 hash 值找到 Segment 数组中的位置 j // hash 是 32 位，无符号右移 segmentShift(28) 位，剩下低 4 位， // 而后和 segmentMask(15) 作一次与操做，也就是说 j 是 hash 值的最后 4 位，也就是槽的数组下标 int j = (hash >>> segmentShift) & segmentMask; // 刚刚说了，初始化的时候初始化了 segment[0]，可是其余位置仍是 null， // ensureSegment(j) 对 segment[j] 进行初始化 if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); // 3. 插入新值到 槽 s 中 return s.put(key, hash, value, false); }
这里的主流程是在第一层表格的操做。就根据key的hash值，找到Segment[]数组的桶序号，而后先初始化这个segment[j]（构造器中只初始化了segment[0]），而后进入这个segmetn[j]，交给这个segment[j]（局部HashMap）继续执行put操做。
求j的时候，hash值移了segmentShift后，恰好只剩后面四位（默认状况的话），恰好等于segmentMask15（4位）的位数，而后再相与就获得一个序号咯。
而后就经过s.put(key, hash, value, false)；进入Segment内部的那个局部Hashmap的put方法。、
先看看这个初始化segment[j]的方法。
ensureSegment(j)：
private Segment<K,V> ensureSegment(int k) { final Segment<K,V>[] ss = this.segments; long u = (k << SSHIFT) + SBASE; // raw offset Segment<K,V> seg; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 这里看到为何以前要初始化 segment[0] 了， // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] // 为何要用“当前”，由于 segment[0] 可能早就扩容过了 Segment<K,V> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 内部的数组 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 再次检查一遍该槽是否被其余线程初始化了。 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 使用 while 循环，内部用 CAS，当前线程成功设值或其余线程成功设值后，退出 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }
这里须要考虑并发，由于极可能会有多个线程同时进来初始化同一个槽 segment[k]，不过只要有一个成功了就能够。
这里就用构造其中已经初始化好了的segment[0]（也可能已经有元素了）的数据来构造segment[j]咯，而后再用自旋的CAS操做来更新segment数组中的j桶，更新成功或者是有别的线程更新成功都会跳出循环。
再来看segment里面的局部HashMap的put方法。
Segment里面的hashMap的put方法：
final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 在往该 segment 写入前，须要先获取该 segment 的独占锁 // 先看主流程，后面还会具体介绍这部份内容 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { // 这个是 segment 内部的数组 HashEntry<K,V>[] tab = table; // 再利用 hash 值，求应该放置的数组下标 int index = (tab.length - 1) & hash; // first 是数组该位置处的链表的表头 HashEntry<K,V> first = entryAt(tab, index); // 下面这串 for 循环虽然很长，不过也很好理解，想一想该位置没有任何元素和已经存在一个链表这两种状况 for (HashEntry<K,V> 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 { // node 究竟是不是 null，这个要看获取锁的过程，不过和这里都没有关系。 // 若是不为 null，那就直接将它设置为链表表头；若是是null，初始化并设置为链表表头。 if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; // 若是超过了该 segment 的阈值，这个 segment 须要扩容 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); // 扩容后面也会具体分析 else // 没有达到阈值，将 node 放到数组 tab 的 index 位置， // 其实就是将新的节点设置成原链表的表头 setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { // 解锁 unlock(); } return oldValue; }
咱们能够看到，代码一开始就先去得到所在Segment的锁：
HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
首先，这个tryLock()是个快速得到锁的方法，得到的话就返回ture，那么node就赋值为null。
若是没得到锁的话，说明存在竞争，那么就进入scanAndLockForPut方法。这个方法的话其实也就是不断去尝试得到这个Segment的锁，里面还有可能顺便初始化下这个node元素。（就可能顺便构造下你要插入的那个键值对的node）
这个scanAndLockForPut的方法等等下面才去分析，这两行代码的结果就是——得到了segment的锁，而后可能初始化了node也可能没有。（看下面代码会知道node有没有初始化没所谓的）
而后就正常的put操做了。这里是带锁了的，因此不用怕其余的写操做会影响。
能够看到，node为空它就new一个相关Entry，不为空就直接头插入，因此是否是null不影响代码逻辑。
关于这个setEntryAt方法，能够简单看看它的代码：
static final <K,V> void setEntryAt(HashEntry<K,V>[] tab, int i, HashEntry<K,V> e) { UNSAFE.putOrderedObject(tab, ((long)i << TSHIFT) + TBASE, e); }
也是用了个防止重排序的方法，再加上原本Segment里面的table还有Entry里面重要的相关指针都是volatile的，因此可让读操做也安全。
而后就来看这个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; // negative while locating node // 循环获取锁 while (!tryLock()) { HashEntry<K,V> f; // to recheck first below if (retries < 0) { if (e == null) { if (node == null) // speculatively create node // 进到这里说明数组该位置的链表是空的，没有任何元素 // 固然，进到这里的另外一个缘由是 tryLock() 失败，因此该槽存在并发，不必定是该位置 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（单核1多核64），那么不抢了，进入到阻塞队列等待锁 // lock() 是阻塞方法，直到获取锁后返回 else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } else if ((retries & 1) == 0 && // 这个时候是有大问题了，那就是有新的元素进到了链表，成为了新的表头 // 因此这边的策略是，至关于从新走一遍这个 scanAndLockForPut 方法 (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node;
代码全在一个不断尝试拿锁的while循环里进行，代码逻辑大概是用这个retries来控制流程。
当这个retries<0的时候，也就是初始状况，这里作的是遍历这个桶的链表，看看有没这个要put的key的entry，若是有的话就停下来，retries置为0，没有的话顺便new一个node，而后retries置为0。
若是retries尝试的次数太大了，就会lock()，这个方法是堵塞锁，相似synchronized（解锁在put方法中），直到拿到锁才break。
最后一个状况大概是发生了冲突了，就从新走一次这个方法。
这个方法有两个出口，一个是 tryLock() 成功了，循环终止，另外一个就是重试次数超过了 MAX_SCAN_RETRIES，进到 lock() 方法，此方法会阻塞等待，直到成功拿到独占锁。
这个方法就是看似复杂，可是其实就是作了一件事，那就是获取该 segment 的独占锁，若是须要的话顺便实例化了一下 node。
扩容方法rehash
重复一下，segment 数组不能扩容，扩容是 segment 数组某个位置内部的数组 HashEntry\[] 进行扩容，扩容后，容量为原来的 2 倍。
首先，咱们要回顾一下触发扩容的地方，put 的时候，若是判断该值的插入会致使该 segment 的元素个数超过阈值，那么先进行扩容，再插值，读者这个时候能够回去 put 方法看一眼。
该方法不须要考虑并发，由于到这里的时候，是持有该 segment 的独占锁的。
看代码：
// 方法参数上的 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]; // 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’ int sizeMask = newCapacity - 1; // 遍历原数组，老套路，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置 for (int i = 0; i < oldCapacity ; i++) { // e 是链表的第一个元素 HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; // 计算应该放置在新数组中的位置， // 假设原数组长度为 16，e 在 oldTable[3] 处，那么 idx 只多是 3 或者是 3 + 16 = 19 int idx = e.hash & sizeMask; if (next == null) // 该位置处只有一个元素，那比较好办 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // e 是链表表头 HashEntry<K,V> lastRun = e; // idx 是当前链表的头结点 e 的新位置 int lastIdx = idx; // 下面这个 for 循环会找到一个 lastRun 节点，这个节点以后的全部元素是将要放到一块儿的 for (HashEntry<K,V> last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // 将 lastRun 及其以后的全部节点组成的这个链表放到 lastIdx 这个位置 newTable[lastIdx] = lastRun; // 下面的操做是处理 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 node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }
找桶用的是直接从新和新的capacity - 1的值与的方法。
转移操做用了个改良的算法吧，没怎么认真看，直接上参考文章的分析：
这里的扩容比以前的 HashMap 要复杂一些，代码难懂一点。上面有两个挨着的 for 循环，第一个 for 有什么用呢？
仔细一看发现，若是没有第一个 for 循环，也是能够工做的，可是，这个 for 循环下来，若是 lastRun 的后面还有比较多的节点，那么此次就是值得的。由于咱们只须要克隆 lastRun 前面的节点，后面的一串节点跟着 lastRun 走就是了，不须要作任何操做。
我以为 Doug Lea 的这个想法也是挺有意思的，不过比较坏的状况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素，那么此次遍历就有点浪费了。不过 Doug Lea 也说了，根据统计，若是使用默认的阈值，大约只有 1/6 的节点须要克隆。
get过程
相对于 put 来讲，get 真的不要太简单。
计算 hash 值，找到 segment 数组中的具体位置，或咱们前面用的“槽”
槽中也是一个数组，根据 hash 找到数组中具体的位置
到这里是链表了，顺着链表进行查找便可
public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; // 1. hash 值 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 2. 根据 hash 找到对应的 segment if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { // 3. 找到segment 内部数组相应位置的链表，遍历 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; }
这里用了个getObjectVolatile来保证读的可见性。
并发问题分析
如今咱们已经说完了 put 过程和 get 过程，咱们能够看到 get 过程当中是没有加锁的，那天然咱们就须要去考虑并发问题。
添加节点的操做 put 和删除节点的操做 remove 都是要加 segment 上的独占锁的，因此它们之间天然不会有问题，咱们须要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操做。
1. put 操做的线程安全性。
- 初始化槽，这个咱们以前就说过了，使用了 CAS 来初始化 Segment 中的数组。
- 添加节点到链表的操做是插入到表头的，因此，若是这个时候 get 操做在链表遍历的过程已经到了中间，是不会影响的。固然，另外一个并发问题就是 get 操做在 put 以后，须要保证刚刚插入表头的节点被读取，这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject还有相关的volatile的field。
- 扩容。扩容是新建立了数组，而后进行迁移数据，最后面将 newTable 设置给属性 table。因此，若是 get 操做此时也在进行，那么也不要紧，若是 get 先行，那么就是在旧的 table 上作查询操做；而 put 先行，那么 put 操做的可见性保证就是 table 使用了 volatile 关键字。
2. remove 操做的线程安全性。
remove 操做咱们没有分析源码，因此这里说的读者感兴趣的话仍是须要到源码中去求实一下的。
get 操做须要遍历链表，可是 remove 操做会”破坏”链表。
若是 remove 破坏的节点 get 操做已通过去了，那么这里不存在任何问题。
若是 remove 先破坏了一个节点，分两种状况考虑。 一、若是此节点是头结点，那么须要将头结点的 next 设置为数组该位置的元素，table 虽然使用了 volatile 修饰，可是 volatile 并不能提供数组内部操做的可见性保证，因此源码中使用了 UNSAFE 来操做数组，请看方法 setEntryAt。二、若是要删除的节点不是头结点，它会将要删除节点的后继节点接到前驱节点中，这里的并发保证就是 next 属性是 volatile 的。
总之，首先segment分段所保证了单一的写、删是无并发危险的。
而后读和这些更改性的操做呢，首先是经过table的volatile，而后涉及table中桶元素的替换（要访问桶的第一个元素的时候），就利用setEntryAt方法中的unsafe的保证有序性的方法。
涉及读非头节点的，除了以上措施还有就是Entry中的next属性也是volatile的。
后期补的：
由于这个table虽说是volatile的，但它里面的元素是不能有volatile的效果的：Java数组在元素层面的元数据设计上的缺失，没法表达元素是final、volatile等语义，数组元素就跟没有标volatile的成员字段同样，没法保证线程之间可见性。
因此，咱们看到，这里关于在数组里面取元素的操做，都用的是getObjectVolatile之类的，经过这个来弥补数组的这个不足；
而后我看到好多关于set的方法，用的是putOrderedObject，大概是由于这个只是保证禁止指令重排序，开销比putVolatile版本的小吧，并且应该是这里只须要禁止重排序就能够保证并发安全了吧。毕竟有segment锁，并且get方法又有getObjectVolatile
3、ConcurrentHashMap——1.8
1.8的concurrentHashMap真心难懂，特别是扩容还有转移方法……
这个版本的hashMap摒弃了Segment的概念，主要是采用CAS算法，底层用的是和1.8的HashMap同样的数组+链表+红黑树的实现。
emmm关于红黑树就不在这里讲了
先讲几个重要的属性和须要了解的东西
sizeCtl
这个是在ConcurrenthashMap中很重要的一个field，它在流程控制和逻辑代码上起着重要的做用。
-
不一样状态，sizeCtl所表明的含义也有所不一样。
- 未初始化：
- sizeCtl=0：表示没有指定初始容量。
- sizeCtl>0：表示初始容量。
-
初始化中：
- sizeCtl=-1,标记做用，告知其余线程，正在初始化
-
正常状态：
- sizeCtl=0.75n ,扩容阈值
-
扩容中:
- sizeCtl < 0 : 表示有其余线程正在执行扩容
- sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2 :表示此时只有一个线程在执行扩容
- 未初始化：
ForwardingNode
一个用于链接两个table的节点类。它包含一个nextTable指针，用于指向下一张表。并且这个节点的key value next指针所有为null，它的hash值为-1. 这里面定义的find的方法是从nextTable里进行查询节点，而不是以自身为头节点进行查找。
大量的Unsafe和CAS
在ConcurrentHashMap中，随处能够看到U, 大量使用了U.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操做（利用Unsafe来得到底层的相关支持CAS的方法），他能够大大下降锁代理的性能消耗。这个算法的基本思想就是不断地去比较当前内存中的变量值与你指定的一个变量值是否相等，若是相等，则接受你指定的修改的值，不然拒绝你的操做。由于当前线程中的值已经不是最新的值，你的修改极可能会覆盖掉其余线程修改的结果。这一点与乐观锁，SVN的思想是比较相似的。
unsafe静态块
unsafe代码块控制了一些属性的修改工做，好比最经常使用的SIZECTL 。在这一版本的concurrentHashMap中，大量应用来的CAS方法进行变量、属性的修改工做。利用CAS进行无锁操做，能够大大提升性能。
private static final sun.misc.Unsafe U; private static final long SIZECTL; private static final long TRANSFERINDEX; private static final long BASECOUNT; private static final long CELLSBUSY; private static final long CELLVALUE; private static final long ABASE; private static final int ASHIFT; static { try { U = sun.misc.Unsafe.getUnsafe(); Class<?> k = ConcurrentHashMap.class; SIZECTL = U.objectFieldOffset (k.getDeclaredField("sizeCtl")); TRANSFERINDEX = U.objectFieldOffset (k.getDeclaredField("transferIndex")); BASECOUNT = U.objectFieldOffset (k.getDeclaredField("baseCount")); CELLSBUSY = U.objectFieldOffset (k.getDeclaredField("cellsBusy")); Class<?> ck = CounterCell.class; CELLVALUE = U.objectFieldOffset (ck.getDeclaredField("value")); Class<?> ak = Node[].class; ABASE = U.arrayBaseOffset(ak); int scale = U.arrayIndexScale(ak); if ((scale & (scale - 1)) != 0) throw new Error("data type scale not a power of two"); ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); } catch (Exception e) { throw new Error(e); } }
三个关于table中桶的相关操做的核心方法
ConcurrentHashMap定义了三个原子操做，用于对指定位置的节点进行操做。正是这些原子操做保证了ConcurrentHashMap的线程安全。
//得到在i位置上的Node节点 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); } //利用CAS算法设置i位置上的Node节点。之因此能实现并发是由于他指定了原来这个节点的值是多少 //在CAS算法中，会比较内存中的值与你指定的这个值是否相等，若是相等才接受你的修改，不然拒绝你的修改 //所以当前线程中的值并非最新的值，这种修改可能会覆盖掉其余线程的修改结果 有点相似于SVN 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); } //利用volatile方法设置节点位置的值 static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); }
从构造函数的初始化开始看吧
// 这构造函数里，什么都不干 public ConcurrentHashMap() { } public ConcurrentHashMap(int initialCapacity) { if (initialCapacity < 0) throw new IllegalArgumentException(); int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap; }
这个初始化方法有点意思，经过提供初始容量，计算了 sizeCtl，sizeCtl = 【 (1.5 * initialCapacity + 1)，而后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10，那么获得 sizeCtl 为 16，若是 initialCapacity 为 11，获得 sizeCtl 为 32。
emm如今这个sizeCtl好像冲当一个capacity的角色，在put方法中，initial表格的时候好像又会把这个sizeCtl变成相似一个threshold的角色，给扩容的时候作判断。
而后直接经过put方法来更深刻吧
put：
public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); // 获得 hash 值 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) // 初始化数组，后面会详细介绍 tab = initTable(); // 找该 hash 值对应的数组下标，获得第一个节点 f else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是数组该位置为空， // 用一次 CAS 操做将这个新值放入其中便可，这个 put 操做差很少就结束了，能够拉到最后面了 // 若是 CAS 失败，那就是有并发操做，进到下一个循环就行了 if (casTabAt(tab, i, null, new Node<K,V>(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 { // 到这里就是说，f 是该位置的头结点，并且不为空 V oldVal = null; // 获取数组该位置的头结点的监视器锁 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { // 头结点的 hash 值大于 0，说明是链表 // 用于累加，记录链表的长度 binCount = 1; // 遍历链表 for (Node<K,V> e = f;; ++binCount) { K ek; // 若是发现了"相等"的 key，判断是否要进行值覆盖，而后也就能够 break 了 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; } } } 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; } } } } // binCount != 0 说明上面在作链表操做 if (binCount != 0) { // 判断是否要将链表转换为红黑树，临界值和 HashMap 同样，也是 8 if (binCount >= TREEIFY_THRESHOLD) // 这个方法和 HashMap 中稍微有一点点不一样，那就是它不是必定会进行红黑树转换， // 若是当前数组的长度小于 64，那么会选择进行数组扩容，而不是转换为红黑树 // 具体源码咱们就不看了，扩容部分后面说 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // addCount(1L, binCount); return null; }
put是在一个for的无限循环中进行的，大概分为这几个状况：
1. 表格为空，初始化先。
2. 桶中没有东西，那就直接用CAS来为桶中的头指针设值。
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是数组该位置为空， // 用一次 CAS 操做将这个新值放入其中便可，这个 put 操做差很少就结束了，能够拉到最后面了 // 若是 CAS 失败，那就是有并发操做，进到下一个循环就行了 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin }
3. else if ((fh = f.hash) == MOVED)，就帮助数据迁移。
4. 桶中不为空，且不是处于迁移状态，那么就用synchronized来进行put操做。这里面包括找到key的替换操做；没找到key的链表尾插入操做；数量达到阈值的树化操做；原本就是树结点交给树来作put操做
put完后，出去循环，而后addCount方法，这个方法也是有点烦的，大概作了两件事：
-
对 table 的长度加一。不管是经过修改 baseCount，仍是经过使用 CounterCell。当 CounterCell 被初始化了，就优先使用他，再也不使用 baseCount。
-
检查是否须要扩容，或者是否正在扩容。若是须要扩容，就调用扩容方法，若是正在扩容，就帮助其扩容。
下面来慢慢看这些方法。
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { // 初始化的"功劳"被其余线程"抢去"了 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin // CAS 一下，将 sizeCtl 设置为 -1，表明抢到了锁 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // DEFAULT_CAPACITY 默认初始容量是 16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 初始化数组，长度为 16 或初始化时提供的长度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 将这个数组赋值给 table，table 是 volatile 的 table = tab = nt; // 若是 n 为 16 的话，那么这里 sc = 12 // 其实就是 0.75 * n sc = n - (n >>> 2); } } finally { // 设置 sizeCtl 为 sc，咱们就当是 12 吧 sizeCtl = sc; } break; } } return tab; }
这个比较简单，主要就是初始化一个合适大小的数组，而后会设置 sizeCtl，就是让sizeCtl变成一个相似threshold的角色，给扩容的时候作判断用的。
初始化方法中的并发问题是经过对 sizeCtl 进行一个 CAS 操做来控制的。
而后就是关于扩容、transfer的操做了，这个是真滴难懂，代码还长，我暂时先记一下它的思路吧。
关于扩容的思路（来自：https://blog.csdn.net/varyall/article/details/81283231）
jdk8中，采用多线程扩容。整个扩容过程，经过CAS设置sizeCtl，transferIndex等变量协调多个线程进行并发扩容。
先介绍几个相关的field：
nextTable：扩容时，把table中的元素迁移至nextTable，扩容时非空。
sizeCtl：上面讲了，这里放个图：
transferIndex：扩容索引，表示已经分配给扩容线程的table数组索引位置。主要用来协调多个线程，并发安全地获取迁移任务（hash桶）。
其实说白了，就是多个线程来一块儿扩容，每一个线程要作的，就是把当前transferIndex到transferIndex - stride个桶的数据迁移到新table中去。（transferIndex一开始在数组尾巴那，往前挪）
看两个图理解下：
1 在扩容以前，transferIndex 在数组的最右边 。此时有一个线程发现已经到达扩容阈值，准备开始扩容。
2 扩容线程，在迁移数据以前，首先要将transferIndex左移（以cas的方式修改transferIndex=transferIndex-stride(要迁移hash桶的个数)），获取迁移任务。每一个扩容线程都会经过for循环+CAS的方式设置transferIndex，所以能够确保多线程扩容的并发安全。
forwardingNode：扩容索引，表示已经分配给扩容线程的table数组索引位置。主要用来协调多个线程，并发安全地获取迁移任务（hash桶）。简单地说，迁移完的桶就会被设为这个结点，这个结点的hash值是-1，也就是常量MOVED的值。
看看扩容过程：
- 线程执行Put，发现容量要扩容了，这个时候的transferIndex = table.length = 32。
- 扩容线程A 以cas的方式修改transferindex=31-16=16 ,而后按照降序迁移table[31]--table[16]这个区间的hash桶。
- 迁移hash桶时，会将桶内的链表或者红黑树，按照必定算法，拆分红2份，将其插入nextTable[i]和nextTable[i+n]（n是table数组的长度）。 迁移完毕的hash桶,会被设置成ForwardingNode节点，以此告知访问此桶的其余线程，此节点已经迁移完毕。
- 此时，线程2访问到了ForwardingNode节点，若是线程2执行的put或remove等写操做，那么就会先帮其扩容。若是线程2执行的是get等读方法，则会调用ForwardingNode的find方法，去nextTable里面查找相关元素。
5. 若是准备加入扩容的线程，发现如下状况，放弃扩容，直接返回。
-
- 发现transferIndex=0,即全部node均已分配
- 发现扩容线程已经达到最大扩容线程数
总之扩容就是：多线程无锁扩容的关键就是经过CAS设置sizeCtl与transferIndex变量，协调多个线程对table数组中的node进行迁移。
get的分析
public V get(Object key) { Node<K,V>[] tab; Node<K,V> 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; } // 若是头结点的 hash 小于 0，说明 正在扩容，或者该位置是红黑树 else if (eh < 0) // 参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k) 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; }
get 方法历来都是最简单的，这里也不例外：
- 计算 hash 值
- 根据 hash 值找到数组对应位置: (n – 1) & h
- 根据该位置处结点性质进行相应查找
- 若是该位置为 null，那么直接返回 null 就能够了
- 若是该位置处的节点恰好就是咱们须要的，返回该节点的值便可
- 若是该位置节点的 hash 值小于 0，说明正在扩容（-1），或者是红黑树（-2），就用find来解决get获取。（扩容的话若是为-1，说明结点转移了，就去nextTable里面去get）
- 若是以上 3 条都不知足，那就是链表，进行遍历比对便可
我本身对1.8的concurrentHashMap的并发分析：
1. 涉及写的方面
- 若是写的时候发现，table都仍是null，就要初始化，这个初始化显然要考虑并发，这里的并发主要经过对sizeCtl的CAS操做还有table的volatile来实现。扩容是个While循环，第一时间会判断sizeCtl是否是小于0，是的话说明有线程在初始化这个table了，就yield；不然就能够进行初始化操做，第一步就CAS改sizeCtl为-1，若是失败从新进入循环，成功的话就搞一个table赋值给类变量table咯，这个类变量table是volatile因此能够保证其余线程的可见性。
- 若是table中某个桶为空，就直接用cas把新结点设为桶的那个首节点，失败的话会结合外层循环造成自旋CAS。咱们知道table虽然是volatile，但table中的元素的写操做并不具备volatile特性，因此这里是经过CAS操做来解决这个问题的。
- 若是table中某个桶不为空，也就是下面是个链表或者是个红黑树，这个时候就要synchronized(桶的头节点)了，这是put操做惟一一个用到锁的地方，可见对比1.7好了不少，由于锁的颗粒度从segment编导是一个bucket的位置，也就是说桶下面的东西的写操做直接是确定不会有冲突的了。
- 若是须要扩容，则通过CAS设置sizeCtl与transferIndex变量，协调多个线程对table数组中的node进行迁移。
2. 涉及读方面的操做
读咱们能够看到是彻底没有锁的，get能够大体分为三种状况，1. 桶的头节点直接就是那个想要的key，直接就返回这个node的val；2. 若是桶元素的hash值小于0，多是红黑树多是forwardingNode，交给这个node的find方法，可能去新的table中get，也可能交给红黑树get；3. 都不是，就在这个桶下面的链表中遍历。
- 第一种状况，直接读桶，也就是读Node[] table的某个元素，这个为何能够不加锁？？若是读的时候有人在put怎么办？这里我以为主要是经过这个tabAt()方法来解决的，咱们能够看到tabAt方法是用getObjectVolatile来实现的，因此能够保证读对写的可见性吧。
- 第二种状况，另外一个table中find同样的吧，红黑树不作分析……
- 第三种状况，遍历的时候，可能涉及到链表的增加，删除什么的；node结点元素的变化。链表变化，其实就next指针的变化，但next指针是volatile的，因此对写操做是具备可见性的；而后node元素变化？也不怕，由于node的val也是volatile的，在多线程环境下线程A修改结点的val或者新增节点的时候是对线程B可见的。
4、参考文章：
http://www.importnew.com/28263.html——《Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析》本文主要是根据这篇的主线来转载的。
https://www.cnblogs.com/throwable/p/9139947.html——《JAVA中神奇的双刃剑--Unsafe》
https://www.cnblogs.com/seyer/p/5819904.html——网上不少讲解都这个版本
https://blog.csdn.net/varyall/article/details/81283231——《ConcurrentHashMap源码分析（JDK8） 扩容实现机制》这篇讲扩容机制的，思路讲得很清晰了，也有关键源码的分析。
- 1. ConcurrentHashMap总结
- 2. ConcurrentHashMap 总结
- 3. ConcurrentHashMap 总结（ 上 ）
- 4. 简单理解ConcurrentHashMap
- 5. Hashtable 与 ConcurrentHashMap 总结
- 6. HTTP简单总结
- 7. rsync简单总结
- 8. J2SE简单总结
- 9. VGG简单总结
- 10. Vuex简单总结
- 更多相关文章...
- • Docker 资源汇总 - Docker教程
- • Thymeleaf简单格式化输出 - Thymeleaf 教程
- • 算法总结-双指针
- • 算法总结-回溯法
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. ConcurrentHashMap总结
- 2. ConcurrentHashMap 总结
- 3. ConcurrentHashMap 总结（ 上 ）
- 4. 简单理解ConcurrentHashMap
- 5. Hashtable 与 ConcurrentHashMap 总结
- 6. HTTP简单总结
- 7. rsync简单总结
- 8. J2SE简单总结
- 9. VGG简单总结
- 10. Vuex简单总结