主要内容：
- ConcurrentHashMap数据结构
- 继承关系、关键属性、构造函数
- 插入元素
ConcurrentHashMap概述(基于jdk1.7)
ConcurrentHashMap是线程安全的集合，与Hashtable相比ConcurrentHashMap效率高。并发环境下Hashtable效率低下的原因是访问Hashtable的多个线程都竞争同一把锁，当然访问慢了。那我们很容易想到如果有多把锁，每把锁作用于集合的一部分数据，那么访问多线程访问不同数据段的数据时就不会发生锁竞争的情况，那么效率当然就提高了。这其实就是ConcurrentHashMap的锁分离技术。
锁分离技术：将数据分成一段一段的存储，每一段数据都有一把锁，当一个线程访问某段数据时，其他段数据可以被其他线程锁访问。
ConcurrentHashMap数据结构
ConcurrentHashMap数据结构.png
底层基于分段的数组和链表实现，由
Segment段类型的数组结构和HashEntry类型的数据结构存储数据，将Segement数组声明为final。
- Segment：继承了ReentrantLock，可重入锁。Segment中包含了HashEntry类型的数组。
static final class Segment<K,V> extends ReentrantLock
implements Serializable {
transient volatile HashEntry<K,V>[] table;//HashEntry类型的数组
transient int count;//Segment中保存的元素数量
transient int modCount;//段修改的次数
transient int threshold;//阈值，超过阈值后数组会进行扩容
final float loadFactor;//负载因子
}
-
HashEntry：链表上的节点，定义了键key、值value、哈希值hash以及指向下一个节点的指针next。这里注意到哈希值、键值是
final型的，保证读操作时不需要加锁。值、下一节点指针是volatile型，保证读操作时可以读到最新的值。
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
final void setNext(HashEntry<K,V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
// Unsafe mechanics
static final sun.misc.Unsafe UNSAFE;
static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = HashEntry.class;
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
源码分析
继承关系
Concurrent继承关系
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {
- 继承AbstractMap抽象类，实现Map接口
- 实现java.io.Serialization接口，支持序列化
- 实现ConcurrentMap接口
初始化
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;
//ssize为大于等于concurrencyLevel的最小的2的幂次方
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
/**
* 用于定位Segment，假设ssize为2的n次方，我们可以发现sshift就是n。
* segmentShift为32-n，segmentMask为2^n-1，二进制表示时各位数都是1
*/
this.segmentShift = 32 - sshift;//偏移量，定位参与hash运算的位数
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;//segment中HashEntry数组的长度
while (cap < c)
cap <<= 1;
//创建segments和segments[0]
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
//使用指定初始化容量、负载因子和默认并发等级16，创建空ConcurrentHashMap
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL);
}
//使用指定初始化容量、默认负载因子0.75和默认并发等级16，创建空ConcurrentHashMap
public ConcurrentHashMap(int initialCapacity) {
this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
//使用默认初始化容量16、负载因子和默认并发等级16，创建空ConcurrentHashMap
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1,
DEFAULT_INITIAL_CAPACITY),
DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
putAll(m);
}
发现构造函数最终调用的都是public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)这个方法，参数分别是DEFAULT_INITIAL_CAPACITY初始容量、DEFAULT_LOAD_FACTOR默认负载因子以及DEFAULT_CONCURRENCY_LEVEL默认并发级别，表示将ConcurrentHashMap分为几个段。
插入
ConcurrentHashMap用分段Segment来保存数据，所以插入和获取数据时都需要先定位到对应的段上。
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)//value值不能为空
throw new NullPointerException();
int hash = hash(key);//求出key的hash值
/**
* segmentShift是32-sshift，将计算的hash值无符号右移segmentShift位，其实就是取hash值的高sshift位
* 然后与segmentMask进行按位与操作
**/
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null) //找不到Segment新建一个
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
定位到对应的段上后就需要调用segment的put方法。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
/**
* tryLock方法尝试去获得锁
* 获取锁失败的话，调用scanAndLockForPut获得锁
**/
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;//当前段的HashEntry类型的数组
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);//定位到对应的桶
for (HashEntry<K,V> e = first;;) {//遍历HashEntry链表
if (e != null) {
K k;
/**
* 要插入的键值对已经存在于HashEntry数组中
* onlyIfAbsent为true，表示只能插入不存在的键值对直接退出
* onlyIfAbsent为false，用新值替代旧值
**/
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 {//要插入的键值对不存在HashEntry数组中
if (node != null)//在scanAndLockForPut获得锁的过程中新建了对应的节点
node.setNext(first);//将first节点设置成node的下一个节点
else
node = new HashEntry<K,V>(hash, key, value, first);//新建HashEntry链表
/**
* 如果插入键值对后，超过阈值，则进行扩容；
* 否则直接插入键值对
**/
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;
}
put方法里首先需要先获得Segment的互斥锁，如果没有获得锁的话会调用scanAndLockForPut这个自旋方法，介绍下这个方法具体做了什么。
//
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);//根据hash值获取对应的HashEntry链表
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; //自旋计数器
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;
}
未完。。。。
网友评论