ConcurrentHashMap源码分析
2019-04-04 本文已影响0人
HannahLi_9f1c
这篇博客主要是讨论jdk1.8的源码
ConcurrentHashMap的数据结构和HashMap基本一致，也是数组+链表+红黑树结构，不同于Jdk1.7的分段锁segment来保证并发安全性（我认为可能是因为1.7的size()的操作需要全局加锁，有点臃肿），它使用的是volatile,CAS和synchronized来保证安全性。
- 第一部分我们需要先复习一下锁的一些概念，有利于对源码的解读。
- 第二部分分析ConcurrentHashMap的数据结构。
- 第三部分对重要的API进行深入分析
一、重要的锁概念
1. 内存模型
什么是java内存模型
java内存模型可不是什么堆栈，方法区，那个是java运行时数据的逻辑结构。而内存模型指的是程序中各变量的访问规则，即从内存中存取的底层细节，这里的变量包括实例字段、静态字段和构成数组对象等线程共享的元素，但不包括局部变量和方法参数等线程私有数据。
为什么要引入内存模型
Java中的主内存和工作内存可以类比成物理上的主存和高速缓存：
image.png
-
由于处理器和主存速度的差异，限制了CPU的工作，为了更大限度地利用CPU，引入了高速缓存。在多处理器中，为了缓和处理器和内存的速度差，每个处理器上有一个高速缓存，但同时导致了缓存一致性的问题：当多个处理器对同一数据进行修改要写入内存时，内存该以哪个处理器为准呢，这时候就需要规定一系列协议。
image.png
-
java虚拟机中的主内存和工作主存也会出现类似问题，在多线程中，每个线程都有自己的工作内存，多线程同时对主内存取出的数据副本进行修改之后写入主内存，主内存又要以哪个为准呢？
因此需要定义一系列规则来保证多线程并发的安全性。
- 主内存：存放所有变量，主要存放实例化的对象。从底层上看，主内存对应于物理硬件
- 工作内存：存放的数据是主内存副本拷贝，对应于虚拟栈中的数据，为了加快其速度，这些数据一般放在寄存器和高速缓存中执行
. java内存模型
java内存模型规定了所有变量存放在主内存中，每条线程有自己的所使用变量的主存副本拷贝，线程对对象的所有操作（读取，存储）都必须在工作内存中进行，不能直接修改主内存，不同线程之间也不能访问对方的工作内存变量。工作内存和主内存之间的读写具体为以下的不可再分的操作：
- lock
- unlock
- read
- load
- load
- use
- assign
- store
- write
但是这些规则十分繁琐，通常采用等效的判断规则，即先行发生原则
先行发生规则
- 程序次序规则：同一个线程中，程序的执行按照程序控制流进行
- 管程锁定规则：一个unlock先行发生于后面同一个锁的lock
- volatile：对volatile的写先行发生于读操作
- 线程启动规则：Thread的start()方法先行发生于线程的任意动作
- 线程中断规则：对interrupt的调用先行发生于被中断的线程检测到中断事件的发生
- 对象终结规则：一个对象初始化先行发生于finalize（）方法
- 传递性：A先行发生于B，B先行发生于C，那么A先行发生于C
2. volatile规则
volatile具有两个特点：可见性和有序性
- 可见性：每次一个线程修改了volatile变量，其他线程能够立即得知。但是由于它不能保证原子性，所以在有些场景依然不能保证并发安全，如i++，假如一个线程A通过i++试图修改一个变量值，但是由于i++这个操作，需要有三个操作才能增加这个变量，所以有可能在A线程读取到i时，该变量是正确的，下一步操作的时候却被其他线程改变了变量的值，那么就会引发线程不安全。所以使用volatile要注意是否需要加锁才能保证线程安全。
- 防止指令重排序：声明为volatile变量之后，jvm会为他设置一个内存屏障，使它不与其他的普通变量进行指令优化
- 适用场景
- 运算结果不依赖变量当前值，或者保证只有单一线程修改变量的值
- 变量不需要和其他状态变量共同参与不变约束
- 单例模式双重锁校验
public class Singleton {
private volatile static Singleton uniqueInstance;
private Singleton() {
}
public static Singleton getUniqueInstance() {
//先判断对象是否已经实例过，没有实例化过才进入加锁代码
if (uniqueInstance == null) {
//类对象加锁
synchronized (Singleton.class) {
if (uniqueInstance == null) {
uniqueInstance = new Singleton();
}
}
}
return uniqueInstance;
}
}
这是单例模式的线程安全实现，当两个线程同时A,B进入到getUniqueInstance方法时，假设A先进入同步方法，它发现对象还没有被实例化，执行
uniqueInstance = new Singleton();
但是我们知道对象实例化时有以下步骤：
- JVM会去读取Singleton.class文件，并加载进内存
- 在堆内存中开辟空间，分配地址。
- 并在对象空间中，对对象中的属性进行默认初始化
- 调用对应的构造函数，进行初始化
- 初始化完毕后，将堆内存中的地址值赋给引用变量。
由于指令优化，这几个步骤可能不是按顺序执行。如果是单线程，那么不会有影响，最终结果依然正确。但是如果是多线程，A进入同步方法，没有按顺序实例化，就会导致其他线程获得没有被初始化的实例。所以为了并发安全，用volatile声明，这样能够防止CPU对指令重排序。
3. synchronized规则
synchronized同步代码块
- synchronized 同步语句块的实现使用的是 monitorenter 和 monitorexit 指令，其中 monitorenter 指令指向同步代码块的开始位置，monitorexit 指令则指明同步代码块的结束位置。
- 当执行 monitorenter 指令时，线程试图获取锁也就是获取 monitor(monitor对象存在于每个Java对象的对象头中，synchronized 锁便是通过这种方式获取锁的) 的持有权.
- 当计数器为0则可以成功获取，获取后将锁计数器设为1。相应的在执行 monitorexit 指令后，将锁计数器设为0，表明锁被释放。如果获取对象锁失败，那当前线程就要阻塞等待，直到锁被另外一个线程释放为止。
修饰方法
synchronized 修饰的方法并没有 monitorenter 指令和 monitorexit 指令，取得代之的确实是 ACC_SYNCHRONIZED 标识，该标识指明了该方法是一个同步方法，JVM 通过该 ACC_SYNCHRONIZED 访问标志来辨别一个方法是否声明为同步方法，从而执行相应的同步调用。
4. CAS
synchronized能够同时保证可见性、原子性和有序性，但是它是互斥同步的，也就是一个线程如果不能获取锁之后就会阻塞，性能比较低，而CAS是非阻塞同步，如果获取锁失败，不会阻塞而是不断尝试获取锁，这是一种乐观锁的策略。适合在锁竞争不强的时候使用。
- 原理：CAS有三个操作数，内存值，旧值和和新值，如果V符合旧值，处理器用B替换V值，否则就不更新。这是一条处理器层面的操作，具有原子性。
ConcurrentHashMap的数据结构
1. 静态变量，包括初始值和默认参数
private static final int MAXIMUM_CAPACITY = 1 << 30;// 数组的最大容量
private static final int DEFAULT_CAPACITY = 16;// 数组的默认容量，跟HashMap一样默认16
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;//默认并发数
private static final float LOAD_FACTOR = 0.75f;//负载因子，负载因子较大时存储率高，空间利用率高，但是查找速度有所下降
static final int TREEIFY_THRESHOLD = 8;// 当链表长度大于8时树化为红黑树
static final int UNTREEIFY_THRESHOLD = 6;//红黑树节点数小于6，转化为链表
static final int MIN_TREEIFY_CAPACITY = 64;// 树化的最小容量，需要大于比容量的4倍
private static final int MIN_TRANSFER_STRIDE = 16;//跟扩容的转移结点有关，值过小会影响多线程扩容时占用内存很大
/**
* The number of bits used for generation stamp in sizeCtl.
* Must be at least 6 for 32bit arrays.
*/
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;//帮助扩容得线程得最大值
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
/*
* Encodings for Node hash fields. See above for explanation.
*/
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();
/** For serialization compatibility. */
private static final ObjectStreamField[] serialPersistentFields = {
new ObjectStreamField("segments", Segment[].class),
new ObjectStreamField("segmentMask", Integer.TYPE),
new ObjectStreamField("segmentShift", Integer.TYPE)
};
2. 属性
- Node,数组，链表和红黑树中都是存key,value结点
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
...
}
- ForwardingNode,节点的哈希值为-1，其中存储nextTable的引用。表发生扩容的时候，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;
}
transient volatile Node<K,V>[] table;
//数组，声明为volatile是为了保证可见性，一个线程改变了table，其他线程能够可见
private transient volatile Node<K,V>[] nextTable;//扩容的时候使用
private transient volatile long baseCount;
/**
表初始化和扩容参数，
负数表示正在初始化或扩容
-1初始化，
-(1+n)表示n个线程参与扩容
表是空，值为表大小
如果表未初始化，表示表需要初始化的大小。
如果表初始化完成，表示表的容量，默认是表大小的0.75倍，用这个公式算0.75（n - （n >> > 2））
*/
private transient volatile int sizeCtl;
/**
扩容的时候下一个table的下标
*/
private transient volatile int transferIndex;
/**
* Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
*/
private transient volatile int cellsBusy;
/**
* Table of counter cells. When non-null, size is a power of 2.
*/
private transient volatile CounterCell[] counterCells;
// 视图
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
三、辅助API
- 利用hash算法将对象的hashcode算的分布更均匀，减少碰撞，这样hash冲突发生更少，有利于提高查找速度和空间利用率。
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
- tableSizeFor得到一个数的比他大的2的幂
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
假设n的二进制为01xxx...xxx。接着
- 对n右移1位：001xx...xxx，再位或：011xx...xxx
- 对n右移2为：00011...xxx，再位或：01111...xxx
- 此时前面已经有四个1了，再右移4位且位或可得8个1
同理，有8个1，右移8位肯定会让后八位也为1。 - 综上可得，该算法让最高位的1后面的位全变为1。
- 最后再让结果n+1，即得到了2的整数次幂的值了。
- c-1的原因是一个数本身是二次幂比如16的话，这样算得出的就是32.所以需要减一，这样才能得到一个数的最小二次幂
- 初始化数组
初始化在第一次put操作时进行，需要保证只有一个线程对数组进行初始化
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();
//sizeCtl在上面说过，如果负数，说明正在初始化或者在扩容，需要让出线程
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
//尝试用CAS，如果旧值和内存值相同，赋值为-1，这样别的线程就让出线程，如果CAS失败，说明别的线程改变了值，自旋尝试获取锁
try {
if ((tab = table) == null || tab.length == 0) {
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;
}
- 多线程共同扩容
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) {
ForwardingNode是扩容后数组的类
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
//通过CAS试图加入扩容，此时sizeTrl表示正在扩容线程数 (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return 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) { // 初始化nextTab
try {
@SuppressWarnings("unchecked")
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<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 = 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
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
int runBit = fh & n;
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<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 ((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;
}
}
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);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
四、putVal()和get()
1. PutVal()
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//不同于HashMap，ConcurrentHashMap的key和value都不能为null
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)
tab = initTable();//第一次put操作时，才会对数组进行初始化
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
}
else if ((fh = f.hash) == MOVED)//MOVED说明在扩容，调用helpTransfer进行扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//对对象f加锁
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;//fh>=1说明链表有一个元素
for (Node<K,V> e = f;; ++binCount) {
//对链表遍历
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
//找到key相同节点
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);
return null;
}
- key或者value为null，抛出空指针异常。
- 计算hash值
- 进入循环，如果table尚未初始化，则进行初始化
- 计算的数据下标为null，利用cas新建节点插入桶中
- hash值为MOVED，说明正在扩容，当前线程试图参与扩容
- 对节点对象加synchronized锁
- fh>=0,说明不是树。对链表遍历，找到key相同的Node节点，覆盖旧值；找不到ket相同节点，则新建节点
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
- 如果是树的根节点，用putTreeVal将节点加入树中
- 最后判断是否需要树化
2. 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值
if ((tab = table) != null && (n = tab.length) > 0 &&
表不为null
(e = tabAt(tab, (n - 1) & h)) != null) {
//CAS获取桶中第一个节点
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;
}
花了好几天，总算把ConcurrentHashMap看完了，收获很大，对一些锁的概念能够在源码中体会到它们的作用，而不只是了解个大概。总体上来说ConcurrentHashMap是用很多巧妙的方式实现了并发，CAS的原子性操作在这里体现的淋漓尽致，还有不同于1.7使用继承ReentrantLock的分段锁实现并发，这里更多是使用synchronized加锁，1.6以后synchronized的性能在jvm层面上得到很大优化，性能和ReetrantLock相比基本持平。但是Ree
trantLock的优势主要是：①等待可中断;②可实现公平锁;③可实现选择性通知（锁可以绑定多个条件）。
- 参考书籍:深入理解java虚拟机
- 参考博客：link
