一.总体概述
HashMap在开发中很常用，但HashMap存在一个弊端就是线程不安全，解决办法就是使用Hashtable代替或使用Collections.synchronizedMap(m);将HashMap转换为线程安全的，但这两种方法虽说实现了线程安全，但是并发性能比较差，因为是全表加锁，那么并发界的大牛Doug Lea就为我们提供了ConcurrentHashMap解决这个问题，不仅实现线程安全，还保证了并发的，一举两得。
建议：建议先看完JDK1.7及JDK1.8的HashMap源码，再来查看ConcurrentHashMap源码，会简单很多。
- JDK1.7HashMap：https://blog.csdn.net/qq_36625757/article/details/89577865
- JDK1.8HashMap：https://blog.csdn.net/qq_36625757/article/details/90038751
二.JDK1.7实现
在ConcurrentHashMap中不同于普通的HashMap，不仅有数组+链表的概念，还多了一个Segment数组，一个Segment包含多个HashEntry数组(HashMap中的数组) ，这样就能通过分段加锁，解决多线程下的安全性问题，又不至于像Hashtable那样全表加锁，导致性能下降。
1.构造函数
static final int DEFAULT_INITIAL_CAPACITY = 16;//初始HashEntry[]容量大小
static final float DEFAULT_LOAD_FACTOR = 0.75f;//默认加载因子
static final int DEFAULT_CONCURRENCY_LEVEL = 16;//默认并发级别
static final int MAXIMUM_CAPACITY = 1 << 30;//最大HashEntry[]数组大小
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;//Segment最小容量
static final int MAX_SEGMENTS = 1 << 16; //最大segment容量
static final int RETRIES_BEFORE_LOCK = 2;//锁之前重试次数
final Segment<K,V>[] segments;//Segment数组
//空参构造
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
@SuppressWarnings("unchecked")
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//并发等级大于一个segment最大容量，就为一个segment最大容量
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
int sshift = 0;
int ssize = 1;//记录需要创建的Segment数组大小
while (ssize < concurrencyLevel) {//concurrencyLevel默认为16
++sshift;
ssize <<= 1;//每次循环ssize*=2^1，循环完毕后为16，这里依旧是在找大于且最接近concurrencyLevel的2的幂次方数
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
//校验创建的HashEntry[]长度
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
//初始时c=16/16=1
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
//计算每一个Segment中HashEntry[]长度
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// create segments and segments[0] 构建一个Segment，传入加载因子，阈值，创建的HashEntry[]数组，作为Segment[]的第一个元素
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
//创建Segment[]
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0);// ordered write of segments[0] 通过原子操作，将s0添加到Segment[0]上
this.segments = ss;//赋值成员变量
}
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
static final int MAX_SCAN_RETRIES =Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
//Segment对应的HashEntry数组，使用volatile保证可见性
transient volatile HashEntry<K,V>[] table;
transient int count;//HashEntry[]中存储的键值个数
transient int modCount;
transient int threshold;//HashEntry阈值
final float loadFactor;//HashEntry加载因子
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
//.......
}
空参构造调用有参构造，参数为 16,0.75,16，有参构造函数中创建了Segment数组及第一个Segmen对象，并初始化其中的HashEntry[]数组，初始长度为2，concurrencyLevel为并发级别，默认是 16，也就是说 ConcurrentHashMap 有 16 个 Segments，所以理论上，这个时候，最多可以同时支持 16 个线程并发写，只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值，但是一旦初始化以后，它是不可以扩容的。
- Segment []长度为 16，不可以扩容
- Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是之后插入第一个元素不会扩容，第二个就需要扩容了
- 这里初始化了 segment[0]，其他位置还是 null，至于为什么要初始化 segment[0]，之后会讲解。
- 当前 segmentShift 的值为 32 – 4 = 28，segmentMask 为 16 – 1 = 15，可以把它们简单翻译为移位数和掩码，这两个值马上就会用到
2.put()方法
@SuppressWarnings("unchecked")
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);//计算key的hash值
//根据key的hash值计算Segment[]数组中的j索引位置
int j = (hash >>> segmentShift) & segmentMask;//第一次插入数据时segmentShift=28，segmentMask=15
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null)//如果Segment中第j位置为null
s = ensureSegment(j);//初始第j位置的Segment，之前在构造方法中，只初始化了0位置的Segment
return s.put(key, hash, value, false);//插入数据
}
这里通过ensureSegment()对Segment[j]位置的Segment进行初始化
ensureSegment()方法
//初始化Segment[k]位置的元素
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;//获取Segment[]数组
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {//确定要初始化位置的Segment为null
Segment<K,V> proto = ss[0]; // 获得在构造方法中已经初始化的Segment[0]元素
int cap = proto.table.length;//获得其中HashEntry[]数组长度
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) { // 再次确定要初始化位置的Segment为null
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))//使用UNSAFE的CAS，保证Segment[k]初始化成功
break;
}
}
}
return seg;
}
ensureSegment()中使用CAS原子操作，保证初始化。
put()方法（Segment中）
//Segment中的put()方法，参数：要插入的key，key的hash值，值，false
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//分段锁Segment为了保证插入数据的安全性，再插入之前要获取对象独占锁
//Segment继承ReentrantLock，通过tryLock获取独占锁，获取锁返回null
//没有获取到锁，通过scanAndLockForPut()获取
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;//获取其中的HashEntry[]
int index = (tab.length - 1) & hash;//通过Key的hash值计算需要存储在HashEntry的位置
//获取HashEntry[index]位置的链表的第一个元素
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {//当前循环到的链表节点不等于null
K k;
//节点中的key与当前要存储的key相等
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {//替换存储的值
e.value = value;
++modCount;
}
break;
}
e = e.next;//赋值e，用于下次循环
}
else {//当前循环到的链表节点等于null
if (node != null)// 如果node不为 null，那就直接将它设置为链表表头
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);//新建
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)//如果HashEntry[]中存储的键值个数大于阈值，并且小于最大值
rehash(node);//扩容
else
setEntryAt(tab, index, node);//将节点加入到index链表中，设置为链表表头
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();//释放锁
}
return oldValue;
}
这里在第一步会尝试获取锁，如果没有获取到就通过scanAndLockForPut()获取，scanAndLockForPut()执行完毕肯定获取到了Segment锁，所以put方法是线程安全的，scanAndLockForPut()是控制写锁的关键
scanAndLockForPut()方法，Segment中
//Segment中方法
static final int MAX_SCAN_RETRIES =Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);//通过当前segment对象及key的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) {//e等于null，表示链表为null
if (node == null) // speculatively create node
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），那么不抢了，进入到阻塞队列等待锁
else if (++retries > MAX_SCAN_RETRIES) {
lock();//阻塞方法，获取锁成功后返回
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {//当有新的节点进入到链表并且为表头（其他线程操作的）
e = first = f; // 重新设置e及first
retries = -1;//重新设置retries，相当于重新做while循环
}
}
return node;
}
这里会通过一个while循环尝试获取Segment锁，并循环计数，如果retries>1/64(单核1,多核64)，则通过lock()进入AQS阻塞队列，直到获取锁后，才会返回，跳出循环。
rehash()扩容方法，Segment中
//扩容，Segment中，只对Segment中的HashEntry[]扩容
@SuppressWarnings("unchecked")
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;//保存老的HashEntry[]
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;//扩容为原来的2倍
threshold = (int)(newCapacity * loadFactor);//计算新的阈值
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];//创建新的HashEntry[]
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;//计算在HashEntry[]中存放的位置
if (next == null) //当前节点的下一个节点为null，说明当前链表就一个节点
newTable[idx] = e;//直接赋值
else { //存在链表
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
//循环找到一个 lastRun 节点，这个节点之后的所有元素是将要放到一起的
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;//复制链表
//处理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;
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
- 扩容后的数组长度是原数组的2倍。
- 扩容的是Segmen中的HashEntry[]，不是所有的Segment中的HashEntry[]。
- 扩容不需要考虑并发，因为到这里的时候，是持有该 segment 的独占锁的。
- 大for循环中存在两个下for循环，仔细查看发现第一个for循环如果不要也是可以的，但是，这个 for 循环下来，如果 lastRun 的后面还有比较多的节点，那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点，后面的一串节点跟着 lastRun 走就是了，不需要做任何操作。
至此，ConcurrentHashMap的put就分析完成，其保证获取segment的独占锁主要在scanAndLockForPut()方法中，而Segment继承ReentrantLock，只需要调用tryLock()/lock()就可以获取独占锁。
3.get()方法
public V get(Object key) {
Segment<K,V> s;
HashEntry<K,V>[] tab;
int h = hash(key);//计算key的hash值
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
//通过key的Hash值找到Segment及其中的HashEntry[]
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//通过key找到链表并循环
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;
//如果key相等就返回value
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
- 计算 hash 值，找到 segment 数组中的具体位置，并且找到HashEntry[]
- 在HashEntry[]中，根据 hash 找到数组中具体的位置
- 循环链表找到对应key返回value
4.remove()方法
public V remove(Object key) {
int hash = hash(key);//计算Hash值
Segment<K,V> s = segmentForHash(hash);//通过key获取Segment
return s == null ? null : s.remove(key, hash, null);
}
这里比较简单，主要用于获取Key所在Segment，最后调用segment的remove()方法
remove()方法，Segment中
//Segmen中的remove方法，参数为：key，key的Hash，null
final V remove(Object key, int hash, Object value) {
if (!tryLock())//先尝试获取锁
scanAndLock(key, hash);//位获取到，通过scanAndLock()获取
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;//计算key在HashEtry[]中的存储位置
HashEntry<K,V> e = entryAt(tab, index);//获取这个位置上的链表第一个节点
HashEntry<K,V> pred = null;
while (e != null) {//循环
K k;
HashEntry<K,V> next = e.next;//获取下一个节点
//在链表中找到了key所在的节点
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)//pred==null说明key所在节点为链表第一个节点
setEntryAt(tab, index, next);//直接将下一个节点设置为头节点
else
pred.setNext(next);//否则将当前节点的上一个节点的next设置为下一个节点，这样key所在节点就移除了
++modCount;
--count;
oldValue = v;
}
break;
}
pred = e;//记录当前节点，用于下次循环还是表示上一个节点
e = next;//修改下次循环的节点
}
} finally {
unlock();//释放锁
}
return oldValue;
}
这里上来又尝试获取一次锁，如果没有获取到就调用scanAndLock()获取，scanAndLock()会直到获取到锁后才会返回，所以就能保证remove()线程安全，移除节点又分为移除节点是不是链表头节点，是就直接将下一个节点设置为头结点，否则就将当前节点上一个节点的next设置为下一个节点
scanAndLock()方法，类似于上面scanAndLockForPut()方法，Segment中
private void scanAndLock(Object key, int hash) {
HashEntry<K,V> first = entryForHash(this, hash);//获取头节点
HashEntry<K,V> e = first;
int retries = -1;
while (!tryLock()) {//尝试获取锁，失败循环
HashEntry<K,V> f;
if (retries < 0) {
if (e == null || key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {//循环次数>1/64(单核1多核64)
lock();//进入AQS的阻塞队列中，只有获取锁，才会返回
break;
}
//如果发现头节点被其他线程改变了，重新while循环
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f;
retries = -1;
}
}
}
三：JDK1.8实现
JDK1.8中HashMap实现当链表长度大于等于7且数组长度大于64时，会自动将链表转换为红黑树存储，这样的目的是尽量提高查询效率，而在JDK1.8中ConcurrentHashMap相对于JDK1.7的ConcurrentHashMap也做了优化，JDK1.7中ConcurrentHashMap使用分段锁的形式，比较细粒度的控制线程安全性问题，不至于像Hashteble那样，使用全表加锁，限制了并发性，但看过源码后发现，每次在get或put的时候，都会先查询到链表的第一个元素，换个思路想想，如果能就在第一个记录上加锁，这样不也可以解决线程安全性问题，所以JDK1.8就不存在Segmrnt[]了，而是使用CAS+synchronized的形式解决线程安全性问题，这样就比JDK1.7更加细粒度的加锁，并发性能更好。
1.构造方法
private static final int MAXIMUM_CAPACITY = 1 << 30;// node数组最大容量：2^30=1073741824
private static final int DEFAULT_CAPACITY = 16;// 默认初始值，必须是2的幂次方
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;//数组可能最大值，需要与toArray（）相关方法关联
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;//并发级别，遗留下来的，为兼容以前的版本
private static final float LOAD_FACTOR = 0.75f;//加载因子
static final int TREEIFY_THRESHOLD = 8;//链表转红黑树最大节点
static final int UNTREEIFY_THRESHOLD = 6;//红黑树转链表最小节点
static final int MIN_TREEIFY_CAPACITY = 64;//链表转红黑树数组最大长度
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;// 2^15-1，help resize的最大线程数
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;// 32-16=16，sizeCtl中记录size大小的偏移量
static final int MOVED = -1; // forwarding nodes的hash值
static final int TREEBIN = -2; // 树根节点的hash值
static final int RESERVED = -3; // ReservationNode的hash值
static final int NCPU = Runtime.getRuntime().availableProcessors();// 可用处理器数量
transient volatile Node<K,V>[] table;//存放node的数组
/*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义
*当为负数时：-1代表正在初始化，-N代表有N-1个线程正在 进行扩容
*当为0时：代表当时的table还没有被初始化
*当为正数时：表示初始化或者下一次进行扩容的大小*/
private transient volatile int sizeCtl;
private transient volatile int transferIndex;//用于记录迁移数据时已经迁移到的索引位置
//空参构造
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;
}
//链表保存key，value及key的hash值的数据结构。
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
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;
}
//......
}
//红黑树节点
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;
}
//.......
}
//一个特殊的Node节点，hash值为-1，其中存储nextTable的引用。
final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
//.......
}
//存储树形结构的容器,它提供转换黑红树的一些条件和锁的控制
static final class TreeBin<K,V> extends Node<K,V> {
//指向TreeNode列表和根节点
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; // 增加数据时读锁的状态
//.......
}
2.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();
int hash = spread(key.hashCode());//计算key的Hash值
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();//初始化数组
//获取数组上key对应位置的头节点，如果为null表示该位置不存在链表
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//直接创建一个节点，通过CAS插入到对应位置
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//头节点的hash==-1，表示当前有线程在扩容数组，节点为ForwardingNode类型的，
//hash值为-1，数组在扩容时，会将正在迁移数据原数组的链表或红黑树的头结点设置为
//ForwardingNode类型，表示当前链表或红黑树已经前已完成，也就是说当前正在进行数据迁移
//,如果迁移完毕，会将新数组覆盖掉老数组，也就不会出现ForwardingNode类型的节点
else if ((fh = f.hash) == MOVED)
//当前线程帮着移动数据
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {//获取头节点的对象锁
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {//循环链表
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {//当前循环到的节点与要插入的key相等
//替换其中的value
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {//如果当前节点的下一个节点为null，链表中不存在当前要插入的key
//新建一个节点，作为当前节点的下一个节点
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {//如果头结点是红黑树节点
Node<K,V> p;
binCount = 2;
//将键值加入到红黑树中，返回p如果过!=null,说明key在红黑树中重复了
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
//替换value
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)//如果链表长度>=8
treeifyBin(tab, i);//树化链表
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
第一次插入数据时，如果数组不存在就低用initTable()初始化数组
initTable()方法，初始化数组
//初始化数组
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//当构造方法指定数组大小时，sizeCtl为数组大小，不指定默认为0
if ((sc = sizeCtl) < 0)
Thread.yield(); //当前线程让出CPU执行时间片，让其它线程（优先级高于当前线程）或线程取竞争CPU时间片
//这里通过CAS操作，传入当前ConcurrentHashMap对象，期望值，实际值，需要改变为的值
//如果期望值与实际值相等，则将sizeCtl改为-1，否则CAS操作失败返回false
//CAS操作成功后，上一个if判断就成立了
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
//sc如果大于0，初始化的数组大小即为sizeCtl，否则为16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
//初始化数组
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
通过CAS保证只有一个线程初始化数组成功。
casTabAt()方法，CAS操作主要用来设置链表表头
//设置节点到链表中，参数：数组，需要插入的索引，null，新建节点
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
//通过CAS操作，将新建的节点插入到链表表头
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
helpTransfer()方法，其他线程正在扩容，当前线程帮忙迁移数据
//扩容时，帮忙移动数据
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
//这里也在迁移数据，说明当多线程扩容数组时，这个线程是在帮忙迁移数据，这样迁移效率会更快
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
JDK1.8的扩容是比较麻烦的，当一个线程发起扩容后，其他线程也发现数组需要扩容时，这个线程就会去帮忙迁移数据，而不是扩容，这样效率会更高。
treeifyBin()方法，将链表转换为红黑树
//树化链表，将链表转为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//如果数组长度<64，就只扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);//扩容，传入的值为数组长度的2倍
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {//依旧是获取链表头节点的锁
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {//循环链表
//构建新的树节点
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)//第一次循环是，第一个节点为头结点
hd = p;
else
tl.next = p;//上个节点的next=当前节点
tl = p;
}
//将红黑树加入到数组的对应索引位置
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
tryPresize()方法，扩容
//扩容，size为数组长度的2倍
private final void tryPresize(int size) {
//size>=2^30,c=2^30,否则c为大于size+size/2+1且最接近的2^n数
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
if (tab == null || (n = tab.length) == 0) {//如果数组为null，这初始化，其中的逻辑与initTable()中类似
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)//如果新数组的长度<0或>=2^30跳出循环
break;
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node<K,V>[] nt;
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)//如果transferIndex<=0说明前已完毕，当前while循环跳出
break;
//用CAS将sizeCtl加1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);//迁移数据，nt不等于null，说明此线程不是第一个发起迁移数据的线程
}
//将sizeCtl设置为 (rs << RESIZE_STAMP_SHIFT) + 2)
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
//第一次发起迁移数据的线程是传入null，当第二个线程再来迁移数据时，就会执行上面的迁移方法，不会传入null
transfer(tab, null);
}
}
}
要说JDK1.8的ConcurrentHashMap原理比较难，就是在数组的扩容及数据的迁移上比较难懂，下面简单描述一下数据迁移的流程。
迁移数据机制：原数组长度为 n，所以我们有 n 个迁移任务，让每个线程每次负责一个小任务是最简单的，每做完一个任务再检测是否有其他没做完的任务，帮助迁移就可以了，而 Doug Lea 使用了一个 stride，简单理解就是步长，每个线程每次负责迁移其中的一部分，如每次迁移 16 个小任务。所以，我们就需要一个全局的调度者来安排哪个线程执行哪几个任务，这个就是属性 transferIndex 的作用。第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置，然后从后往前的 stride 个任务属于第一个线程，然后将 transferIndex 指向新的位置，再往前的 stride 个任务属于第二个线程，依此类推。当然，这里说的第二个线程不是真的一定指代了第二个线程，也可以是同一个线程。其实就是将一个大的迁移任务分为了一个个任务包。
transfer()方法，数据的迁移
//移动数据，从数组的后面向前迁移
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//stride 在单核下直接等于n，多核模式下为 (n>>>3)/NCPU，最小值是 16
//stride 可以理解为"步长"，有n个位置是需要进行迁移的，
//将这n个任务分为多个任务包，每个任务包有stride个任务
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE;
//nextTab为扩容后的新数组
//如果 nextTab 为 null，先进行一次初始化
//第一个发起迁移的线程调用此方法时，参数 nextTab 为 null
//之后参与迁移的线程调用此方法时，nextTab 不会为 null
if (nextTab == null) {
try {
@SuppressWarnings("unchecked")
//创建一个新的Node[],长度为现有数组长度的2倍，这里就在扩容
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;//赋值给ConcurrentHashMap的成员变量nextTable
//transferIndex是ConcurrentHashMap成员变量，用于控制迁移的位置
transferIndex = n;
}
int nextn = nextTab.length;
//ForwardingNode 翻译过来就是正在被迁移的 Node
//这个构造方法会生成一个Node，key、value 和 next 都为 null，关键是 hash 为 MOVED（-1）
//后面我们会看到，原数组中位置 i 处的节点完成迁移工作后，
//就会将位置 i 处设置为这个 ForwardingNode，用来告诉其他线程该位置已经处理过了
//所以它其实相当于是一个标志。
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//advance指的是做完了一个位置的迁移工作，可以准备做下一个位置的了
boolean advance = true;
boolean finishing = false;
//i为索引，bound为边界
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//advance=true表示可以进行下一个位置的迁移
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
//这里transferIndex一旦小于等于0，说明原数组的所有位置都有相应的线程去处理了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//通过CAS设置transferIndex的值
//nextBound = (nextIndex > stride ? nextIndex - stride : 0),计算下一次迁移的边界
//未迁移的数组长度是否大于步长，大于则下一次迁移的边界等其差值，否则这次就能迁移完毕，下一次迁移的边界为0
//transferIndex=0是表示迁移完毕，tryPresize()中的while就会跳出循环
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//设置边界
bound = nextBound;
//设置下次迁移的起始索引
i = nextIndex - 1;
//设置fals，表示这次迁移完毕
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
//所有的迁移工作已经完后
nextTable = null;
//设置新数组
table = nextTab;
//重新计算sizeCtl：n是原数组长度，所以sizeCtl得出的值将是新数组长度的0.75倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//通过CAS将sizeCtl-1，表示做完自己的任务了
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;//任务结束退出
//如果(sc - 2)== resizeStamp(n) << RESIZE_STAMP_SHIFT
//表述所有迁移任务都已经完成，finishing=true，进入上面的if
finishing = advance = true;
i = n; //索引等于原数组长度
}
}
//如果位置i处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode空节点
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//该位置处是一个 ForwardingNode（ForwardingNode节点的Hash值等于MOVED（-1）），代表该位置已经迁移过了
else if ((fh = f.hash) == MOVED)
advance = true;
else {
synchronized (f) {//获得数组i位置的头节点的对象锁，开始处理其迁移工作
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {//Hash值大于0表示是链表
int runBit = fh & n;//先计算这个节点在数组中存储的位置
Node<K,V> lastRun = f;
//循环找到链表中lastRun节点
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
//通过以下if判断将链表分为两个链表
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//将两个链表分别放在新链表中，对应位置为 原位置 和 原位置索引+原数组长度位置
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
//将原数组的该位置设置为ForwardingNode，表示已经迁移完毕
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {//如果当前节点是红黑树
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {//循环所有节点
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
//以下的if-else还是将红黑树一分为二
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
// 如果一分为二后，节点数少于 8，那么将红黑树转换回链表
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
//将树/链表发到新数组中，对应位置为 原位置 和 原位置索引+原数组长度位置
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
//将原数组的该位置设置为ForwardingNode，表示已经迁移完毕
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
transfer()方法迁移数据，依次只能迁移stride个长度的数据，所以transfer()一次不能迁移完所有数据，需要由tryPresize()方法的控制，tryPresize()方法中当transferIndex <= 0时，表示迁移完毕，其中while就会跳出循环，整个数据的迁移工作也就完成了。
3.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());//计算key的Hash值
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//如果第一节点的key等于当前要查询的key，就返value
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//头结点的hash值<0,表示正在扩容或者其节点红黑树
//参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)
//ForwardingNode中保存了nextTable为扩容的新数组，其中的find方法会在新数组中查询对应的节点
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {//循环链表直到找到key对应节点返回value
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
ForwardingNode.find()方法
Node<K,V> find(int h, Object k) {
//nextTable为新数组
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;
//找到key对应的节点就返回
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;
}
else//是红黑树节点，通过TreeBin的find方法查找
return e.find(h, k);
}
//如果查找完毕还是没找到就返回为null
if ((e = e.next) == null)
return null;
}
}
}
TreeBin.find()方法
final Node<K,V> find(int h, Object k) {
if (k != null) {
//first为红黑树中的第一个节点
for (Node<K,V> e = first; e != null; ) {
int s; K ek;
if (((s = lockState) & (WAITER|WRITER)) != 0) {
//如果节点key相等就返回当前节点
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;
if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
(READER|WAITER) && (w = waiter) != null)
LockSupport.unpark(w);
}
return p;
}
}
}
return null;
}
4.remove()方法
public V remove(Object key) {
return replaceNode(key, null, null);
}
final V replaceNode(Object key, V value, Object cv) {
int hash = spread(key.hashCode());//计算hash值
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0 ||
(f = tabAt(tab, i = (n - 1) & hash)) == null)
break;
else if ((fh = f.hash) == MOVED)//如果正在进行数组的迁移数据，就帮忙迁移数据
tab = helpTransfer(tab, f);
else {
V oldVal = null;
boolean validated = false;
synchronized (f) {//加锁
if (tabAt(tab, i) == f) {
if (fh >= 0) {//是链表
validated = true;
for (Node<K,V> e = f, pred = null;;) {
K ek;
//key的hash值与当前节点相等
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
V ev = e.val;
//cv参数传入就为null，会进入if
if (cv == null || cv == ev ||
(ev != null && cv.equals(ev))) {
oldVal = ev;
if (value != null)
e.val = value;
else if (pred != null)
pred.next = e.next;
else
//将数组的i位置替换为当前节点的下一个节点，意思是移除当前节点
setTabAt(tab, i, e.next);
}
break;
}
pred = e;
if ((e = e.next) == null)//循环到最后了就直接退出
break;
}
}
else if (f instanceof TreeBin) {//红黑树
validated = true;
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> r, p;
//树的根节点不等于null,并且当前要移除的key也有对应的节点
if ((r = t.root) != null &&
(p = r.findTreeNode(hash, key, null)) != null) {
V pv = p.val;
if (cv == null || cv == pv ||
(pv != null && cv.equals(pv))) {
oldVal = pv;
if (value != null)
p.val = value;
else if (t.removeTreeNode(p))//从红黑树中移除p节点
setTabAt(tab, i, untreeify(t.first));//如果需要，将红黑树转为链表设置到数组中
}
}
}
}
}
if (validated) {
if (oldVal != null) {
if (value == null)
addCount(-1L, -1);
return oldVal;
}
break;
}
}
}
return null;
}
四：总结
- JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）。
- JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了。
- JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表。
JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock？
- 因为粒度降低了，在相对而言的低粒度加锁方式，synchronized并不比ReentrantLock差，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势就没有了
- JVM的开发团队从来都没有放弃synchronized，而且基于JVM的synchronized优化空间更大，使用内嵌的关键字比使用API更加自然
- 在大量的数据操作下，对于JVM的内存压力，基于API的ReentrantLock会开销更多的内存，虽然不是瓶颈，但是也是一个选择依据