并发容器线程安全应对之道-ConcurrentHashMap
Posted 博学谷狂野架构师
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了并发容器线程安全应对之道-ConcurrentHashMap相关的知识，希望对你有一定的参考价值。
2 并发容器线程安全应对之道
引言
在前面，我们学习了hashmap
大家都知道HashMap不是线程安全（put、删除、修改、递增、扩容都无锁）的
所以在处理并发的时候会出现问题
接下来我们看下J.U.C包里面提供的一个线程安全并且高效Map（ConcurrentHashMap）
看一下，他到底是如何实现线程并发安全的
2.1 并发容器总体概述
目标：学习ConcurrentHashMap基本概念和认识它的数据结构
ConcurrentHashMap概念：
ConcurrentHashMap是J.U.C包里面提供的一个线程安全的HashMap， 在并发编程中使用的频率（Spring）比较高。
数据结构如下
数组+链表+红黑树+锁（synchronized+cas）
总结：
1、数据结构和hashmap一模一样，唯一的区别就是concurrenthashmap在put、删除、修改、递增、扩容和数据迁移的时候都加锁了（syn or cas）
2、加锁只是锁住一个元素，区别于HashTable(整个表，idea可以查看源码来验证)
2.2 并发容器数据结构与继承
目标：
简单认识下ConcurrentHashMap继承关系
总结
ConcurrentHashMap：实现Serializable表示支持序列化
继承AbstractMap（实现map接口），实现了一些基本操作
实现ConcurrentMap接口，封装了map的基本操作
2.3 并发容器源码深度剖析
测试代码
见put部分2.3.1 并发容器成员变量
目标：认识下ConcurrentHashMap成员变量，先有个印象，方便后续源码分析
private static final int MAXIMUM_CAPACITY = 1 << 30; //table最大容量：2^30=1073741824
private static final int DEFAULT_CAPACITY = 16; //默认容量，必须是2的幂数
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; ////数组的建议最大值
private static final int DEFAULT_CONCURRENCY_LEVEL = 16; //并发级别，1.8前的版本分段锁遗留下来的，为兼容以前的版本
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;// 每次进行转移的最小值
//咦？threshold 呢？？？
2.3.2 并发容器构造器
目标：
先认识下ConcurrentHashMap的5个构造器，看下在构造中（第一步）做了哪些事情
1、ConcurrentHashMap()型构造函数
public ConcurrentHashMap()
总结：该构造函数用于创建一个带有默认初始容量 (16)、负载因子 (0.75) 的空映射
2、ConcurrentHashMap(int)型构造函数
private static final int MAXIMUM_CAPACITY = 1 << 30
public ConcurrentHashMap(int initialCapacity)
if (initialCapacity < 0) // 初始容量小于0，抛出异常
throw new IllegalArgumentException();
//到达最大容量的一半以上后，直接取最大容量！
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
// 初始化，sizeCtl是什么鬼？？看上去是容量……
this.sizeCtl = cap;
总结：该构造函数用于创建一个带有指定初始容量的map
3、ConcurrentHashMap(Map<? extends K, ? extends V>)型构造函数
public ConcurrentHashMap(Map<? extends K, ? extends V> m)
this.sizeCtl = DEFAULT_CAPACITY;
// 将集合m的元素全部放入
putAll(m);
总结：该构造函数用于构造一个与给定映射具有相同映射关系的新映射。
4、ConcurrentHashMap(int, float)型构造函数
public ConcurrentHashMap(int initialCapacity, float loadFactor)
this(initialCapacity, loadFactor, 1);
总结：该构造函数用于创建一个带有指定初始容量、加载因子 新的空映射。
5、ConcurrentHashMap(int, float, int)型构造函数
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel)
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 合法性判断
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
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; // 好像是容量？没那么简单，待会往下看
总结：该构造函数用于创建一个带有指定初始容量、加载因子和并发级别的新的空映射
扩展：和HashMap完全一样？错！我们来看一个实例
1）代码实例
package com.cmap;
import org.openjdk.jol.info.ClassLayout;
import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
public class CMapInit
public static void main(String[] args)
HashMap m = new HashMap(15,0.5f);
ConcurrentHashMap cm = new ConcurrentHashMap(15, 0.5f);
//debug here
System.out.println("before put");
m.put(1,1);
cm.put(1,1);
//and here
System.out.println("after put");
System.out.println(ClassLayout.parseInstance(cm).toPrintable());
2）调试，put之前
3）继续，debug到第二步试试，put之后
- 容量并不是我们之前认为的16，而是32
- 而sizeCtl，我们理解，应该类比于hashMap中的threshold，它应该等于 32*0.5=16才对
- 可是最终为24
这是什么神操作？？？
4）原理剖析
先说结论：方法调用的都是tableSizeFor，只不过，Cmap所计算的参数不一样，注意回顾上面的构造函数
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel)
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
//initial = 15, size = 31
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
//所以tableSizeFor做满1运算前，并不是15本身，而是size，也就是31
//运算后，cap=32 ， 不是16
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
那么它啥时候变成24的呢？
//开始之初，table为null，在put时，会触发table的初始化，也就是以下方法
//从put方法的入口可以追踪到，我们猜想它肯定在这里，初始化table的时候
private final Node<K,V>[] initTable()
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0)
//sc = 原来的sizeCtl也就是 32
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1))
try
if ((tab = table) == null || tab.length == 0)
//n = sc = 32 , 默认就是default=16了
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//创建node数组，长度为n，也就是32
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
//创建完复制给table，初始化完成，也就是我们看到的32长度的数组
table = tab = nt;
// n >>> 2 ，相当于n除以4是8， 32-8=24
//实际效果相当于，n* 3/4 ， 也就是 n*0.75 ， 你指定的0.5在初始化时对它没什么用！
sc = n - (n >>> 2);
finally
//在finally中将它赋给了sizeCtl，也就是我们最终看到的24
sizeCtl = sc;
break;
return tab;
那么sizeCtl起不到threshold的作用，它是干嘛的呢？
其实它的作用远远比hashmap中的thredhold大的多，看看官方的说法：
/**
* Table initialization and resizing control. When negative, the
* table is being initialized or resized: -1 for initialization,
* else -(1 + the number of active resizing threads). Otherwise,
* when table is null, holds the initial table size to use upon
* creation, or 0 for default. After initialization, holds the
* next element count value upon which to resize the table.
*/
private transient volatile int sizeCtl;
翻译过来就是这样子：（官方就这么规定的，记住它！）
- 用来控制table的初始化和扩容操作
- 默认为0，int类型的，废话
- -1 代表table正在初始化
- -N 表示有N-1个线程正在进行扩容操作
其余情况：
- 如果table未初始化，表示table需要初始化的大小。
- 如果table初始化完成，表示table的容量，默认是table大小的0.75倍
而修改它的方法也比较多，initTable只是其中的一个：
- initTable()
- addCount（）
- tryPresize（）
- transfer（）
- helpTransfer（）
2.3.3 put方法
目标：1、ConcurrentHashMap增加的逻辑是什么
2、ConcurrentHashMap是如何保证线程安全的
基础回顾：关于compareAndSwapInt（CAS）
一定要理解CAS的原理，Cmap的精髓就在于cas和sync保障了线程安全，下文的源码分析马上要用到它
（画图展示两个线程的cas交互操作）
(U.compareAndSwapInt(this, SIZECTL, sc, -1))
解释：
- 此方法是Java的native方法，并不由Java语言实现。
- 方法的作用是，读取传入对象this在内存中偏移量为SIZECTL位置的值与期望值sc作比较。
- 相等就把-1值赋值给SIZECTL位置的值。方法返回true。
- 不相等，就取消赋值，方法返回false。
- 一般配合循环重试操作，被for或while所包裹
1）测试代码
package com.cmap;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
public class CMapTest
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
public static void main(String[] args)
ConcurrentHashMap<Integer, Integer> m = new ConcurrentHashMap<Integer, Integer>();
for (int i = 0; i < 64; i++)
if (i == 0)
m.put(i, i);//正常新增（演示）
else if (i == 11)
//容量默认16，临界值=12， 那么i=11正好是第12个值，引发扩容
m.put(i, i);//扩容（演示）
else if (i == 10)
m.put(27, 27);
m.put(43, 43);
else if (i == 9)
else if(i==23)
m.put(i,i); // 23, 第二次扩容
else
m.put(i, i);//正常新增
m.get(8);
System.out.println(m);
//哈希冲突
static void testHashCode()
System.out.println((16 - 1) & spread(new Integer(27).hashCode()));
System.out.println((16 - 1) & spread(new Integer(43).hashCode()));
System.out.println((16 - 1) & spread(new Integer(11).hashCode()));
static final int spread(int h)
return (h ^ (h >>> 16)) & HASH_BITS;
2）增加过程
//提示：该方法岔路比较多，要广度优先阅读，先看外围大路，再细分里面的子方法
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; K fk; V fv;
if (tab == null || (n = tab.length) == 0)
tab = initTable();//表为空的话，初始化表，下面会详细介绍【预留1】
//寻址，找到头结点f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null)
//cas在这里！！！
//插槽为空，cas插入元素
//比较是否为null，如果null才会设置并break，否则到else
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; //插入成功，break终止即可，如果不成功，会进入下一轮for
//helpTransfer() 扩容。下小节详细讲，一个个来……【预留2】
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else
V oldVal = null;
//synchronized 在这里！！！
//插槽不为空，说明被别的线程put抢占了槽
//那就加锁，锁的是当前插槽上的头节点f（类似分段锁）
synchronized (f)
if (tabAt(tab, i) == f) //这步的目的是再次确认，链表头元素没有被其他线程动过
if (fh >= 0) // 正常节点的hash值
binCount = 1; //统计节点个数
//沿着当前插槽的Node链往后找
for (Node<K,V> e = f;; ++binCount)
K ek;
//如果找到相同key，说明之前put过
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek))))
oldVal = e.val;
if (!onlyIfAbsent) //abset参数来决定要不要覆盖，默认是覆盖
e.val = value;
break;
Node<K,V> pred = e;
//否则，新key，新Node插入到最后
if ((e = e.next) == null)
pred.next = new Node<K,V>(hash, key, value);
break;
//如果是红黑树，说明已经转化过，按树的规则放入Node
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;
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
if (binCount != 0)
//如果节点数达到临界值，链表转成树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
addCount(1L, binCount); //计数，如果超了，调transfer扩容
return null;
//compareAndSetObject，比较并插入，典型CAS操作
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v)
return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
3）初始化表方法
多线程下initTable的交互流程：
源码：
/**
* 注意点：先以单线程看业务流程，再类比多个线程操作下的并发是如何处理的？
*/
private final Node<K,V>[] initTable()
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) //自旋
//第1个线程这个if不成立，会进入下面，设置为-1
//第2个线程来的时候if成立，注意理解多线程在跑。
if ((sc = sizeCtl) < 0) //注意回顾上面的值，小于0表示正在初始化，或扩容
Thread.yield();//有线程在操作，将当前线程yield让出时间片。唤醒后进入下一轮while
//CAS操作来设置SIZECTL为-1，如果设置成功，表示当前线程获得初始化的资格
//传入对象 & 内存地址 & 期望值 & 将修改的值
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1))
try
if ((tab = table) == null || tab.length == 0)
//再次确认一下，table是null，还没初始化
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//默认容量16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; //初始化table
//给table赋值，注意这个table是volatile的，会被其他线程及时看到！
//一旦其他线程看到不是null，走while循环发现table不等于空就return了
table = tab = nt;
sc = n - (n >>> 2); //计算下次扩容的阈值，容量的0.75
finally
sizeCtl = sc;
break;
return tab;
总结：
- 判断顺序为先看 table=null 再看 sizeCtl = -1
- T1来得早，按部就班进行
- T2 - T4 在不同时间点进入，行动不一样，有的是被cas挡住，有的被table非null挡住
2.3.4 扩容
目标：1、图解+断点分析查看ConcurrentHashMap是如何扩容的
2、图解+断点分析查看ConcurrentHashMap是如何迁移数据的
测试代码
package com.cmap;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
public class CMapTest
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
public static void main(String[] args)
ConcurrentHashMap<Integer, Integer> m = new ConcurrentHashMap<Integer, Integer>();
for (int i = 0; i < 64; i++)
if (i == 0)
m.put(i, i);//正常新增（演示）
else if (i == 11)
m.put(i, i);//扩容 1
else if (i == 10)
m.put(27, 27);
m.put(43, 43);
else if (i == 9)
else if(i==23)
m.put(i,i); // 23, 第二次扩容（演示点，debug打在这里再进去）
else
m.put(i, i);//正常新增
System.out.println(m);
入口：
/*
在上面， putVal方法的最后， 进 addCount()，再跳到最后，发现：
会走到 transfer() 方法，这是真正的扩容操作
同时，Cmap还带有它的特色，也就是 多线程协助扩容，helpTransfer
最后调的也是transfer方法
*/
final V putVal(K key, V value, boolean onlyIfAbsent)
// ……
addCount(1L, binCount);
private final void addCount(long x, int check)
// ...
// 扩容操作的核心在这里
transfer(tab, null);
/**
* Helps transfer if a resize is in progress. 如果正在扩容，上去帮忙
* tab = 旧数组， f=头结点，如果正在扩容，它是一个ForwardNode类型
*/
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以上是关于并发容器线程安全应对之道-ConcurrentHashMap的主要内容，如果未能解决你的问题，请参考以下文章