（一）基于JDK1.7的ConcurrentHashMap
1.基本实现：
由Segement数组和HashEntry组成，与HashMap相同都是数组+链表的结构，而每一个Segment元素存储的是HashEntry数组+链表，这个和HashMap的数据存储结构一样
Segement:
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
// 真正存放数据的桶
transient volatile HashEntry<K,V>[] table;
transient int count;
transient int modCount;
transient int threshold;//扩容阈值
final float loadFactor;//加载因子
}
hashEntry：
static final class HashEntry<K,V> {
final int hash;
// key值初始化后不能改变
final K key;
//volatile保证读到的数据为最新值
volatile V value;
//volatile保证读到的数据为最新的
volatile HashEntry<K,V> next;
} 关系如图所示：
锁分段技术：HashTable效率低下的原因是多个线程竞争同一把锁，而ConcurrentHashMap把数据分成了一段一段的存储，每一段数据都分给一把锁，当一段数据被一个线程访问时，其他段的数据也能被其他线程所访问。
2.ConcurrentHashMap实现存储和读取：
1.put（）
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
先通过key定位到Segement，之后再具体Segement进行具体的添加：
先进行再散列：减少散列冲突，使元素均匀的分布在不同的Segment上，从而提高容器的存取效率
通过散列算法定位： segments[(hash>>>segmentShift)&segementMask]
1 final V put(K key, int hash, V value, boolean onlyIfAbsent) {
2 HashEntry<K,V> node = tryLock() ? null :
3 scanAndLockForPut(key, hash, value);
4 V oldValue;
5 try {
6 HashEntry<K,V>[] tab = table;
7 int index = (tab.length - 1) & hash;
8 HashEntry<K,V> first = entryAt(tab, index);
9 for (HashEntry<K,V> e = first;;) {
10 if (e != null) {
11 K k;
12 if ((k = e.key) == key ||
13 (e.hash == hash && key.equals(k))) {
14 oldValue = e.value;
15 if (!onlyIfAbsent) {
16 e.value = value;
17 ++modCount;
18 }
19 break;
20 }
21 e = e.next;
22 }
23 else {
24 if (node != null)
25 node.setNext(first);
26 else
27 node = new HashEntry<K,V>(hash, key, value, first);
28 int c = count + 1;
29 if (c > threshold && tab.length < MAXIMUM_CAPACITY)
30 rehash(node);
31 else
32 setEntryAt(tab, index, node);
33 ++modCount;
34 count = c;
35 oldValue = null;
36 break;
37 }
38 }
39 } finally {
40 unlock();
41 }
42 return oldValue;
43 }
首先尝试获取锁，如果获取失败则调用scanAndLockForPut（）自旋获取锁。
scanAndLockForPut
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // 定位节点时为负数
//(1)
while (!tryLock()) {
HashEntry<K,V> f; // 首先在下面重新检查
if (retries < 0) {
if (e == null) {
if (node == null) // 推测性地创建节点
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
//(2)
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // 如果Entry改变则重新遍历
retries = -1;
}
}
return node;
}
代码（1）尝试自旋获取锁。
代码（2）如果重试的次数达到了 MAX_SCAN_RETRIES 则改为阻塞锁获取，保证能获取成功。
流程：
1.首先尝试获取锁，如果获取失败说明与其它线程存在竞争，则调用scanAndLockForPut()
2. 计算HashEntry的数组下标并定位到HashEntry链表的首节点
3.遍历链表，如果不为空则比较key是否相同，相同则覆盖
4.为空创建一个新的结点并添加到hash链的头部
5.判断元素个数是否超过扩容阈值，超过则调用resize扩容
6.最后释放锁
2.get（）
1 public V get(Object key) {
2 Segment<K,V> s; // manually integrate access methods to reduce overhead
3 HashEntry<K,V>[] tab;
4 int h = hash(key);
5 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
6 if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
7 (tab = s.table) != null) {
8 for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
9 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
10 e != null; e = e.next) {
11 K k;
12 if ((k = e.key) == key || (e.hash == h && key.equals(k)))
13 return e.value;
14 }
15 }
16 return null;
17 }
通过key的hash值，调用UNSAFE.getObjectVolatile方法，由于是volatile版本，可以实现线程之间的可见性（遵循happend-before原则），故可以从最新的segments数组中获取该key所在的segment；
然后根据key的hash值，获取该segment内部的哈希表table数组的下标，从而获取该key所在的链表的头结点，然后从链表头结点开始遍历该链表，最终如果没有找到，则返回null或者找到对应的链表节点，返回该链表节点的value。
3.size：
一开始并不对Segment加锁，而是直接尝试将所有的Segment元素中的count相加，这样执行两次，然后将两次的结果对比，如果两次结果相等则直接返回；而如果两次结果不同，则再将所有Segment加锁（将所有segement的put，remove，clean都锁住），然后再执行统计得到对应的size值
（二）基于JDK1.8的ConcurrentHashMap
1.重要概念： 采用Node+CAS+Synchronized来保证并发安全的实现。
2.put方法
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //两次hash，减少hash冲突，可以均匀分布
int binCount = 0;
for (Node<K,V>[] tab = table;;) { //对这个table进行迭代
Node<K,V> f; int n, i, fh;
//这里就是上面构造方法没有进行初始化，在这里进行判断，为null就调用initTable进行初始化，属于懒汉模式初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果i位置没有数据，就直接无锁插入
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;
//如果以上条件都不满足，那就要进行加锁操作，也就是存在hash冲突，锁住链表或者红黑树的头结点
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) { //表示该节点是链表结构
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//这里涉及到相同的key进行put就会覆盖原先的value
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) { //插入链表尾部
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) { //如果链表的长度大于8时就会进行红黑树的转换
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//统计size，并且检查是否需要扩容
return null;
}流程：
1.先对key的hashcode进行第二次hash，减少哈希冲突
2.如果table没有初始化，则调用initTable进行table的初始化
3.如果table的i位置没有数据，则直接进行无锁插入（调用casTabAt）
4.判断是否在进行扩容，如果是则协助扩容
5.以上条件都不满足，则使用synchronized进行加锁操作
6.如果是链表则进行链表的插入，如果是红黑树则进行红黑树的插入操作
7.如果链表的个数是否大于阈值，大于则转换为红黑树
8.最后调用addCount，统计size，检查否需要扩容
initable:
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设为-1
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
/*
unsafe.compareAndSwapInt这个方法，这个方法有四个参数，其中第一个参数为需要改变的对象，第二个为偏移量(即之前求出来的valueOffset的值)，第三个参数为期待的值，第四个为更新后的值
*/
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY; //DEFAULT_CAPACITY为16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2); //扩容阈值为新容量的0.75倍
}
} finally {
sizeCtl = sc; //扩容保护
}
break;
}
}
return tab;
}
table的初始化时未加锁的，当腰初始化的时候调用compareAndSwapInt将sizeCtl视为-1，sizeCtl由votaile修饰，保证了可见性，因而别的线程再调用次方法时会执行Thread.yield（）
addCount
3.扩容实现：
1.扩容时机：
1.调用put后会调用addCount（），内部检查sizeCtl是否需要扩容
2.链表转红黑树时如果table容量小于64
3.调用putAll（）之类一次性加入大量元素
2.总体思想：分工合作，多个线程一个协作扩容，将表拆分，让每个线程处理自己的区间
3.重要属性：
1. sizeCtl：控制标志符
当前未初始化：
=0 未指定初始化容量
>0由指定的初始化容量计算，再寻找最近的2的幂次
初始化中：
-1 table正在初始化
-N 表示正有N-1个线程执行扩容操作
初始化完成：
=table.length*0.75
2.static变量：
// 用于生成每次扩容都唯一的生成戳的数，最小是6。
private static int RESIZE_STAMP_BITS = 16;
// 最大的扩容线程的数量
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 移位量，把生成戳移位后保存在sizeCtl中当做扩容线程计数的基数，相反方向移位后能够反解出生成戳
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
3.
4.
5.transfer方法
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 每核处理的量小于16，则强制赋值16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; //构建一个nextTable对象，其容量为原来容量的两倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
// 连接点指针，用于标志位（fwd的hash值为-1，fwd.nextTable=nextTab）
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 当advance == true时，表明该节点已经处理过了
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 控制 --i ,遍历原hash表中的节点
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// 用CAS计算得到的transferIndex
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; // table 指向nextTable
sizeCtl = (n << 1) - (n >>> 1); // sizeCtl阈值为原来的1.5倍
return; // 跳出死循环，
}
// CAS 更扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作
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
}
}
// 遍历的节点为null，则放入到ForwardingNode 指针节点
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// f.hash == -1 表示遍历到了ForwardingNode节点，意味着该节点已经处理过了
// 这里是控制并发扩容的核心
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
// 节点加锁
synchronized (f) {
// 节点复制工作
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// fh >= 0 ,表示为链表节点
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);
}
// 在nextTable i 位置处插上链表
setTabAt(nextTab, i, ln);
// 在nextTable i + n 位置处插上链表
setTabAt(nextTab, i + n, hn);
// 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
setTabAt(tab, i, fwd);
// advance = true 可以执行--i动作，遍历节点
advance = true;
}
// 如果是TreeBin，则按照红黑树进行处理，处理逻辑与上面一致
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;
}
}
// 扩容后树节点个数若<=6，将树转链表
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;
}
}
}
}
}
}4.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 &&
(e = tabAt(tab, (n - 1) & h)) != null) {//读取首节点的Node元素
if ((eh = e.hash) == h) { //如果该节点就是首节点就返回
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//hash值为负值表示正在扩容，这个时候查的是ForwardingNode的find方法来定位到nextTable来
//查找，查找到就返回
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode，那就往下遍历
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}- 计算hash值，定位到该table索引位置，如果是首节点符合就返回
- 如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回
- 以上都不符合的话，就往下遍历节点，匹配就返回，否则最后就返回null
5.size操作：元素总数 = baseCount + sum(CounterCell)
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a; //变化的数量
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
今天的文章ConcurrentHashMap源码学习笔记jdk1.7&1.8分享到此就结束了，感谢您的阅读。
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
如需转载请保留出处：https://bianchenghao.cn/81816.html
