Repository navigation
jdk collection cocuhashmap
wangkaiwu edited this page May 17, 2023
·
2 revisions
构造函数
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
public ConcurrentHashMap() {
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
private static final int DEFAULT_CAPACITY = 16;
private transient volatile int sizeCtl;
private final Node<K, V>[] initTable() {
Node<K, V>[] tab;
int sc;
while ((tab = table) == null || tab.length == 0) {
// 第一次肯定是 null
if ((sc = sizeCtl) < 0) {
// sizeCtl 为负的情况,一般就是正在扩容,所以当前线程要让出cpu使用权
Thread.yield();
} else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
// 尝试将 sizeCtl 赋值为 -1,变相实现了一个锁
try {
if ((tab = table) == null || tab.length == 0) {
// 默认的 n 是 16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
// nt 长度是 16
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
table = tab = nt;
// 相当于 sc = n * 0.75,也就是容量阈值
sc = n - (n >>> 2);
}
} finally {
// 修改阈值
sizeCtl = sc;
}
break;
}
}
return tab;
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// key 的 hashCode
int hash = spread(key.hashCode());
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) {
// 赋初值
} else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null, new Node<K, V>(hash, key, value))) {
break;
}
}
//............
}
addCount(1L, binCount);
return null;
}
}以上方法设计两个知识点:
- tabAt 用于获取链表的首元素
- casTabAt 用于对链表的首元素赋值
- addCount 增加容器数量
put方法相关知识
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
private static final Unsafe U = Unsafe.getUnsafe();
private static final int ABASE;
private static final int ASHIFT;
static {
// 数组对象头的大小。8(markWord) + 4(类元指针) + 4(数组大小) = 16
ABASE = U.arrayBaseOffset(Node[].class);
int scale = U.arrayIndexScale(Node[].class);
// 如果对象指针是64位(8字节),那么对数组元素寻址的话,需要序号乘以8来计算偏移地址
// 而通过 << 3 位,也同样能实现 * 8 效果,因此
// ASHIFT: 64位是 3,压缩指针后是 2
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
}
// 之所以要用 unsafe,是因为想通过原子的方式获取数组元素的最新值
static final <K, V> Node<K, V> tabAt(Node<K, V>[] tab, int i) {
// getObjectAcquire:传入偏移量和对象,获取对象对应位置的值,并将该值作为对象的指针
return (Node<K, V>) U.getObjectAcquire(tab, ((long) i << ASHIFT) + ABASE);
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
// 与上面的get类似,这里是通过 cas 的方式对数组的某个位置进行赋值
// true: cas 成功,false: 失败
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);
}
}类似于 LongAdder
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
private transient volatile CounterCell[] counterCells;
private final void addCount(long x, int check) {
// 大概就是从 counterCells 中选一个元素,然后+1
}
private final void fullAddCount(long x, boolean wasUncontended) {
// 首次调用 addCount 时,counterCells 的初始长度是 2
CounterCell[] rs = new CounterCell[2];
rs[h & 1] = new CounterCell(x);
counterCells = rs;
init = true;
// 当触发扩容的时候,数组扩充2倍
counterCells = Arrays.copyOf(cs, n << 1);
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
final V putVal(K key, V value, boolean onlyIfAbsent) {
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K, V>[] tab = table; ; ) {
Node<K, V> f;
// fh: first hash
int n, i, fh;
// fk: first key
K fk;
// fv: first value
V fv;
if (tab == null || (n = tab.length) == 0) {
// 赋初值
} else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
} else if ((fh = f.hash) == MOVED) {
} else if (onlyIfAbsent) {
} else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
// hash
if (fh >= 0) {
// 树或者队列的 size
binCount = 1;
for (Node<K, V> e = f; ; ++binCount) {
// ek:element key
K ek;
// hash 肯定不一致
if (e.hash == hash) {
}
Node<K, V> pred = e;
if ((e = e.next) == null) {
// e 作为 f 的 next,肯定是空
pred.next = new Node<K, V>(hash, key, value);
break;
}
}
}
// ...........
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD) {
}
if (oldVal != null) {
}
break;
}
}
}
addCount(1L, binCount);
return null;
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
// sizeCtl 是容量的阈值,但是当发生扩容的时候会变为 -1
private transient volatile int sizeCtl;
private final void addCount(long x, int check) {
long b, s;
// check 表示遍历链表查找空位的次数。如果 = 0,表示本次添加元素不涉及链表的添加(换个说法,没有发生hash碰撞)
if (check >= 0) {
Node<K, V>[] tab, nt;
int n, sc;
while (s >= (long) (sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n) << RESIZE_STAMP_SHIFT;
if (sc < 0) {
// 说明正在扩容
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
s = sumCount();
}
}
}
}public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable {
private static final int MIN_TRANSFER_STRIDE = 16;
// cpu 核心个数
static final int NCPU = Runtime.getRuntime().availableProcessors();
private final void transfer(Node<K, V>[] tab, Node<K, V>[] nextTab) {
int n = tab.length, stride;
// NCPU 是 8,因此 (NCPU > 1) ? (n >>> 3) / NCPU : n 的最终值是 (n >>> 3 / NCPU)
// 如果 n 是16,则 stride 最终值是 1
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
// 1 < 16 因此,最终是 16
stride = MIN_TRANSFER_STRIDE;
}
if (nextTab == null) {
// 扩容 2 倍
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n << 1];
nextTab = nt;
nextTable = nextTab;
// n 是原数组的大小
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K, V> fwd = new ForwardingNode<K, V>(nextTab);
boolean advance = true;
boolean finishing = false;
for (int i = 0, bound = 0; ; ) {
Node<K, V> f;
int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing) {
} else if ((nextIndex = transferIndex) <= 0) {
} else if (U.compareAndSetInt(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
// nextBound = (nextIndex > stride ? nextIndex - stride : 0)
// transferIndex 是默认的上限,每次 -stride
bound = nextBound;
// i 是当前值
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
//........
} else if ((f = tabAt(tab, i)) == null) {
// 相当于插入一个标记,碰到这个标记就说明当前访问的是旧数组
// 需要去新数组访问对应位置
advance = casTabAt(tab, i, null, fwd);
} else if ((fh = f.hash) == MOVED) {
// 已经处理过了
} else {
synchronized (f) {
if (tabAt(tab, i) == f) {
// 后面不看了,就是在尝试把 stride 个链表的元素转移到新数组中
}
}
}
}
}
}- 每次 put 完成后,判断 size 是否超过了阈值(容量的 3/4),然后触发扩容
- 扩容时会产生新数组,且旧数组的元素尝试设置为 MOVED 元素,表明该列正在被扩容。
- 扩容的主要工作就是把旧数组中的元素转移到新数组中,在尚未执行完成前不会用新数组替换旧数组
- 如果扩容尚未完成,且新线程执行了 put 操作,且 put 的位置恰好是 MOVED 则会尝试协助转移。
- 协助转移就是把16个链表的内容转移到新数组
- 如果无需帮忙转移,则直接获取新数组对应位置的元素,并直接进行写入