ConcurrentHashMap为什么线程安全
2024-01-07 19:03:10
文章目录
本文假设大家已经对HashMap的底层原理有所了解，接下来将在HashMap的基础上解答为什么ConcurrentHashMap是线程安全的。
还是老样子，我们将从几个核心问题出发，解答ConcurrentHashMap为何是线程安全的原因。
- 初始化数组时的线程安全
- put操作的线程安全
- 扩容操作的线程安全
- 统计容器大小的线程安全
- get线程安全
一、必要知识
1. 成员属性
* ---------------- Fields -------------- */
/**
* The array of bins. Lazily initialized upon first insertion.
* Size is always a power of two. Accessed directly by iterators.
*/
*这里要注意一下，volatile修饰了table，这样的好处是当某一个线程扩容时，另一个线程可见，
*虽然volatile对于不能保证引用类型的变量内部可见，但是这里时扩容，换了一个数组地址发生了改变，所以是可见的
transient volatile Node<K,V>[] table;
/**
* The next table to use; non-null only while resizing.
*/
*当扩容的时候使用nextTable
private transient volatile Node<K,V>[] nextTable;
/**
* Base counter value, used mainly when there is no contention,
* but also as a fallback during table initialization
* races. Updated via CAS.
*/
* 用于统计容量
private transient volatile long baseCount;
/**
* 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.
*/
*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义
*当为负数时：-1代表正在初始化，-N代表有N-1个线程正在 进行扩容
*当为0时：代表当时的table还没有被初始化
*当为正数时：当初始化或扩容完成后，为 下一次的扩容的阈值大小
private transient volatile int sizeCtl;
/**
* The next table index (plus one) to split while resizing.
*/
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;
// views
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
2. Node存储结构
static class Node<K,V> implements Map.Entry<K,V> {
//key的hash值
final int hash;
final K key;
//val和next都会在扩容时发生变化，所以加上volatile来保持可见性和禁止重排序，
//保证了读的可见性，所以读的方法，并没有加锁。
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;
}
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 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;
}
}
比较重要的点就是在它的成员属性加了volatile关键字，保证了可见性和有序性。
3. TreeNode
树化节点
static final class TreeNode<K,V> extends Node<K,V> {
TreeNode<K,V> parent; // red-black tree links
TreeNode<K,V> left;
TreeNode<K,V> right;
TreeNode<K,V> prev; // needed to unlink next upon deletion
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);
}
......
4. TreeBin
static final class TreeBin<K,V> extends Node<K,V> {
//指向TreeNode列表和根节点
TreeNode<K,V> root;
volatile TreeNode<K,V> first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
//读写锁状态
static final int WRITER = 1; // set while holding write lock 获取写锁的状态
static final int WAITER = 2; // set when waiting for write lock 等待写锁的状态
static final int READER = 4; // increment value for setting read lock 增加数据时读锁的状态
/**
* Tie-breaking utility for ordering insertions when equal
* hashCodes and non-comparable. We don't require a total
* order, just a consistent insertion rule to maintain
* equivalence across rebalancings. Tie-breaking further than
* necessary simplifies testing a bit.
*/
static int tieBreakOrder(Object a, Object b) {
int d;
if (a == null || b == null ||
(d = a.getClass().getName().
compareTo(b.getClass().getName())) == 0)
d = (System.identityHashCode(a) <= System.identityHashCode(b) ?
-1 : 1);
return d;
}
/**
* Creates bin with initial set of nodes headed by b.
*/
TreeBin(TreeNode<K,V> b) {
super(TREEBIN, null, null, null);
this.first = b;
………………………………
}
TreeBin从字面含义中可以理解为存储树形结构的容器，而树形结构就是指TreeNode，所以TreeBin就是封装TreeNode的容器，它提供转换黑红树的一些条件和锁的控制。
二、源码解析
1. 初始化数组时的线程安全
entry数组的初始化并不是在构造函数里面，而是在第一次调用put方法的时候检测是否初始化了，如果没有那么进行初始化。（这里和hashmap是一样的）
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
//【每次循环都获取最新的Node数组引用】
while ((tab = table) == null || tab.length == 0) {
//sizeCtl是一个标记位，若为-1也就是小于0，代表有线程在进行初始化工作了
if ((sc = sizeCtl) < 0)
//【让出CPU时间片】
Thread.yield(); // lost initialization race; just spin
//【CAS操作，将本实例的sizeCtl变量设置为-1】
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
//如果CAS操作成功了，代表本线程将负责初始化工作
try {
//再检查一遍数组是否为空
if ((tab = table) == null || tab.length == 0) {
//在初始化Map时，sizeCtl代表数组大小，默认16
//所以此时n默认为16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//Node数组
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
//将其赋值给table变量
table = tab = nt;
//通过位运算，n减去n二进制右移2位，相当于乘以0.75
//例如16经过运算为12，与乘0.75一样，只不过位运算更快
sc = n - (n >>> 2);
}
} finally {
//将计算后的sc（12）直接赋值给sizeCtl，表示达到12长度就扩容
//由于这里只会有一个线程在执行，直接赋值即可，没有线程安全问题
//只需要保证可见性
sizeCtl = sc;
}
break;
}
}
return tab;
}
使用cas自旋+标记位（标记位必然是volatile的） 来确保只有一个线程能够初始化。
2. put操作的线程安全
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//对key的hashCode进行散列
int hash = spread(key.hashCode());
int binCount = 0;
//【一个无限循环，直到put操作完成后退出循环】
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//当Node数组为空时进行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//Unsafe类volatile的方式取出hashCode散列后通过与运算得出的Node数组下标值对应的Node对象
//【此时的Node对象若为空，则代表还未有线程对此Node进行插入操作】
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)
//【帮助扩容】
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//对Node对象进行加锁，插入到当前结点的拉链后面
synchronized (f) {
//二次确认此Node对象还是原来的那一个
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
//【无限循环，直到完成put】
for (Node<K,V> e = f;; ++binCount) {
K ek;
//和HashMap一样，先比较hash，再比较equals
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) {
//和链表头Node节点不冲突，就将其初始化为新Node作为上一个Node节点的next
//形成链表结构
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
...
}
如果当前数组中Node结点为空，那么cas尝试加入，成功则结束，如果不成功，再一次循环。如果正在扩容，那么帮助扩容，整个map扩容完成后才能够put。在进行put的时候会把当前的Node结点锁住，也就是把这个桶给锁了，然后再put保证安全插入。
3. 扩容操作的线程安全
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//根据机器CPU核心数来计算，一条线程负责Node数组中多长的迁移量
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
//本线程分到的迁移量
//假设为16（默认也为16）
stride = MIN_TRANSFER_STRIDE; // subdivide range
//nextTab若为空代表线程是第一个进行迁移的
//初始化迁移后的新Node数组
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//这里n为旧数组长度，左移一位相当于乘以2
//例如原数组长度16，新数组长度则为32
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变量为新数组
nextTable = nextTab;
//假设为16
transferIndex = n;
}
//假设为32
int nextn = nextTab.length;
//【标示Node对象，此对象的hash变量为-1】
//【在get或者put时若遇到此Node，则可以知道当前Map正在迁移】
//【传入nextTab对象】
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;
//i为当前正在处理的Node数组下标，每次处理一个Node节点就会自减1
if (--i >= bound || finishing)
advance = false;
//假设nextIndex=16
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//由以上假设，nextBound就为0
//且将nextIndex设置为0
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//bound=0
bound = nextBound;
//i=16-1=15
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
}
}
//此时i=15，取出Node数组下标为15的那个Node，若为空则不需要迁移
//直接设置占位标示，代表此Node已处理完成
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//【检测此Node的hash是否为MOVED，MOVED是一个常量-1，也就是上面说的占位Node的hash】
//【如果是占位Node，证明此节点已经处理过了，跳过i=15的处理，继续循环】
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//【锁住这个Node，开始迁移这个桶】
synchronized (f) {
//确认Node是原先的Node
if (tabAt(tab, i) == f) {
//ln为lowNode，低位Node，hn为highNode，高位Node
//这两个概念下面以图来说明
Node<K,V> ln, hn;
if (fh >= 0) {
//此时fh与原来Node数组长度进行与运算
//如果高X位为0，此时runBit=0
//如果高X位为1，此时runBit=1
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
//这里的Node，都是同一Node链表中的Node对象
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//正如上面所说，runBit=0，表示此Node为低位Node
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
//Node为高位Node
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;
//若hash和n与运算为0，证明为低位Node，原理同上
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
//这里将高位Node与地位Node都各自组成了两个链表
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//【将低位Node设置到新Node数组中，下标为原来的位置】
setTabAt(nextTab, i, ln);
//【将高位Node设置到新Node数组中，下标为原来的位置加上原Node数组长度】
setTabAt(nextTab, i + n, hn);
//【将此Node设置为占位Node，代表处理完成，这里最后才会把原table的Node结点置为fwd，也就是说，如果get的时候发现时fwd，那么说明此node已经迁移完了，那么也就可以通过fwd找到新的数组中的位置。】
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;
}
}
//【链表放入TreeBin会转为红黑树】
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;
}
}
}
}
}
}
第一点说明：区分ln（低位Node）、hn（高位Node）
举个例：如果一开始容量是16，那么8和24在一个桶里面，那么如果要是扩容成32，那么8和24就在不同的桶里面了，因为每次只会扩容为原来的两倍，所以原来桶里面的元素只会在扩容后的固定两个桶里面。所以可以分成两个桶直接插入到新的map里面。
第二点说明：红黑树桶仍然会更具高低位转为两个拉链，如果两个拉链超过了最小树化的长度，那么会放入到TreeBin中，TreeBin负责把拉链转为红黑树。
4. 统计容器大小的线程安全
整体使用的是高并发累加器的思想（如果不了解，可以在我之前的博客里找到）。
//计数，并检查长度是否达到阈值
private final void addCount(long x, int check) {
//【计数桶】
CounterCell[] as; long b, s;
//【如果counterCells不为null，则代表已经初始化了，直接进入if语句块】
//【若竞争不严重，counterCells有可能还未初始化，为null，先尝试CAS操作递增baseCount值】
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
//进入此语句块有两种可能
//1.counterCells被初始化完成了，不为null
//2.CAS操作递增baseCount值失败了，说明有竞争
CounterCell a; long v; int m;
//标志是否存在竞争
boolean uncontended = true;
//1.先判断计数桶是否还没初始化，则as=null，进入语句块
//2.判断计数桶长度是否为空或，若是进入语句块
//3.这里做了一个线程变量随机数，与上桶大小-1，若桶的这个位置为空，进入语句块
//4.到这里说明桶已经初始化了，且随机的这个位置不为空，尝试CAS操作使桶加1，失败进入语句块
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
//统计容器大小
s = sumCount();
}
...
}
//出现竞争，导致CAS失败
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
//计数桶初始化好了，一定是走这个if语句块
if ((as = counterCells) != null && (n = as.length) > 0) {
//从计数桶数组随机选一个计数桶，若为null表示该桶位还没线程递增过
if ((a = as[(n - 1) & h]) == null) {
//查看计数桶busy状态是否被标识
if (cellsBusy == 0) { // Try to attach new Cell
//若不busy，直接new一个计数桶
CounterCell r = new CounterCell(x); // Optimistic create
//CAS操作，标示计数桶busy中
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
//在锁下再检查一次计数桶为null
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
//将刚刚创建的计数桶赋值给对应位置
rs[j] = r;
created = true;
}
} finally {
//标示不busy了
cellsBusy = 0;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
//走到这里代表计数桶不为null，尝试递增计数桶
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
break;
else if (counterCells != as || n >= NCPU)
collide = false; // At max size or stale
//若CAS操作失败了，到了这里，会先进入一次，然后再走一次刚刚的for循环
//若是第二次for循环，collide=true，则不会走进去
else if (!collide)
collide = true;
//计数桶扩容，一个线程若走了两次for循环，也就是进行了多次CAS操作递增计数桶失败了
//则进行计数桶扩容，CAS标示计数桶busy中
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try {
//确认计数桶还是同一个
if (counterCells == as) {// Expand table unless stale
//将长度扩大到2倍
CounterCell[] rs = new CounterCell[n << 1];
//遍历旧计数桶，将引用直接搬过来
for (int i = 0; i < n; ++i)
rs[i] = as[i];
//赋值
counterCells = rs;
}
} finally {
//取消busy状态
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h);
}
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
//初始化计数桶…
//若有线程同时初始化计数桶，由于CAS操作只有一个线程进入这里
boolean init = false;
try { // Initialize table
//再次确认计数桶为空
if (counterCells == as) {
//【初始化一个长度为2的计数桶】
CounterCell[] rs = new CounterCell[2];
//h为一个随机数，与上1则代表结果为0、1中随机的一个
//也就是在0、1下标中随便选一个计数桶，x=1，放入1的值代表增加1个容量
rs[h & 1] = new CounterCell(x);
//将初始化好的计数桶赋值给ConcurrentHashMap
counterCells = rs;
init = true;
}
} finally {
//最后将busy标识设置为0，表示不busy了
cellsBusy = 0;
}
if (init)
break;
}
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
break; // Fall back on using base
}
}
计数桶
计数桶初始长度为2，在竞争大的时候不够用，就会对计数桶进行扩容，两个问题：
- 什么条件下会进行计数桶扩容
在CAS操作递增计数桶失败了3次之后，会进行扩容计数桶操作，注意此时同时进行了两次随机定位计数桶来进行CAS递增的，所以此时可以保证大概率是因为计数桶不够用了，才会进行计数桶扩容 - 扩容操作是怎么样的
计数桶长度增加到两倍长度，数据直接遍历迁移过来，由于计数桶不像HashMap数据结构那么复杂，有hash算法的影响，加上计数桶只是存放一个long类型的计数值而已，所以直接赋值引用即可。
5. get线程安全
所有的get操作都是无锁的，所以说这些数据需要是volatile的。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
//此过程与HashMap的get操作无异，不多赘述
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;
}
//【当hash<0,有可能是在迁移,使用fwd占位Node去查找新table中的数据】
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;
}
eh<0的情况详细分析，同时也分析在写的时候读的情况。
这里eh只会等于-1，map在进行扩容，最终到新的table上面去寻找。如果到新table上发现时链表，那么直接可以线性查找，如果是一个TreeBin的话，也就是说这个桶还在转换，由链表转为红黑树，由于红黑树结点有一个next属性，也就是说，在没有构造完成红黑树之前，还是会进行线性查找（TreeBin的find方法），如果构造完成了的话，那么就不是TreeBin，而是TreeNode。
文章来源:https://blog.csdn.net/flyingmylife/article/details/135428992
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！