概述
ConcurrentHashMap,通过这个名字，可以知道Concurrent是并发的，HashMap是我们常用的一种用来存放键值对的数据结构，所以ConcurrentHashMap就是一种用来解决高并发的HashMap。这个是JDK在1.5之后提供的一种数据结构。我们知道Java中其实已经有了HashTable这个线程安全的Map，但是他是通过对整个table使用synchronized来保证线程安全的，当有多个线程对HashTable进行读写的时候，每次只能有一个线程会获取到HashTable的对象锁，别的只能处于等待状态，所以在高并发的情况下，它的效率是比较低的。 当然JDK还提供了另外一种方式，那就是Collections.synchronizedMap(hashMap)，这个底层也是通过synchronized对要高并发的数据结构来实现线程安全的。 通过上面的分析可以知道，ConcurrentHashMap为了解决synchronized在高并发时的低效问题，ConcurrentHashMap之所以比HashTable高效，是因为他并不是对整个HashMap上锁，而且在JDK1.8跟之前的JDK1.6，1.7的实现都不一样，下面进行对比一下
- JDK 1.6:ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构， 一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素， 每个Segment守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁。
- JDK 1.8：放弃了Segment的概念，而是直接用Node数组+链表+红黑树的数据结构来实现，并发控制使用Synchronized和CAS来操作，整个看起来就像是优化过且线程安全的HashMap。
继承关系
通过继承关系，可以发现，ConcurrentHashMap并没有继承HashMap,而是单独实现了ConcurrentMap这个接口，其余的跟HashMap是一样的
成员变量
//table的最大容量
private static final int MAXIMUM_CAPACITY = 1 << 30;
//table的默认容量16
private static final int DEFAULT_CAPACITY = 16;
//默认的并发度为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;
//存放Node的数组
transient volatile Node<K,V>[] table;
private transient volatile Node<K,V>[] nextTable;
private transient volatile long baseCount;
//sizeCtl作为table是否在初始化或者在resize的标志，如果为负数说明正在初始化或者扩容，需要table进行CRUD操作的线程只能等待，稍后进行重试，否则就可以直接进行CRUD操作
//正数则表明table没有初始化
private transient volatile int sizeCtl;
//该节点是否已经移动到新数组中去
static final int MOVED = -1;
//是否是TREEBIN节点
static final int TREEBIN = -2;
除此之外，还有几个类需要解释一下
基本原理
TreeNode
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<K,V> find(int h, Object k) {
return findTreeNode(h, k, null);
}
//查找相应的树节点
final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
//此处省略一万行代码......
}
跟一般的树节点都差不多，在前面分析二叉树的时候也差不多是这种数据结构，所以还是比较好理解的，注释说Nodes for use in TreeBins，也就是说这个TreeNode是用来服务TreeBin的，那么继续看Treebins
TreeBins
这个通过注释可以知道，是用来包装TreeNode的，然后作为红黑树的树节点
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;
volatile TreeNode<K,V> first;//TreeNode的根节点
volatile Thread waiter;//waiter线程
volatile int lockState;//当前节点的锁状态
// values for lockState
static final int WRITER = 1; // 获得写数据的锁状态
static final int WAITER = 2; // 等待写数据的锁状态
static final int READER = 4; // 添加数据时的锁状态
TreeBin(TreeNode<K,V> b) {
super(TREEBIN, null, null, null);
this.first = b;//头结点赋值
TreeNode<K,V> r = null;
//递归遍历传过来的Node，从根节点开始
for (TreeNode<K,V> x = b, next; x != null; x = next)
{
next = (TreeNode<K,V>)x.next;//拿到下一个节点
x.left = x.right = null;//将左右节点置空，因为需要重新赋值
//判断r节点是否为空
if (r == null) {
x.parent = null;//根节点置空
x.red = false;//黑子树
r = x;//x赋值为r节点
}
else {
K k = x.key;//拿到key
int h = x.hash;//计算hash值
Class<?> kc = null;//K对应的泛型实体类
//从根节点r开始遍历
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;
else
xp.right = x;
r = balanceInsertion(r, x);
break;
}
}
}
}
this.root = r;//将r赋值给根节点
assert checkInvariants(root);
}
}
ForwardingNode
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<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
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) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
}
ForwardingNode有两个功能，一个是用来指向nextTable节点，便于将原有的table数组复制到新的table中，另外一个是用来占位空节点，告诉其他节点这个节点已经被处理过了，自动遍历后面的节点。
CAS操作
tabAt
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);
}
获得在i索引位置上的Node节点
casTabAt
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);
}
- 1.利用CAS算法设置i位置上的Node节点。之所以能实现并发是因为他指定了原来这个节点的值是多少
- 2.在CAS算法中，会比较内存中的值与你指定的这个值是否相等，如果相等才接受你的修改，否则拒绝你的修改
setTabAt
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
利用volatile方法设置节点位置的值
核心方法
构造方法
之前分析过HashMap的源码，所以看这些还是比较轻松的，ConcurrentHashMap采用的是懒加载，也就是存入第一对KeyValue的时候才会去初始化table，所以前面四个构造方法只是做了一些参数配置，只有最后一个构造方法，才会去真正的初始化，下面重点看一下最后一个方法
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
public void putAll(Map<? extends K, ? extends V> m) {
tryPresize(m.size());
for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
putVal(e.getKey(), e.getValue(), false);
}
最后调用了putAll方法，这个会在后面的put方法里面再重点讲解。
transfer方法
transfer实际上执行的就是table数组的扩容操作，
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//创建一个容量是原来2倍的table数组
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;
transferIndex = n;
}
int nextn = nextTab.length;
//创建一个ForwardingNode，指向新table，并且用于填充已经移动的Node位置
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> 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;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
//如果所有的节点都已经完成复制工作
//就把nextTable赋值给table 清空临时对象nextTable
nextTable = null;
table = nextTab;
// 扩容至现在容量的0.75倍
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
}
}
else if ((f = tabAt(tab, i)) == null)
//用ForwardingNode来填充Null节点
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
//如果当前节点移动过，将次标志设置为true
//这样后续线程访问便可以直接跳过此节点
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
//普通Node节点
int runBit = fh & n;//通过runBit将链表打散，便于高效遍历
Node<K,V> lastRun = f;
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 ((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);
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {
//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);
//通过h & n的值将TreeBin的节点进行打散
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;
}
}
//如果扩容后已经不再需要tree的结构 反向转换为链表结构
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;
//在nextTable的i位置上插入一个链表
setTabAt(nextTab, i, ln);
//在nextTable的i+n的位置上插入另一个链表
setTabAt(nextTab, i + n, hn);
//在table的i位置上插入forwardNode节点 表示已经处理过该节点
setTabAt(tab, i, fwd);
//设置advance为true 返回到上面的while循环中 就可以执行i--操作
advance = true;
}
}
}
}
}
}
- 1.根据当前数组长度n，新建一个两倍长度的数组nextTable；
- 2.初始化ForwardingNode节点，其中保存了新数组nextTable的引用，在处理完每个槽位的节点之后当做占位节点，表示该槽位已经处理过了
- 3.通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值
- 4.如果槽位中没有节点，则通过CAS插入在第二步中初始化的ForwardingNode节点，用于告诉其它线程该槽位已经处理过了
- 5.如果当前Node已经被线程A处理了，那么线程B处理到这个节点时，取到该节点的hash值应该为MOVED，值为-1，则直接跳过，继续处理下一个Node
- 6.处理不为null的节点，如果是链表结构，先定义两个变量节点ln和hn,通过CAS把ln链表设置到新数组的i位置，hn链表设置到i+n的位置
- 7.如果该槽位是红黑树结构，同样将节点分成两类，通过CAS插入到nextTable中
helpTransfer
如果一个线程在进行扩容的时候中断了，但是此时nextTable已经创建了，那么他就会调用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;
}
这是一个协助扩容的方法。这个方法被调用的时候，当前ConcurrentHashMap一定已经有了nextTable对象，首先拿到这个nextTable对象，调用上面讲到的transfer方法来进行扩容
put方法
首先调用了put方法
public V put(K key, V value) {
return putVal(key, value, false);
}
紧接着调用了putVal方法，下面来分析下一年putVal方法
final V putVal(K key, V value, boolean onlyIfAbsent) {
//key和value都不能为空
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//计算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)
//table为null或者table的长度为0，初始化table
tab = initTable();
//根据hash值计算table的索引
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果索引处为null,则直接在此处进行插入
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
//通过CAS进行插入，插入成功之后跳出循环
break;
}
else if ((fh = f.hash) == MOVED)
//当遇到表ForwardNode时，需要对表进行整合
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;
//如果hash值跟key值都相同，则进行值替换
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) {
//判断链表长度，如果链表长度大于阈值，进行树节点转换
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//改变table中的元素的个数
return null;
}
下面分解一下步骤：
- 1.计算记录的key的hashCode，然后通过hashcode计算table的index位置
- 2.获取table[index]，为null直接采用casTabAt插入一条新的记录
- 3.不为null，则锁住根节点，然后通过节点的hashcode值来区分是链表还是红黑树
- 4.遍历链表或者红黑树，找到相应的节点，进行更新操作
- 5.如果是更新操作，则不需要进行transfer操作，否则判断一下更新之后的size有没有超过阈值，主要是判断链表的节点数跟table的数组长度
如何保证线程安全
当对table中的某个节点进行写操作的时候，如果为null,直接进行插入，注意如果不为null，则用syncronized关键字锁住当前节点，然后进行写操作。为什么为空的时候插入不需要上锁，因为ConcurrentHashMap采用的是CAS的乐观锁机制，如果有线程在进行修改，就会在外面等待，然后过一会儿再过来查询一下，这样可以保证高效并发。
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());//计算hash值
//根据hash值定位元素
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
//如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点
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;
}
- 1.通过计算key的hash值进而计算key在table中的index
- 2.拿到table[index],为null,直接返回null
- 3.如果table[index]不为null，通过Node的hash值来判断是链表还是红黑树
- 4.遍历到相应的value之后直接返回，否则返回null
如何保证线程安全 首先当多个线程同时读取的时候，ConcurrentHashMap给每个Node节点加了volatile关键字修饰来保证可见性。
总结
看了很多数据结构，感觉ConcurrentHashMap是最复杂的，里面涉及到了很多的CAS操作，以及硬件层的线程安全操作，分析了好几天，才弄懂，下面总结一下c的JDK1.7跟1.8的区别。
JDK 1.7
JDK1.8
对比分析
- 1.7采用链表，1.8使用红黑树来替换之前的链表，提高了查找的效率。
- 1.7版本锁的粒度是基于Segment的，采用的是分段锁，粒度是每个数组1.8的实现降低锁的粒度到每个Node节点，
- 1.7采用的Segment继承自ReentrantLock，自带锁，1.8的数据结构采用数组+链表+红黑树，使用了大量 的CAS操作，提升了高并发时的性能，丢弃了分段锁的概念。
- 1.7采用的是ReentrantLock进行同步，1.8 采用了synchronized进行同步，其实一开始没理解，因为ReentrantLock比synchronized更加灵活，后来思考了一下觉得是因为锁的粒度进一步降低，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势不大。
