1.为什么会有concurrentHashMap？
2.concurrentHashMap数据结构
3.阈值
private static final int MAXIMUM_CAPACITY = 1 << 30;
private static final int DEFAULT_CAPACITY = 16;
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;
4.重要的成员变量
transient volatile Node<K,V>[] table;
默认为null，初始化发生在第一次插入操作，默认大小为16的数组，用来存储Node节点数据，扩容时大小总是2的幂次方。
private transient volatile Node<K,V>[] nextTable;
默认为null，扩容时新生成的数组，其大小为原数组的两倍。
private transient volatile int sizeCtl;
用来控制table的初始化和扩容操作，默认为0。
-1 代表table正在初始化
-N 表示有N-1个线程正在进行扩容操作
其余情况：
1、如果table未初始化，表示table需要初始化的大小。
2、如果table初始化完成，表示table的容量，默认是table大小的0.75倍。
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;
}
//省略部分代码。。。。。。。
}
保存key，value及key的hash值的数据结构。
其中value和next都用volatile修饰，保证并发的可见性。
static 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;
}
}
一个特殊的Node节点，hash值为-1，其中存储nextTable的引用。
只有table发生扩容的时候，ForwardingNode才会发挥作用，作为一个占位符放在table中表示当前节点为null或则已经被移动。
5.构造方法
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY
: tableSizeFor(initialCapacity +(initialCapacity >>> 1) + 1));
//大于(1.5倍initialCapacity+1)的最小的2的幂次
this.sizeCtl = cap;
}
publicConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
publicConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel)
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}
6.实例初始化
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; justspin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2); //0.75*capacity
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
7.put方法
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key ==null || value ==null)throw new NullPointerException();
int hash =spread(key.hashCode());
int binCount =0;
for (Node[] tab =table;;) {
Node f; int n, i, fh;
if (tab ==null || (n = tab.length) ==0)
tab = initTable();
else if ((f =tabAt(tab, i = (n -1) & hash)) ==null) {
if (casTabAt(tab, i, null, new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
} else if ((fh = f.hash) ==MOVED)// 当前Map在扩容，先协助扩容，在更新值。
tab = helpTransfer(tab, f);
else {
V oldVal =null;
synchronized (f) {
if (tabAt(tab, i) == f) {//链表头节点
if (fh >=0) {
binCount =1;
for (Node e = f;; ++binCount) {
K ek;
if (e.hash == hash &&((ek = e.key) == key ||(ek !=null && key.equals(ek)))) {//节点存在，修改节点的值
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node pred = e;
if ((e = e.next) ==null) {// 节点不存在，添加到链表末尾
pred.next =new Node(hash, key, value, null);
break;
}
}
}
else if (finstanceof TreeBin) {
Node p;
binCount =2;
if ((p = ((TreeBin)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;
}
}
}
addCount(1L, binCount); // 统计节点个数，检查是否需要resize
return null;
}
1、如果数组还未初始化，那么进行初始化，这里会进行初始化，这里会通过一个CAS操作将sizeCtl设置为-1，设置成功的，可以进行初始化操作；
2、根据key的hash值找到对应的桶，如果桶还不存在，那么通过一个CAS操作来设置桶的第一个元素，失败的继续执行下面的逻辑，即向桶中插入或更新；
3、如果找到的桶存在，但是桶的第一元素的hash值是-1，说明此时桶正在进行迁移操作，这一块会在下面的扩容中详细谈及；
4、如果找到的桶存在，那么要么是链表结构，要么是红黑树结构，此时需要获取该桶的锁，在锁定的情况下执行链表或红黑树的插入或更新。
如果桶中的第一个元素的hash值大于0，说明是链表结构，则对链表插入或者更新；
如果桶中的第一个元素是TreeBin，说明是红黑树结构，则按照红黑树的方式进行插入或者更新。
在锁的保护下，插入或者更新完毕后，如果是链表结构，需要判断链表中元素的数量是否超过8(默认)，一旦超过，就需要考虑进行数组扩容，或者是链表转红黑树。
8.扩容
什么时候触发扩容？
ConcurrentHashMap的扩容操作可以允许多个线程并发进行，那么就要处理好任务的分配工作。每个线程获取一部分桶的迁移任务，如果当前线程的任务完成，查看是否还有未迁移的桶，若有则继续领取任务执行，若没有则退出。在退出时需要检查是否还有其他线程在参与迁移工作，如果有则自己什么也不做直接退出，如果没有了则执行最后的收尾工作。
1）根据当前数组的长度n，新建一个两倍长度的数组nextTable;
if (nextTab ==null) {// initiating
try {
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n <<1];
nextTab = nt;
}catch (Throwable ex) {// try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
2）初始化ForwardingNode节点，其中保存了新数组nextTable的引用，在处理完每个桶的节点之后当作站位节点，表示该桶已经处理过了
int nextn = nextTab.length;
ForwardingNode fwd =new ForwardingNode(nextTab);
boolean advance =true;
boolean finishing =false; // to ensure sweep before committing nextTab
3）通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值，并初始化i和bound值，i指当前处理的桶序号，bound指需要处理的桶边界；
for (int i =0, bound =0;;) {
Node f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance =false;
else if ((nextIndex =transferIndex) <=0) {
i = -1;
advance =false;
}
else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ?nextIndex - stride :0))) {
bound = nextBound;
i = nextIndex -1;
advance =false;
}
}
4）if判断是否所有的桶都完成了复制，如果完成 而且该线程是最后一个，则进行收尾工作；如果未完成，则帮忙复制；
if (i <0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable =null;
table = nextTab;
sizeCtl = (n <<1) - (n >>>1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc =sizeCtl, sc -1)) {
if ((sc -2) !=resizeStamp(n) <<RESIZE_STAMP_SHIFT)
return;
finishing = advance =true;
i = n; // recheck before commit
}
}
5）如果当前的通为空，则通过CAS插入ForwardingNode节点，告诉其他线程该桶已经处理过了；
else if ((f =tabAt(tab, i)) ==null)
advance =casTabAt(tab, i, null, fwd);
6)如果该桶的hash值为MOVE，表示其他线程正在处理，直接跳过，继续处理下一个
else if ((fh = f.hash) ==MOVED)
advance =true; // already processed
7)如果该桶是一个链表结构，先定义两个变量ln,hn,分别保存hash值的第X位位0和1的节点
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node ln, hn;
if (fh >=0) {
int runBit = fh & n;
Node lastRun = f;
for (Node 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 p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) ==0)
ln =new Node(ph, pk, pv, ln);
else
hn =new Node(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance =true;
}
9.get方法
public V get(Object key) {
Node[] tab; Node 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;
}
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;
}
10
static final int spread(int h) {
return (h ^ (h >>>16)) &HASH_BITS;
}
}
