concurrenthashmap 1.7/1.8

一、ConcurrentHashMap1.7

1.7的ConcurrentHashMap采用分段锁的机制，实现并发的更新操作，底层采用数组+链表的存储结构。
其包含两个核心静态内部类 Segment和HashEntry。

Segment
该类继承了ReentrantLock重入锁来当作锁的角色，每个Segment对象维护了每个散列映射表中的若干个桶（每个桶是由若干个 HashEntry 对象链接起来的链表）
HashEntry
该类用来封装映射表中的键值类

一个ConcurrentHashMap实例中包含若干个 (默认分配16个segment)Segment 对象组成的数组，1.7的ConcurrentHashMap的结构如下:

这样的结构保证了：一个线程占用一把锁（segment）访问其中一段数据的时候，其他段的数据也能被其它的线程访问。

二、ConcurrentHashMap1.8（重点）

ConcurrentHashMap在JDK1.8中取消了segment分段锁，而采用CAS和synchronized来保证并发安全。数据结构跟HashMap1.8的结构一样，数组+链表/红黑二叉树。synchronized只锁定当前链表或红黑二叉树的首节点，这样只要hash不冲突，就不会产生并发，效率又提升N倍。

结构示意图:

CAS的操作：

重要参数源码：

    //最大容量为2的30次方private static final int MAXIMUM_CAPACITY = 1 << 30;//默认大小为16private static final int DEFAULT_CAPACITY = 16;//默认并发数为16private static final int DEFAULT_CONCURRENCY_LEVEL = 16;//负载参数为0.75private static final float LOAD_FACTOR = 0.75f;//链表转换红黑树节点数阈值为8static final int TREEIFY_THRESHOLD = 8;//红黑树转换链表节点数阈值为6static final int UNTREEIFY_THRESHOLD = 6;//链表转换红黑树容量阈值为64（Map容量不到64时，链表转红黑树之前会先扩容）static final int MIN_TREEIFY_CAPACITY = 64;//每个cpu强制处理的最小Map容量数private static final int MIN_TRANSFER_STRIDE = 16;//生成sizeCtl所使用的bit位private static int RESIZE_STAMP_BITS = 16;//参与扩容的最大线程数private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;//移位量，把生成戳移位后保存在sizeCtl中当做扩容线程计数的基数，相反方向移位后能够反解出生成戳private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;// hash值是-1，表示这是一个forwardNode节点static final int MOVED     = -1; // hash值是-2  表示这时一个TreeBin节点static final int TREEBIN   = -2; // ReservationNode的hash值static final int RESERVED  = -3;// 可用处理器数量static final int NCPU = Runtime.getRuntime().availableProcessors();//Map对应的Hash桶数组transient volatile Node<K,V>[] table;//扩容时候新建的Hash桶数组，注意transient关键字，该字段不会被序列化private transient volatile Node<K,V>[] nextTable;//用于节点计数private transient volatile long baseCount;//非常非常重要的一个参数，统领全局//sizeCtl = -1，表示有线程正在进行初始化操作，防止多线程同时初始化Map  //sizeCtl = -(1 + nThreads)，表示有nThreads个线程正在进行扩容操作  //sizeCtl > 0，表示接下来的初始化操作中的Map容量，或者表示初始化/扩容完成后的阈值//sizeCtl = 0，默认值private transient volatile int sizeCtl;//用以维护多线程扩容时候的线程安全private transient volatile int transferIndex;

初始化源码put()方法分析：

public V put(K key, V value) {return putVal(key, value, false);
}/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {//ConcurrentHashMap 不允许插入null键，HashMap允许插入一个null键if (key == null || value == null) throw new NullPointerException();//计算key的hash值int hash = spread(key.hashCode());int binCount = 0;//for循环的作用：因为更新元素是使用CAS机制更新，需要不断的失败重试，直到成功为止。for (Node<K,V>[] tab = table;;) {// f：链表或红黑二叉树头结点，向链表中添加元素时，需要synchronized获取f的锁。Node<K,V> f; int n, i, fh;//判断Node[]数组是否初始化，没有则进行初始化操作if (tab == null || (n = tab.length) == 0)tab = initTable();//通过hash定位Node[]数组的索引坐标，是否有Node节点，如果没有则使用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}//检查到内部正在移动元素（Node[] 数组扩容）else if ((fh = f.hash) == MOVED)//帮助扩容tab = helpTransfer(tab, f);else {V oldVal = null;//锁住链表或红黑二叉树的头结点synchronized (f) {//判断f是否是链表的头结点if (tabAt(tab, i) == f) {//如果fh>=0 是链表节点if (fh >= 0) {binCount = 1;//遍历链表所有节点for (Node<K,V> e = f;; ++binCount) {K ek;//如果节点存在，则更新valueif (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;}}}//TreeBin是红黑二叉树节点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) {//如果链表长度已经达到临界值8 就需要把链表转换为树结构if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}//将当前ConcurrentHashMap的size数量+1，看要不要扩容addCount(1L, binCount);return null;
}

get()方法源码：

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;//用key的hash重新散列，用来获取node[]的位置(下标)int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&//获取下标的位置(e = tabAt(tab, (n - 1) & h)) != null) {// e 为对应下标的初始Nodeif ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}//扩容中else if (eh < 0)//在槽中遍历查找，正在扩容所以节点为ForwardingNode,ForwardingNode中的find实际上是在扩容后的新表中进行查找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;
}

节点Node类：

    //节点的静态内部类，键值对存储的地方static class Node<K,V> implements Map.Entry<K,V> {final int hash;final K key;//val值和下一个节点Node<K,V> next都被volatile关键字修饰，保证线程安全volatile V val;volatile Node<K,V> next;//初始化方法Node(int hash, K key, V val, Node<K,V> next) {this.hash = hash;this.key = key;this.val = val;this.next = next;}public final K getKey()       { return key; }public final V getValue()     { return val; }public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }public final String toString(){ return key + "=" + val; }//为了线程安全setValue不允许调用，会直接抛异常public final V setValue(V value) {throw new UnsupportedOperationException();}//重写equals方法public final boolean equals(Object o) {Object k, v, u; Map.Entry<?,?> e;return ((o instanceof Map.Entry) &&(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&(v = e.getValue()) != null &&(k == key || k.equals(key)) &&(v == (u = val) || v.equals(u)));}//用以支持map.get()方法，会在子类中重写Node<K,V> find(int h, Object k) {Node<K,V> e = this;if (k != null) {do {K ek;if (e.hash == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;} while ((e = e.next) != null);}return null;}}

树节点（TreeNode）类：

    //树节点的静态内部类，与TreeBin共同提供红黑树功能static final class TreeNode<K,V> extends Node<K,V> {//红黑树的基本参数TreeNode<K,V> parent;  TreeNode<K,V> left;TreeNode<K,V> right;//其实还维护着链表指针TreeNode<K,V> prev;    boolean red;//构造方法TreeNode(int hash, K key, V val, Node<K,V> next,TreeNode<K,V> parent) {super(hash, key, val, next);this.parent = parent;}   //重写find方法Node<K,V> find(int h, Object k) {return findTreeNode(h, k, null);}//find方法实现，从树的根部开始遍历节点final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {if (k != null) {TreeNode<K,V> p = this;do  {int ph, dir; K pk; TreeNode<K,V> q;TreeNode<K,V> pl = p.left, pr = p.right;if ((ph = p.hash) > h)p = pl;else if (ph < h)p = pr;else if ((pk = p.key) == k || (pk != null && k.equals(pk)))return p;else if (pl == null)p = pr;else if (pr == null)p = pl;else if ((kc != null ||(kc = comparableClassFor(k)) != null) &&(dir = compareComparables(kc, k, pk)) != 0)p = (dir < 0) ? pl : pr;//递归遍历右子树else if ((q = pr.findTreeNode(h, k, kc)) != null)return q;elsep = pl;} while (p != null);}return null;}}

包含红黑树根结点的Treeibin类：

    //拥有红黑树的根节点，维护着红黑树的读写锁static final class TreeBin<K,V> extends Node<K,V> {TreeNode<K,V> root;volatile TreeNode<K,V> first;volatile Thread waiter;volatile int lockState;//持有写锁状态static final int WRITER = 1;//等待写锁状态static final int WAITER = 2; //持有读锁状态 static final int READER = 4; // 在hashCode相等并且不是Comparable类时才使用此方法进行判断大小static int tieBreakOrder(Object a, Object b) {int d;if (a == null || b == null ||(d = a.getClass().getName().compareTo(b.getClass().getName())) == 0)d = (System.identityHashCode(a) <= System.identityHashCode(b) ?-1 : 1);return d;}//构造方法，根据头节点定义红黑树TreeBin(TreeNode<K,V> b) {super(TREEBIN, null, null, null);this.first = b;TreeNode<K,V> r = null;for (TreeNode<K,V> x = b, next; x != null; x = next) {next = (TreeNode<K,V>)x.next;x.left = x.right = null;if (r == null) {x.parent = null;x.red = false;r = x;}else {K k = x.key;int h = x.hash;Class<?> kc = null;for (TreeNode<K,V> p = r;;) {int dir, ph;K pk = p.key;if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||(dir = compareComparables(kc, k, pk)) == 0)dir = tieBreakOrder(k, pk);TreeNode<K,V> xp = p;if ((p = (dir <= 0) ? p.left : p.right) == null) {x.parent = xp;if (dir <= 0)xp.left = x;elsexp.right = x;r = balanceInsertion(r, x);break;}}}}this.root = r;assert checkInvariants(root);}//根节点加写锁private final void lockRoot() {if (!U.compareAndSwapInt(this, LOCKSTATE, 0, WRITER))contendedLock(); // offload to separate method}//根节点释放写锁private final void unlockRoot() {lockState = 0;}//因为ConcurrentHashMap的写方法会给头节点加锁，所以读写锁不用考虑写写竞争的情况，只用考虑读写竞争的情况private final void contendedLock() {boolean waiting = false;for (int s;;) {//没有线程持有读锁时尝试获取写锁if (((s = lockState) & ~WAITER) == 0) {//没有线程持有写锁时尝试获取写锁if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {//拿到锁后将等待线程清空（等待线程是它自己）if (waiting)waiter = null;return;}}//有线程持有写锁且本线程状态不为WAITER时else if ((s & WAITER) == 0) {//尝试占有waiting线程if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {waiting = true;waiter = Thread.currentThread();}}//有线程持有写锁且本线程状态为WAITER时，堵塞自己else if (waiting)LockSupport.park(this);}}//重写find方法，当写锁被持有时使用链表查询的方法final Node<K,V> find(int h, Object k) {if (k != null) {for (Node<K,V> e = first; e != null; ) {int s; K ek;//写锁被持有时使用链表的方法遍历if (((s = lockState) & (WAITER|WRITER)) != 0) {if (e.hash == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;e = e.next;}//写锁没被持有时，持有一个读锁，用红黑树的方法遍历else if (U.compareAndSwapInt(this, LOCKSTATE, s,s + READER)) {TreeNode<K,V> r, p;try {p = ((r = root) == null ? null :r.findTreeNode(h, k, null));} finally {Thread w;//当当前线程持有最后一个读锁的时候通知waiter线程获取写锁if (U.getAndAddInt(this, LOCKSTATE, -READER) ==(READER|WAITER) && (w = waiter) != null)LockSupport.unpark(w);}return p;}}}return null;}//用以实现Map.putVal的树部分final TreeNode<K,V> putTreeVal(int h, K k, V v) {Class<?> kc = null;boolean searched = false;for (TreeNode<K,V> p = root;;) {int dir, ph; K pk;if (p == null) {first = root = new TreeNode<K,V>(h, k, v, null, null);break;}else if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;else if ((pk = p.key) == k || (pk != null && k.equals(pk)))return p;else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||(dir = compareComparables(kc, k, pk)) == 0) {if (!searched) {TreeNode<K,V> q, ch;searched = true;if (((ch = p.left) != null &&(q = ch.findTreeNode(h, k, kc)) != null) ||((ch = p.right) != null &&(q = ch.findTreeNode(h, k, kc)) != null))return q;}dir = tieBreakOrder(k, pk);}TreeNode<K,V> xp = p;if ((p = (dir <= 0) ? p.left : p.right) == null) {TreeNode<K,V> x, f = first;first = x = new TreeNode<K,V>(h, k, v, f, xp);if (f != null)f.prev = x;if (dir <= 0)xp.left = x;elsexp.right = x;//当父节点是黑节点时候，直接挂一个红节点，不用加锁if (!xp.red)x.red = true;//其余时候可能需要旋转红黑树，重新平衡，这里加写锁else {lockRoot();try {root = balanceInsertion(root, x);} finally {unlockRoot();}}break;}}assert checkInvariants(root);return null;}//移除红黑树节点final boolean removeTreeNode(TreeNode<K,V> p) {TreeNode<K,V> next = (TreeNode<K,V>)p.next;TreeNode<K,V> pred = p.prev;  // unlink traversal pointersTreeNode<K,V> r, rl;if (pred == null)first = next;elsepred.next = next;if (next != null)next.prev = pred;if (first == null) {root = null;return true;}//如果红黑树规模太小，返回True，转换为链表if ((r = root) == null || r.right == null ||(rl = r.left) == null || rl.left == null)return true;//红黑树规模大时，加写锁，在树上删除节点lockRoot();try {TreeNode<K,V> replacement;TreeNode<K,V> pl = p.left;TreeNode<K,V> pr = p.right;if (pl != null && pr != null) {TreeNode<K,V> s = pr, sl;while ((sl = s.left) != null) // find successors = sl;boolean c = s.red; s.red = p.red; p.red = c; // swap colorsTreeNode<K,V> sr = s.right;TreeNode<K,V> pp = p.parent;if (s == pr) { // p was s's direct parentp.parent = s;s.right = p;}else {TreeNode<K,V> sp = s.parent;if ((p.parent = sp) != null) {if (s == sp.left)sp.left = p;elsesp.right = p;}if ((s.right = pr) != null)pr.parent = s;}p.left = null;if ((p.right = sr) != null)sr.parent = p;if ((s.left = pl) != null)pl.parent = s;if ((s.parent = pp) == null)r = s;else if (p == pp.left)pp.left = s;elsepp.right = s;if (sr != null)replacement = sr;elsereplacement = p;}else if (pl != null)replacement = pl;else if (pr != null)replacement = pr;elsereplacement = p;if (replacement != p) {TreeNode<K,V> pp = replacement.parent = p.parent;if (pp == null)r = replacement;else if (p == pp.left)pp.left = replacement;elsepp.right = replacement;p.left = p.right = p.parent = null;}root = (p.red) ? r : balanceDeletion(r, replacement);if (p == replacement) {  // detach pointersTreeNode<K,V> pp;if ((pp = p.parent) != null) {if (p == pp.left)pp.left = null;else if (p == pp.right)pp.right = null;p.parent = null;}}} finally {unlockRoot();}assert checkInvariants(root);return false;}

ForwardingNode类（用以标记已经处理过的Hash桶）

     //一个在扩容方法中使用的内部类，用以标记已经处理过的Hash桶static final class ForwardingNode<K,V> extends Node<K,V> {final Node<K,V>[] nextTable;//构造方法，ForwardingNode节点的Hash值为MOVED，nextTable指向扩容后的新MapForwardingNode(Node<K,V>[] tab) {super(MOVED, null, null, null);this.nextTable = tab;}//重写了Node中的find方法Node<K,V> find(int h, Object k) {//使用循环，避免多次碰到ForwardingNode导致递归过深outer: for (Node<K,V>[] tab = nextTable;;) {Node<K,V> e; int n;if (k == null || tab == null || (n = tab.length) == 0 ||(e = tabAt(tab, (n - 1) & h)) == null)return null;for (;;) {int eh; K ek;if ((eh = e.hash) == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;if (eh < 0) {//遇到ForwardingNode节点的处理，相当于递归操作if (e instanceof ForwardingNode) {tab = ((ForwardingNode<K,V>)e).nextTable;continue outer;}elsereturn e.find(h, k);}if ((e = e.next) == null)return null;}}}}

减少hash碰撞的方法：

//先用高16位异或然后和HASH_BITS进行&计算 ，减少碰撞
static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;
}

重点讲一下扩容机制:

当往hashMap中成功插入一个key/value节点时，有可能触发扩容动作：
1、如果新增节点之后，所在链表的元素个数达到了阈值 8，则会调用treeifyBin方法把链表转换成红黑树，不过在结构转换之前，会对数组长度进行判断，如果数组长度n小于阈值MIN_TREEIFY_CAPACITY，默认是64，则会调用tryPresize方法把数组长度扩大到原来的两倍，并触发transfer方法，重新调整节点的位置。实现如下：

2、新增节点之后，会调用addCount方法记录元素个数，并检查是否需要进行扩容，当数组元素个数达到阈值时，会触发transfer方法，重新调整节点的位置。

transfer实现
transfer方法实现了在并发的情况下，高效的从原始组数往新数组中移动元素，假设扩容之前节点的分布如下，这里区分蓝色节点和红色节点，是为了后续更好的分析：

在上图中，第14个槽位插入新节点之后，链表元素个数已经达到了8，且数组长度为16，优先通过扩容来缓解链表过长的问题，实现如下：
1、根据当前数组长度n，新建一个两倍长度的数组nextTable；

2、初始化ForwardingNode节点，其中保存了新数组nextTable的引用，在处理完每个槽位的节点之后当做占位节点，表示该槽位已经处理过了；

3、通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值，并初始化i和bound值，i指当前处理的槽位序号，bound指需要处理的槽位边界，先处理槽位15的节点；

4、在当前假设条件下，槽位15中没有节点，则通过CAS插入在第二步中初始化的ForwardingNode节点，用于告诉其它线程该槽位已经处理过了；

5、如果槽位15已经被线程A处理了，那么线程B处理到这个节点时，取到该节点的hash值应该为MOVED，值为-1，则直接跳过，继续处理下一个槽位14的节点；

6、处理槽位14的节点，是一个链表结构，先定义两个变量节点ln和hn，按我的理解应该是lowNode和highNode，分别保存hash值的第X位为0和1的节点，具体实现如下：

使用fn&n可以快速把链表中的元素区分成两类，A类是hash值的第X位为0，B类是hash值的第X位为1，并通过lastRun记录最后需要处理的节点，A类和B类节点可以分散到新数组的槽位14和30中，在原数组的槽位14中，蓝色节点第X为0，红色节点第X为1，把链表拉平显示如下：

1、通过遍历链表，记录runBit和lastRun，分别为1和节点6，所以设置hn为节点6，ln为null；
2、重新遍历链表，以lastRun节点为终止条件，根据第X位的值分别构造ln链表和hn链表：
ln链：和原来链表相比，顺序已经不一样了

hn链：

通过CAS把ln链表设置到新数组的i位置，hn链表设置到i+n的位置；
7、如果该槽位是红黑树结构，则构造树节点lo和hi，遍历红黑树中的节点，同样根据hash&n算法，把节点分为两类，分别插入到lo和hi为头的链表中，根据lo和hi链表中的元素个数分别生成ln和hn节点，其中ln节点的生成逻辑如下：
（1）如果lo链表的元素个数小于等于UNTREEIFY_THRESHOLD，默认为6，则通过untreeify方法把树节点链表转化成普通节点链表；
（2）否则判断hi链表中的元素个数是否等于0：如果等于0，表示lo链表中包含了所有原始节点，则设置原始红黑树给ln，否则根据lo链表重新构造红黑树。

最后，同样的通过CAS把ln设置到新数组的i位置，hn设置到i+n位置。
本文参考
本文参考
本文参考