ConcurrentHashMap
数据结构
ConcurrentHashMap的数据结构与HashMap基本类似， 区别在于：
1、内部在数据写入时加了同步机制(分段锁)保证线程安全，读操作是无锁操作；
2、扩容时老数据的转移是并发执行的，这样扩容的效率更高
ConcurrentHashMap 线程安全的具体实现方式
JDK 1.7
ConcurrentHashMap基于ReentrantLock实现分段锁
将数据分为一段一段的存储，然后给每一段数据配一把锁，底层数据结构：Segment 数组 + HashEntry 数组 ，Segment 实现了 ReentrantLock,所以 Segment 是一种可重入锁，扮演锁的角色
源码分析：
构造函数：
/**
* The default initial capacity for this table,
* used when not otherwise specified in a constructor.
*/
static final int DEFAULT_INITIAL_CAPACITY = 16;
/**
* The default load factor for this table, used when not
* otherwise specified in a constructor.
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/**
* The default concurrency level for this table, used when not
* otherwise specified in a constructor.
*/
static final int DEFAULT_CONCURRENCY_LEVEL = 16;//并发级别，segment数组的大小
//segment的可以包含数据最大的个数
static final int MAX_SEGMENTS = 1 << 16; // slightly conservative
//容量最大值
static final int MAXIMUM_CAPACITY = 1 << 30;
//segment里面最小的容量为2
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
注意：
DEFAULT_INITIAL_CAPACITY与DEFAULT_CONCURRENCY_LEVEL结合可以计算出每个segment里有多少个数。个数大致计算等于DEFAULT_INITIAL_CAPACITY/DEFAULT_CONCURRENCY_LEVEL
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
/**
* Creates a new, empty map with the specified initial
* capacity, load factor and concurrency level.
*/
@SuppressWarnings("unchecked")
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//检验参数是否合法
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching arguments
int sshift = 0;//记录ssize右移的位数，也就是segment[]容量是2的几次幂
//表示segment数组的大小
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;//可以保证segment数组的大小是2幂次方的倍数
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
//判断
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;//<< 后将值赋值给cap
// create segments and segments[0]
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);//cap * loadFactor阈值用于向segment数组进行判断。保证segment长度不会超出
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize]
//将新生成的s0对象放到segment[0]
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
注意：segment的扩容只与segment有关，当一个元素放入segment中时回去判断当前segment对象内部的阈值是多少，是否超过阈值。如果超过就进行扩容，扩容是扩容segment数组。扩容先创建一个扩容后的数组再将数据移入新数组再跟换segment对象。segment数组的长度是不变的。
put方法：
public V put(K key, V value) {
Segment<K,V> s;
//value不可以为空
if (value == null)
throw new NullPointerException();
//取hash值
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 取第j个位置的元素
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
/**
* The per-segment table.
*/
transient volatile HashEntry<K,V>[] table;
/**
* The number of elements.
*/
transient int count;
/**
* The total number of mutative operations in this segment.
*/
transient int modCount;
/**
* The table is rehashed when its size exceeds（超出） this threshold.
*/
transient int threshold;
/**
* The load factor（因子） for the hash table. Even though this value
* is same for all segments
*/
final float loadFactor;
/**
* The maximum number of times to tryLock in a prescan before
* possibly blocking on acquire in preparation for a locked
* segment operation. On multiprocessors, using a bounded
* number of retries maintains cache acquired while locating
* nodes.
*/
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
说明：Runtime.getRuntime().availableProcessors() 返回Java虚拟机可用的处理器数量。
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
//这个放法是写在segment内置对象里面的
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//尝试加锁（segment继承了ReentrantLock），如果获取不到锁调用scanAndLockForPut来获取锁
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;//取得segment数组内部的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;
//查看key是否相等
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
//记录value值
oldValue = e.value;
// 如果标记不存在的时候，才进行插入操作
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
//创建一个新的hashEntry对象
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;//count代表segment数组当前位置存储的元素个数
//检查是否满足扩容的条件
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
//将表的大小加倍并重新打包条目，同时将给定的节点添加到新表中
rehash(node);//扩容
else
//把数组放到segment数组内部的table的第i个位置
setEntryAt(tab, index, node);//保证是内存里的值，而不是线程里面的值
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
/**
* Scans for a node containing given key while trying to
* acquire lock, creating and returning one if not found。
*/
//当put方法尝试获取锁失败后，调用该方法获取锁
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
//获取hash对应
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
//根据实际情况来new hashEntry对象
if (retries < 0) {
//first为空或者遍历到了尾节点k
if (e == null) {
if (node == null) // speculatively create node
//创建hashEntry对象
node = new HashEntry<K,V>(hash, key, value, null);
// 将retries设置为0 ， 下次循环时就不进入这个分支（if (retries < 0)）
retries = 0;
}
// 遍历到的当前节点中存在相等的key
else if (key.equals(e.key))
// 将retries设置为0 ， 下次循环时就不进入这个分支（if (retries < 0)）
retries = 0;
//first不为空并且key不相等，继续遍历下一个节点
else
e = e.next;
}
// 如果retries > 64(处理器数量大于1)或1（处理器数量小于1）
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
//(retries & 1) == 0 retries为偶数的时候成立
else if ((retries & 1) == 0 &&
//判断节点是否发生了变化
(f = entryForHash(this, hash)) != first) {
//如果条目改变，重新遍历
e = first = f; // re-traverse if entry changed
//重新赋值为-1
retries = -1;
}
}
return node;
}
}
/**
* Gets the table entry for the given segment and hash
*/
@SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V> entryForHash(Segment<K,V> seg, int h) {
HashEntry<K,V>[] tab;
return (seg == null || (tab = seg.table) == null) ? null :
(HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
}
//使用volatile读语义获取给定表的第i个元素(如果非null)。
@SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V> entryAt(HashEntry<K,V>[] tab, int i) {
return (tab == null) ? null :
(HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)i << TSHIFT) + TBASE);
}
注意：
- ? put方法将键/值对添加到 hashMap 中
- tryLock():先判断当前的这把锁能不能获取到，如果能获取就获取锁并立马返回true，如果不能获取到立即返回false.(不阻塞)
- Lock():如果能获取到立即获取到，如果不能获取就阻塞知道获取到锁。
- 当调用tryLocak方法尝试获取锁失败调用scanAndLockForPut获取锁的原因：主要原因是锁的选择，我们可以看到scanAndLockForPut方法中获取锁的方式是
while (!tryLock()) { //dosomething},用这种方式相较于lock（）获取锁，用lock()获取锁会阻塞，但是更不占用cpu。用while (!tryLock()) { //dosomething}获取锁不阻塞，并且更占cpu但是再获取锁的可以做其他的事情。可以做的事情是，new HashEntry()对象，之后可以切换成lock
putIfAbsent方法：
@SuppressWarnings("unchecked")
public V putIfAbsent(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
(segments, (j << SSHIFT) + SBASE)) == null)
s = ensureSegment(j);
return s.put(key, hash, value, true);
}
注意：如果 hashMap 中不存在指定的键，则将指定的键/值对插入到 hashMap 中。
ensureSegment方法
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
//获取segment对象数组
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
//看segment数组的第u个位置是不是空，不为空说明被另外一个线程生成出来了，那么直接返回就行
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
Segment<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;//ss[0]数组的长度
float lf = proto.loadFactor; //table进行扩容的一个阈值，定义map中有多少元素时，map快满了。map size大于这个阈值，则有可能会对table进行扩容
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
//再次进行检查，当前位置是否为空
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);//创建对象
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
//尝试将生成的对象放到数组的第u个位置，如果成功就返回
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
? 注意：当有多个线程需要向同一个位置生成一个segment对象，当有前面的对象生成了segment对象时，之后的线程只需要使用前面生成的对象就可。
rehash（扩容）方法
/**
* The table is rehashed when its size exceeds this threshold.
* (The value of this field is always <tt>(int)(capacity *
* loadFactor)</tt>.)
*/
transient int threshold;//阈值
/**
* The load factor for the hash table. Even though this value
* is same for all segments, it is replicated to avoid needing
* links to outer object.
* @serial
*/
final float loadFactor;//加载因子
/**
* Doubles size of table and repacks entries, also adding the
* given node to new table
*/
@SuppressWarnings("unchecked")
private void rehash(HashEntry<K,V> node) {
/*
* Reclassify nodes in each list to new table. Because we
* are using power-of-two expansion, the elements from
* each bin must either stay at same index, or move with a
* power of two offset. We eliminate unnecessary node
* creation by catching cases where old nodes can be
* reused because their next fields won't change.
* Statistically, at the default threshold, only about
* one-sixth of them need cloning when a table
* doubles. The nodes they replace will be garbage
* collectable as soon as they are no longer referenced by
* any reader thread that may be in the midst of
* concurrently traversing table. Entry accesses use plain
* array indexing because they are followed by volatile
* table write.
*/
HashEntry<K,V>[] oldTable = table;//
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;//翻倍
threshold = (int)(newCapacity * loadFactor);//计算阈值
//创建一个新的newTable
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
//为了取余 （ hash & (sizeMask-1) ）
int sizeMask = newCapacity - 1;
//遍历旧数组
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
//当前元素不为空
if (e != null) {
HashEntry<K,V> next = e.next;
//当前元素的hash值（新数组的位置）
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
//将当前节点当前节点放到新数组
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
//当前元素
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
//遍历当前的列表
//最后达到的效果：列表最后连续的需要移动到idx位置的一个或多个元素 放到了新数组的对应位置
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
//计算元素的新的位置
int k = last.hash & sizeMask;
// k
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
// 直接将列表最后连续的需要移动到idx位置的一个或多个元素的头部元素地址复制给对应位置的table
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
//记录新table中对应下标的元素
HashEntry<K,V> n = newTable[k];
//头插法，插入数据
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
//构造函数
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
get方法
/**
* Returns the value to which the specified key is mapped,
* or {@code null} if this map contains no mapping for the key.
*/
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
//根据key来计算hash
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
remove方法
/**
* Removes the key (and its corresponding value) from this map.
* This method does nothing if the key is not in the map.
*/
//移除成功返回移除的旧值
public V remove(Object key) {
int hash = hash(key);
//Get the segment for the given hash
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.remove(key, hash, null);
}
/**
* Remove; match on key only if value null, else match both.
*/
final V remove(Object key, int hash, Object value) {
//尝试加锁
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
//Gets the ith element of given table (if nonnull) with volatile read semantics.
HashEntry<K,V> e = entryAt(tab, index);
HashEntry<K,V> pred = null;
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)
setEntryAt(tab, index, next);
else
pred.setNext(next);
++modCount;
--count;
oldValue = v;
}
break;
}
pred = e;
e = next;
}
} finally {
unlock();
}
return oldValue;
}
/**
* Scans for a node containing the given key while trying to
* acquire lock for a remove or replace operation. Upon
* return, guarantees that lock is held. Note that we must
* lock even if the key is not found, to ensure sequential
* consistency of updates.
*/
private void scanAndLock(Object key, int hash) {
// similar to but simpler than scanAndLockForPut
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
int retries = -1;
while (!tryLock()) {
HashEntry<K,V> f;
if (retries < 0) {
if (e == null || 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;
retries = -1;
}
}
}
