这篇文章主要关注往ConcurrentHashMap放入元素的情况。即put(K key,V value)方法。
ConcurrentHashMap put进一个key,value的简化的过程例如以下:
1.取key的hash值,算出在存放的Segment数组下标。
2.找到segment数组下标后,取出这个Segment。然后计算出须要存放在Segment中HashEntry的数组下标
3.最后将key,value放入
具体步骤如图:
接下去分析源代码。首先是获取要存放元素的segment的源代码,代码例如以下:
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的hash值,hash(key)方法会尽量打散hash,降低hash冲突。
计算出hash后。会使用语句int j = (hash >>> segmentShift) & segmentMask; 来得到segment数组下标。
然后使用UNSAFE.getObject来尝试从segment数组获取segment。
假设为空这时会调用ensureSegment(j)方法创建一个并CAS设置到segment数组其中。
因为相对来讲segment仅仅在第一次不存在的时候才会创建并放入segment数组中,也仅仅有在这一步会发生与其他线程的竞争。
因此对于segment的创建及放入到segment数组其中,ConcurrentHashMap採用的是CAS操作,来原子性的放入segment。这也符合对于少量或
中度并发的情形适合用CAS操作。
有了segment后。调用segment的put方法将元素插入到对应的HashEntry数组其中。
segment的put方法源代码例如以下:
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
这部分因为对于每一个segment的put操作。相对来讲竞争会比較激烈。因此这里会使用lock来进行同步控制。
分下面步骤进行:
1.这里会首先调用ReentrantLock的tryLock方法,看能否够获取到锁。能则进入到第3步。
否则进入第2步。调用scanAndLockForPut获取锁。
2.调用scanAndLockForPut方法尝试获取CAS获取锁。
进入这种方法时先尝试获取一次。假设获取到则立即返回。
否则会进行多次CAS获取锁,
在获取锁的过程中假设要放入的元素在HashEntry数组中对应位置不存在则先创建一个。
源代码例如以下:
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; // negative while locating node
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
能够看到一旦retries次数假设超过MAX_SCAN_RETRIES(这里是64次)则会调用ReentrantLock的lock方法堵塞的获取锁(在lock方法中还是会用CAS获取锁。假设还是不能获取到
则将此线程放入CLH队列,最后堵塞。具体见深入并发AQS二)。
比較有意思的是最后一个
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first)
这部分代码,这里推断假设当前HashEntry要存放位置的首结点,假设有其他线程已经完毕了插入的操作,则会将retries置为-1。
ConcurrentHashMap觉得这样的情况之后会非常快获取到锁。
一直反复CAS获取锁,获取到后返回node。
3.已经获取到锁后,就能够放HashEntry中放入元素了。
这里分两种情况:
一种是当前HashEntry数组对应位置存在这个key元素,这时会将当前HashEntry的value替换成新的value。
一种是当前HashEntry数组对应位置不存在这个key元素。这时则会将元素插入到对应位置。
对于于第一种情况的源代码在上述put方法其中,部分片断例如以下:
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
//这里假设e不为空。则说明当前HashEntry位置已经元素在,这时遍历这个冲突链,看是否当前key已经存在于HashEntry其中
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
//当前存在key元素,这时替换它
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
//这个分支觉得当前位置HashEntry无元素存在或者当前不存在同样key的HashEntry。
//这时假设前面scanAndLockForPut已经返回创建的HashEntry结点,则直接将这个新node结点的next指针指向HashEntry位置的首结点
if (node != null)
node.setNext(first);
else
//node为空则新建一个HashEntry,用于存放key。value
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
能够看到这里有个优化部分是调用
scanAndLockForPut
方法时。在CAS尝试获取锁时。假设当前要存放元素的HashEntry数组位置没有不论什么元素。这时觉得竞争较少。所以会投机地先创建一个node 用于存放key,value。
否则这时因为不确定是否该创建新的结点,因为有可能key值已经存在。这时其实仅仅须要进行更新就可以。当然也有觉得竞争较激烈的因素存在。