ConcurrentHashMap介绍
ConcurrentHashMap是一个 在juc包下的 map， 线程安全。 在jdk.1.7 之前采用数组+ 链表的结构 并且采用分段锁机制 来保证线程安全，而jdk1.8之后 他改成了 数组+ 链表+ 红黑树，线程安全方面也改成了 cas+ synchronized 来保证线程安全。下面我们来看看ConcurrentHashMap的 源码是怎样实现的(jdk1.8)
属性
// 散列表最大容量
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 散列表默认容量
private static final int DEFAULT_CAPACITY = 16;
// 最大数组长度
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
// 默认并发级别 jdk1.7 之前遗留的 1.8只用于初始化
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 final int RESIZE_STAMP_BITS = 16;
// 65535 标识并发扩容最大线程数量
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 扩容相关
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// node 结点的hash 是-1 表示 当前结点是forwardingNode结点
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
// 0x7fffffff = 31个1 用于将一个负数变成一个正数 但是不是取绝对值
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
// 系统CPu数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
// 散列表
transient volatile Node<K,V>[] table;
// 扩容用的临时散列表
private transient volatile Node<K,V>[] nextTable;
// LongAdder 的baseCount
private transient volatile long baseCount;
/**
sizeCtl <0
1. -1 的时候 表示table正在初始化(有线程正在初始化 ， 当前线程应该自旋等待)
2. 其他情况 表示当前map正在进行扩容 高16位表示 扩容的标识戳 ， 低16位表示 扩容线程数量
sizeCtl = 0
表示创建数组 使用默认容量 16
sizeCtl >0
1. 如果table 未初始化 表示 初始化大小
2. 如果table 已经初始化 表示下次扩容的阈值
*/
private transient volatile int sizeCtl;
//扩容过程中，记录当前进度，所有线程都需要从transferIndex中分配区间任务，去执行自己的任务
private transient volatile int transferIndex;
// 0 表示 无锁 1 表示加锁
private transient volatile int cellsBusy;
// LongAdder 中的cells 数组 当baseCount发生竞争后 会创建cells 数组
// 线程会通过计算hash值 取到自己的cell中
private transient volatile CounterCell[] counterCells;
内部类
- 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);
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;// n 长度 e扩容创建的新表使用的寻址算法的桶位
//在新扩容表中 重新定位hash对应的结点
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;
}
}
}
}
static class Node<K,V> implements Map.Entry<K,V> {
final int hash; // key的哈希值
final K key; // key值
volatile V val; // value
volatile Node<K,V> next; // 下一个结点
Node(int hash, K key, V val) {
this.hash = hash;
this.key = key;
this.val = val;
}
Node(int hash, K key, V val, Node<K,V> next) {
this(hash, key, val);
this.next = next;
}
public final K getKey() { return key; }
public final V getValue() { return val; }
public final int hashCode() { return key.hashCode() ^ val.hashCode(); }
public final String toString() {
return Helpers.mapEntryToString(key, val);
}
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
public final boolean equals(Object o) {
Object k, v, u; Map.Entry<?,?> e;
return ((o instanceof Map.Entry) &&
(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
(v = e.getValue()) != null &&
(k == key || k.equals(key)) &&
(v == (u = val) || v.equals(u)));
}
/**
* Virtualized support for map.get(); overridden in subclasses.
*/
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}
put 方法及其相关
public V put(K key, V value) {
return putVal(key, value, false);
}
// onlyIfAbsent --> ture 表示如果遇到相同的key 进行不进行置换 false 表示置换
final V putVal(K key, V value, boolean onlyIfAbsent) {
// key && value 都不能是null
if (key == null || value == null) throw new NullPointerException();
// 通过扰动函数 计算出hash 高16位也参与运算
int hash = spread(key.hashCode());
// binCount 表示当前k-v 封装成node 后插入到指定桶位后 在桶位中所属链表的下标位置
// 表示当前桶位已经树化成红黑树
int binCount = 0;
for (Node<K,V>[] tab = table;;) {// 自旋过程
/*
f ->头结点 n->代表table的长度 i->索引
fh->头结点hash fk->头结点的key fv->头结点的value
*/
Node<K,V> f; int n, i, fh; K fk; V fv;
// 判断哈希表有没有创建 如果没有就执行初始化哈希表的操作
if (tab == null || (n = tab.length) == 0)
// 初始化哈希表
tab = initTable();
// tabAt(tab,i =(n-1)&hash) 得到头结点
// 判断头结点是不是为空 前置条件 表已经创建
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//成功 如果头结点是空 直接用cas 的操作 尝试用new的结点替换 头结点
// 失败 说明其他线程已经插入了 需要自旋 继续操作
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break;
}
// 前置条件 表已经创建 && 头结点不是空
// 判断头结点的hash 是不是等于-1 看是不是forwardingNode结点 如果是 说明哈希表正在处于扩容的情况，
else if ((fh = f.hash) == MOVED)
// 当前线程帮忙扩容
tab = helpTransfer(tab, f);
// 前置条件 表已经创建 && 头结点不是空 && 不处于扩容中
//头结点是key相同的结点 并且不支持替换 就直接返回当前结点的值
else if (onlyIfAbsent
&& fh == hash
&& ((fk = f.key) == key || (fk != null && key.equals(fk)))
&& (fv = f.val) != null)
return fv;
// 前置条件 表已经创建 && 头结点不是空 && 不处于扩容中 && 头结点不是key相同的结点
else {
// 用于记录旧的value
V oldVal = null;
// 锁住头结点
synchronized (f) {
// 再次查询头结点是不是等于f
// 防止你加锁的过程中 别人已经修改了头结点的值，导致操作出现问题
if (tabAt(tab, i) == f) {
// 前置条件 头结点没有改变
// 判断头结点 是不是普通链表
if (fh >= 0) {
// 当前列表的末尾，bincount表示链表的长度
//2 : 当前插入的key 与链表的某个元素key 一样 当前插入操作表示冲突位置 （binCount-1）
binCount = 1;
// 自旋
for (Node<K,V> e = f;; ++binCount) {
K ek; // 迭代遍历结点的key
if (e.hash == hash && // 找到了key相同的结点
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;// 保留老的值
if (!onlyIfAbsent)
e.val = value;// 更新value
break;
}
// 记录用pred记录上一个结点
Node<K,V> pred = e;
// 判断next 是不是为空
if ((e = e.next) == null) {
// 没有key相同的结点 就 插入在末尾
pred.next = new Node<K,V>(hash, key, value);
break;
}
}
}
// fh<0 说明是Treebin
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");
}
}
// 当前桶位不是 null 是红黑树 或者 链表
if (binCount != 0) {
// 是不是需要树化
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null) // key 有没有相同的 发生冲突的情况
return oldVal;
break;
}
}
}
// 统计table 一共有多少数据
// 是不是需要扩容
addCount(1L, binCount);
return null;
}
private final void addCount(long x, int check) {
//cs 是cells 单元
//b 是未发生竞争的base
// s 是元素数量
CounterCell[] cs; long b, s;
// LongAdder 操作
if ((cs = counterCells) != null ||
!U.compareAndSetLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell c; long v; int m;
boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 ||
(c = cs[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
// 如果check >=0 说明一定是put 调用的 addCount
if (check >= 0) {
// tab 表示 map.table
// nt 是nexttable
//n 是数组长度
//sc 是sizeCtl的长度
Node<K,V>[] tab, nt; int n, sc;
//自旋
//条件1 s >= (long)(sc = sizeCtl)
// ture 1: 当前sizeCtl 是一个负数 表示正在扩容中。
// 2 : 当前sizeCtl 是一个正数 表示扩容阈值
// false 表示当前table 尚未达到扩容条件
// 条件3 ： 当前table长度小于最大值限制 可以扩容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
// 获取到唯一标识戳
int rs = resizeStamp(n);
//条件1
// ture 说明当前线程获取的扩容唯一标识撮 不是本次扩容
// false 说明当前线程获取到的扩容唯一标识撮 是本次扩容
// 条件2
// ture 表示扩容完毕了 false 表示在扩容过程中
// 条件3
// ture : 触发的帮助扩容的达到最大值 65535-1
// false 可以参与
// 条件4
// ture 扩容结束
// false 扩容正在进行
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// 当前线程正在扩容中 当前线程参与
// ture 可以参与
//fasle 进行自旋 大概率还会来到这里
//条件失败 1 当前有很多线程都尝试修改sizeCtl 可能会导致和内存的不一样
// transfer 任务内的线程也修改sizeCtl
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1))
// 协助扩容线程 支持nextTable参数
transfer(tab, nt);
}
// 条件成立 说明当前线程是触发扩容的第一个线程 在transfer方法需要做一些扩容的准备工作
else if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
//触发扩容的线程 不持有nextTable
transfer(tab, null);
s = sumCount();
}
}
}
get方法
public V get(Object key) {
// tab 引用map.table
// e 当前元素
// p 目标结点
// n 长度
// eh 当前元素的hash
// ek 当前元素的key
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());// 通过扰动运算后得到 更散列的hash值
// 表已经创建了 而且头结点不等于null
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;
}
// -1 fwd结点说明table正在扩容 且当前查询的已经被迁移走了
// -2 Treebin 需要使用Treebin方法查询
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;
}
扩容方法及其相关
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
// n 扩容之前table数组的长度 tride 分配给线程任务的步长
int n = tab.length, stride;
// 假设stride 为16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// 触发扩容的线程 需要做一些扩容准备工作
if (nextTab == null) {
try {
// 创建了比扩容之前大1倍的table
@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;// 记录迁移数据整体位置的一个标记 从1开始
}
// 新表长度
int nextn = nextTab.length;
// 新表的引用 当某个桶位处理完毕 将设置为fwd 结点
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;// 推进标记
boolean finishing = false; // 完成标记
// i 表示分配给当前线程任务，执行到的桶位
// bound 表示分配给当前线程的下界限制
int i = 0, bound = 0;
for (;;) {// 自旋
// f 桶位的头结点 fh 头结点的hash
Node<K,V> f; int fh;
/**
* 1 给当前线程分配区间
* 2 维护当前线程任务进度（i 当前处理的桶位
* 3 维护map对象全局范围内的进度
*/
while (advance) {
// 分配任务区间的变量 nextIndex分配开始的下标 nextBound 结束下标
int nextIndex, nextBound;
// --i>=bound 当前线程任务尚未完成， 还有相应的区间桶位要处理 --i 表示下一个处理的桶位
// 不成立 表示已经完成 或者未分配
if (--i >= bound || finishing)
advance = false;
// 前置条件 任务已经完成 或者还没有分配任务
// 条件成立 表示对象全局范围内的桶位都分配完毕了 没有区间分配了
// 设置全局i变量为-1 执行退出迁移相关的程序
// 不成立 全局范围内的桶位尚未分配还有区间可分配
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// 前置条件 当前线程需要分配任务区间 2 全局范围内的桶位尚未分配还有区间可分配
// 条件成功 说明给当前任务分配成功
// 失败 就是 说明分配给当前线程失败，可能发生竞争
else if (U.compareAndSetInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
// 区间复制给bound
bound = nextBound;
i = nextIndex - 1;
advance = false; // 结束
}
}
// i<0 的情况 表示当前线程未分配到任务
if (i < 0 || i >= n || i + n >= nextn) {
int sc;// sizeCtl 的临时变量
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 条件成立 说明 设置sizeCtl 低16位-1成功 当前线程可以退出
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//条件成立 说明当前线程不是最后一个退出transfer任务线程
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
// 退出
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
//前置条件
//casr1:说明当前桶未 没有存放数据 只需要将此处设置为fwd结点
// 如果不成立 说明桶未有数据
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//casr2:条件成立 说明当前桶位已经迁移过了，当前线程不用再处理
// 直接在此更新当前线程任务索引，在处理下一个桶位。
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
// 前置条件 当前桶位有数据 而且node结点不是fwd结点 说明这些数据需要迁移
else {
// sync 加锁当前桶位的头结点
synchronized (f) {
// 防止 在你加锁投之前 头对象被其他线程修改过 导致你加锁对象错误
if (tabAt(tab, i) == f) {
// ln 低位链表 hn 高位链表
Node<K,V> ln, hn;
if (fh >= 0) { // 条件成立表示当前桶位是链表
//lastRun
// 可以获取当前链表 末尾连续高位不变的 node
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;
}
}
// 说明 lastRun链表 为低位链表 让ln 引用lastRun
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
//迭代链表 当循环结点 不等于lastRun
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);
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;
}
}
}
}
}
}
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
// nextTable 就是fwd.nextTable
Node<K,V>[] nextTab; int sc;
//条件三:
// 恒成立
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
// 拿当前表的长度 获取扩容标识戳
int rs = resizeStamp(tab.length);
// 条件1 ：
// 成立 标识当前扩容正在执行
//不成立 1. nextTable被设置为null 已经扩容完毕了
// 2。 再次出发扩容了 nextTable 已经过期了
//条件2 成立 说明扩容正在进行
// 不成立 扩容结束了 扩容结束之后最后退出的线程会把nextTable设置为table
// 条件3： 扩容正在进行中
// 不成立 sizeCtl 是当前扩容阈值 扩容已经结束了
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.compareAndSetInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
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());
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;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
V ev = e.val;
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
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;
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))
setTabAt(tab, i, untreeify(t.first));
}
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (validated) {
if (oldVal != null) {
if (value == null)
addCount(-1L, -1);
return oldVal;
}
break;
}
}
}
return null;
}
