本讲的主要内容
一、基础知识夯实
二、ConcurrentHashMap 在1.7 和 1.8 ，不同版本的实现方式
一、基础知识夯实
1.首先是我们需要懂得几点是在java中的集中运算
直接看代码吧
public class IntToBinary { public static void main(String[] args) throws UnsupportedEncodingException { int data = 4; System.out.println("the 4 is "+Integer.toBinaryString(data)); //位与 &(1&1=1 1&0=0 0&0=0) System.out.println("the 4 is "+Integer.toBinaryString(4)); System.out.println("the 6 is "+Integer.toBinaryString(6)); System.out.println("the 4&6 is "+Integer.toBinaryString(4&6)); //位或 | (1|1=1 1|0=1 0|0=0) System.out.println("the 4|6 is "+Integer.toBinaryString(4|6)); //位非~（~1=0 ~0=1） System.out.println("the ~4 is "+Integer.toBinaryString(~4)); //位异或 ^ (1^1=0 1^0=1 0^0=0)(有相同的就是 1 ) System.out.println("the 4^6 is "+Integer.toBinaryString(4^6)); // <<有符号左移 >>有符号的右移 >>>无符号右移 System.out.println("the 1 << 4 is "+Integer.toBinaryString(4 << 1)); System.out.println("the 1 >>> 4 is "+Integer.toBinaryString(4 >>> 1)); //取模的操作 a % (2^n) 等价于 a&(2^n-1) System.out.println("the 345 % 16 is "+(345%16)+ " or "+(345&(16-1))); } }
总结：我们知道的是，位运算是再接再2进制操作的，其他的运算也是最后要转化为 2 进制运算的，所以周期要长。
2.什么是hash：
Hash，一般翻译做“散列”，也有直接音译为“哈希”的，就是把任意长度的输入，通过散列算法，变换成固定长度的输出，该输出就是散列值。这种转换是一种压缩映射，也就是，散列值的空间通常远小于输入的空间，
所有散列函数都有如下一个基本特性：根据同一散列函数计算出的散列值如果不同，那么输入值肯定也不同。但是，根据同一散列函数计算出的散列值如果相同，输入值不一定相同。
两个不同的输入值，根据同一散列函数计算出的散列值相同的现象叫做碰撞。衡量一个哈希函数的好坏的重要指标就是发生碰撞的概率以及发生碰撞的解决方案。任何哈希函数基本都无法彻底避免碰撞，常见的解决碰撞的方法有以下几种：1.开放寻址；2、再散列；3、链地址法（相同hash值的元素用链表串起来）。
3.比较 HashTable、HashMap、ConcurrentHashMap（待补充。。。）；
4.现在我们先说一下，为什么HashMap在高并发下是不安全的，原因是：在多线程情况下，会导致hashmap出现链表闭环，一旦进入了闭环get数据，程序就会进入死循环，所以导致HashMap是非线程安全的。
具体的可以参考文章：https://blog.csdn.net/qq_32534441/article/details/84202979
总结过来就是：
1.Hashmap在插入元素过多的时候需要进行Resize，
Resize的条件是 HashMap.Size >= Capacity * LoadFactor。
2.Hashmap的Resize包含扩容和ReHash两个步骤，ReHash在并发的情况下可能会形成链表环。
二、ConcurrentHashMap 在1.7 和 1.8 ，不同版本的实现方式
因为现在使用1.8的版本还是比较多了，所以我们这里知识知道 1.7 的版本实现方式就好了，主要的我们看一下1.8的实现；
1. JDK 7 的实现方式：
待补充。。。
2. JDK 8 的具体实现：
1).与1.7相比的重大变化
1、 取消了segment数组，直接用table保存数据，锁的粒度更小，减少并发冲突的概率。
存储数据时采用了链表+红黑树的形式，纯链表的形式时间复杂度为O(n)，红黑树则为O（logn），性能提升很大。什么时候链表转红黑树？当key值相等的元素形成的链表中元素个数超过8个的时候。
其实可以看出JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap，相对而言，ConcurrentHashMap只是增加了同步的操作来控制并发，从JDK1.7版本的ReentrantLock+Segment+HashEntry，到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言，总结如下思考
1). JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）
2). JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了
3). JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档
JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock，我觉得有以下几点:
因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了
JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然
在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据
2).主要数据结构和关键变量
1、默认配置参数
/* ---------------- Constants -------------- */ //最大容量 private static final int MAXIMUM_CAPACITY = 1 << 30; //默认容量 private static final int DEFAULT_CAPACITY = 16; //array的大小，只用于toarray方法 static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; //根据这个数来计算segment的个数,segment的个数是仅小于这个数且是2的几次方的一个数（ssize） private static final int DEFAULT_CONCURRENCY_LEVEL = 16; //加载因子 private static final float LOAD_FACTOR = 0.75f; //变成红黑树的链表节点数，超过8就由链表转换成红黑树 static final int TREEIFY_THRESHOLD = 8; //当树节点小于6自动转换成链表 static final int UNTREEIFY_THRESHOLD = 6; // 在转变成树之前，还会有一次判断，只有键值对数量大于 64 才会发生转换。这是为了避免在哈希表建立初期，多个键值对恰好被放入了同一个链表中而导致不必要的转化。 static final int MIN_TREEIFY_CAPACITY = 64; //每次进行转移的最小值这个值作为一个下限来避免Rsisize遇到过多的内存争用 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; // 一系列的标识 static final int MOVED = -1; // hash for forwarding nodes static final int TREEBIN = -2; // hash for roots of trees static final int RESERVED = -3; // hash for transient reservations static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash // 获取可用的CPU个数 static final int NCPU = Runtime.getRuntime().availableProcessors(); // 进行序列化的属性 private static final ObjectStreamField[] serialPersistentFields = { new ObjectStreamField("segments", Segment[].class), new ObjectStreamField("segmentMask", Integer.TYPE), new ObjectStreamField("segmentShift", Integer.TYPE) };
2、Node Entry
需要多注意代码里面的注释，Node的本质是链表，其中ConcurrentHashMap中的HashEntry相对于HashMap中的Entry有一定的差异性：HashEntry中的value以及next都被volatile修饰，这样在多线程读写过程中能够保持它们的可见性。
//Node节点定义，可以看出Node是一个键值对 static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; //ConcurrentHashMap中的HashEntry相对于HashMap中的Entry有一定的差异性：HashEntry中的value以及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(); } 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))); } /** * 它增加了find方法辅助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; } }
3、TreeNodes
/** * 树节点类，另外一个核心的数据结构。 当链表长度过长的时候，会转换为TreeNode。 但是与HashMap不相同的是，它并不是直接转换为红黑树， 而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。 而且TreeNode在ConcurrentHashMap集成自Node类，而并非HashMap中的集成自LinkedHashMap.Entry<K,V>类 */ static final class TreeNode<K,V> extends Node<K,V> { TreeNode<K,V> parent; // red-black tree links TreeNode<K,V> left; TreeNode<K,V> right; TreeNode<K,V> prev; // needed to unlink next upon deletion 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; } Node<K,V> find(int h, Object k) { return findTreeNode(h, k, null); } /** * Returns the TreeNode (or null if not found) for the given key * starting at given root. */ 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; else p = pl; } while (p != null); } return null; } }
4、TreeBin：
存储树形结构的容器，提供转换红黑树的一些条件和锁的控制。
5.核心机制Unsafe与CAS
在ConcurrentHashMap中，随处可以看到U, 大量使用了U.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操作，他可以大大降低锁代理的性能消耗。这个算法的基本思想就是不断地去比较当前内存中的变量值与你指定的一个变量值是否相等，如果相等，则接受你指定的修改的值，否则拒绝你的操作。因为当前线程中的值已经不是最新的值，你的修改很可能会覆盖掉其他线程修改的结果。这一点与乐观锁，SVN的思想是比较类似的。
6.三个核心方法（tabAt 、casTabAt 、setTabAt）
@SuppressWarnings("unchecked")
//获得在i位置上的Node节点
//getObjectVolatile没有使用锁同步，而是使用轻量级同步volatile保证读到的是最新的对象。
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算法中，会比较内存中的值与你指定的这个值是否相等，如果相等才接受你的修改，否则拒绝你的修改
//因此当前线程中的值并不是最新的值，这种修改可能会覆盖掉其他线程的修改结果
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);
}
7.put（方法）
- 懒汉式初始化；
- 如果当前是null，那么就初始化；
- 要是不是null，当前线程要 yeild(),让出执行权；先让相应的线程初始化容器
- 已经初始化前提下，若该位置没有元素，直接插入（new Node<>）（通过CAS操作）,
- 要是有扩容的，先执行扩容操作；
- 扩容之后的长度，都是2 的指数幂（当前是 13 ，完事会是 16）
- 有hash冲突，锁住链表/红黑树节点进行插入；
- 在1/2/3/4完事之后，再判断是不是要转换为红黑树（在链表大于64时。）
- 都完事了，在统计整体长度，是不是要扩容；
具体代码：
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) { //不允许 key或value为null 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; //如果table为空的话，初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); //根据hash值计算出在table里面的位置 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 } //当遇到表连接点时，需要进行整合表的操作 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; //结点上锁 这里的结点可以理解为hash值相同组成的链表的头结点 synchronized (f) { if (tabAt(tab, i) == f) { //fh〉0 说明这个节点是一个链表的节点 不是树的节点 if (fh >= 0) { binCount = 1; //在这里遍历链表所有的结点 for (Node<K,V> e = f;; ++binCount) { K ek; //如果hash值和key值相同 则修改对应结点的value值 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; } } } } if (binCount != 0) { //如果链表长度已经达到临界值8 就需要把链表转换为树结构 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //将当前ConcurrentHashMap的元素数量+1 addCount(1L, binCount); return null; }
我们可以发现JDK8中的实现也是锁分离的思想，只是锁住的是一个Node，而不是JDK7中的Segment，而锁住Node之前的操作是无锁的并且也是线程安全的，建立在之前提到的3个原子操作上。
8.get（方法）
public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; //计算hash值 int h = spread(key.hashCode()); //根据hash值确定节点位置 if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { //如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } //如果eh<0 说明这个节点在树上 直接寻找 else if (eh < 0) 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;
9.size（方法）（弱一致性）
当前的数目+正在并发的数目
Node类存放实际的key和value值。
sizeCtl：
负数：表示进行初始化或者扩容,-1表示正在初始化，-N，表示有N-1个线程正在进行扩容
正数：0 表示还没有被初始化，>0的数，初始化或者是下一次进行扩容的阈值
TreeNode 用在红黑树，表示树的节点, TreeBin是实际放在table数组中的，代表了这个红黑树的根。