JDK1.7中的ConcurrentHashMap
Posted 愉悦滴帮主)
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JDK1.7中的ConcurrentHashMap相关的知识，希望对你有一定的参考价值。
目录
JDK1.7中的ConcurrentHashMap
前言：
1. HashMap在高并发的环境下，执行put操作会导致HashMap的Entry链表形成环形数据结构，从而导致Entry的next节点始终不为空，因此产生死循环获取Entry
2. HashTable虽然是线程安全的，但是效率低下，当一个线程访问HashTable的同步方法时，其他线程如果也访问HashTable的同步方法，那么会进入阻塞或者轮训状态。
通过了解HashMap我们知道HashMap时线程不安全的，但是HashTable虽然采用 public synchronized put(key,value)的上锁形式来保证线程安全，显然这样做的效率是非常慢的。假设有两个线程，线程1和线程2，线程1想在table[1]的位置put一个值，线程2想在table[3]的位置同样也put一个值，两个线程同时put显然不会造成脏数据，但是hashTable只能进来一个线程，也就是线程1执行完毕后再去执行线程2，这样效率太慢。我们想要分区分段的对数组进行加锁那么就需要ConcurrentHashMap。
1.ConcurrentHashMap的结构
对比hashmap,我们知道hashmap数组对象类型是Entry对象类型，而ConcurrentHashMap数组对象类型是Segment[]数组，segment[]数组的对象类型为HashEntry类型(一个Segment里面包含一个HashEntry数组，每个HashEntry是一个链表结构，当对HashEntry数组的数据进行修改时，必须首先获得与它对应的Segment锁！)。形如：
Segment [0] |
Segment[1] |
ConcurrentHashMap它采锁分段技术 来保证高效的并发操作！
ConcurrentHashMap把容器分为多个 segment（片段） ，每个片段有一把锁，当多线程访问容器里不同数据段的数据时，线程间就不会存在竞争关系；一个线程占用锁访问一个segment的数据时，并不影响另外的线程访问其他segment中的数据
2.ConcurrentHashMap的常量
// 默认初始容量
static final int DEFAULT_INITIAL_CAPACITY = 16;
// 默认加载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;
// 默认segment层级
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
// 最大容量
static final int MAXIMUM_CAPACITY = 1 << 30;
// segment最小容量
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
// 一个segment最大容量
static final int MAX_SEGMENTS = 1 << 16;
// 锁之前重试次数
static final int RETRIES_BEFORE_LOCK = 2;
3.ConcurrentHashMap的构造函数
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
public ConcurrentHashMap(int initialCapacity) {
this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL);
}
再用构造方法的时候我们要先了解几个重要的参数：DEFAULT_INITIAL_CAPACITY 初始容量 默认值16 , DEFAULT_LOAD_FACTOR 加载因子, DEFAULT_CONCURRENCY_LEVEL 并发级别 默认值16
构造方法工作原理：
1.首先对参数进行校验，然后通过while循环找到大于等于2的幂次方数作为ssize大小也就是segment数组长度，默认为16。
2.通过int c = initialCapacity / ssize得到常量c，默认为1。然后判断c * ssize < initialCapacity如果是则ssize++。然后判断cap < c如果是则 cap <<= 1。保证cap是2的幂次方数。
3.然后创建segment[0]位置不为null的segment数组。目的是为了进行segment.put操作时，需要重新计算segment对应hashEntry数组的大小，我们需要一个原型作为参考去掉多余计算步骤。
@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;
// segment数组的长度是由concurrentLevel计算来的，segment数组的长度是2的N次方，
// 默认concurrencyLevel = 16, 所以ssize在默认情况下也是16,此时 sshift = 4
// sshift相当于ssize从1向左移的次数
int ssize = 1; //表示segment数组的大小
//找到大于等于并发级别的2的幂次方数
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// 段偏移量，默认值情况下此时segmentShift = 28
this.segmentShift = 32 - sshift;
// 散列算法的掩码，默认值情况下segmentMask = 15
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
//MIN_SEGMENT_TABLE_CAPACITY默认为2
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// create segments and segments[0]
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
// 创建ssize长度的Segment数组
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}如果我们不传参数去创建ConcurrentHashMap则结构如下：
4.ConcurrentHashMap的put方法
说明：
1.首先校验value为不为空，为空则抛出空指针异常。
2.调用hash()方法算出key的hashcode,ConcurrentHashMap的hash()方法与HashMap的方法是一样的，这里就不做详细说明。
3.通过(hash >>> segmentShift) & segmentMask算出segment数组的索引位置。segmentMask在构造方法中（ this.segmentMask = ssize - 1）已经算好了，默认值为15。segmentShift 在构造方法中（this.segmentShift = 32 - sshift）已经算好了，值为28。
Class sc = Segment[].class;
SBASE = UNSAFE.arrayBaseOFFset(sc);
ss = UNSAFE.arrayIndexScale(sc);
sshift = 31-Integer.numberOfLeadingZeros(ss);
Integer.numberOfLeadingZeros(ss)返回32位二进制数bit位为1之前的bit位数
System.out.println(Integer.numberOfLeadingZeros(1)); //0000 0000 0000 0000 0000 0000 0000 0001 31
System.out.println(Integer.numberOfLeadingZeros(16)); //0000 0000 0000 0000 0000 0000 0001 0000 27
System.out.println(Integer.numberOfLeadingZeros(32)); //0000 0000 0000 0000 0000 0000 0010 0000 26
4. UNSAFE.getObject(segments, (j << SSHIFT) + SBASE))这行代码的意思就是取segment数组第j个位置元素的对象。与下面UNSAFE的简单使用中案例一UNSAFE.getObject(person.table,base+1*ns)这行代码的意思是一样的。
5.调用ensureSegment()生成segment对象并插入到segment【i】位置处。ensureSegment用于确定指定的Segment是否存在，不存在则会创建。注意这样只是乐观锁的一种写法并不能保证绝对的线程安全。源码如下：
/ * *
*扫描包含给定键的节点，同时尝试
*获取锁，如果没有找到，创建并返回一个。在
* return，保证锁被持有。不同于大多数方法:因为
*循环速度无所谓，我们不妨让循环慢一些。
如果key没有找到，返回一个新的节点，否则为null
* /
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
//判断数组第u个位置是不是为空，不为空则返回其他线程生成的segment对象
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
// 以初始化时创建的第一个坑位的ss[0]作为模版进行创建
Segment<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
// 二次检查是否有其它线程创建了这个Segment
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
//这里通过自旋的CAS方式对segments数组中偏移量为u位置设置值为s，这是一种不加锁的方式，
//万一有多个线程同时执行这一步，那么只会有一个成功，而其它线程在看到第一个执行成功的线程结果后
//会获取到最新的数据从而发现需要更新的坑位已经不为空了，那么就跳出while循环并返回最新的seg
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}6.调用segment对象的put方法
首先遍历该链表，如果在该链中找到相同的key，则用新值替换旧值，并退出循环，如果没有和key相同的，一直遍历到链尾，链尾的next为null，如果node不为空则调用 node.setNext(first)方法，把当前的Entry插入到表头。否则创建一个Entry插入到表头。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//这里是并发的关键，每一个Segment进行put时，都会加锁
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
//tab是当前segment所连接的HashEntry数组
HashEntry<K,V>[] tab = table;
//确定key的hash值所在HashEntry数组的索引位置
int index = (tab.length - 1) & hash;
//取得要放入的HashEntry链的链头
HashEntry<K,V> first = entryAt(tab, index);
//遍历当前HashEntry链
for (HashEntry<K,V> e = first;;) {
//如果链头不为null
if (e != null) {
K k;
//如果在该链中找到相同的key，则用新值替换旧值，并退出循环
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
//如果没有和key相同的，一直遍历到链尾，链尾的next为null，进入到else
e = e.next;
}
else {//如果没有找到key相同的，则把当前Entry插入到链头
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
//此时数量+1
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;
}
scanAndLockForPut方法：调用segment对象的put方法的时候会调用tryLock()尝试去加锁【trylock如果获取不到锁会立即返回false,lock获取锁的时候，如果获取不到就会阻塞，一直等到获取锁成功】。如果tryLock加锁成功则node为空，如果trylock失败则调用scanAndLockForPut()方法scanAndLockForPut()方法的目的是在线程等待的时间内做一些准备工作，创建HashEntry对象。
scanAndLockForPut()方法的工作过程：首先是循环的尝试获取锁，通过循环去遍历链表。循环的时候有三个分支，第一步先判断retries是否为-1，如果为是则判断HashEntry是否为空，如果为空则创建HashEntry对象，如果不为空就判断该链的节点是否相同。如果都不是则进入下一节点。第二步：retries ++并判断retries是否达到最大值，如果是则调用Lock()方法一直等待到获取锁成功。第三步：在循环中有一个重试机制，每一次循环retries 都会加一，当retries 为偶数的时候，会重试判断当前链的头是不是发生改变，如果发生改变则重新遍历该链表。最后返回node。
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; ;//检查重试机制
while (!tryLock()) {
HashEntry<K,V> f; ;//首先检查下面
if (retries < 0) {
if (e == null) {
if (node == null) //推测创建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; //如果条目改变，则重新遍历
retries = -1;
}
}
return node;
} public V put(K key, V value) {
Segment<K,V> s;
//ConcurrentHashMap的key和value都不能为null
if (value == null)
throw new NullPointerException();
//这里对key求hash值，并确定应该放到segment数组的索引位置
int hash = hash(key);
//j为索引位置，思路和HashMap的思路一样
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
//生成segment对象
s = ensureSegment(j);
//这里很关键，找到了对应的Segment，则把元素放到Segment中去
return s.put(key, hash, value, false);
}
5.UNSAFE的简单使用
案例1：通过UNSAFE来获取内存中数组下标为1的值
public class Person {
private Integer i = 0;
private static sun.misc.Unsafe UNSAFE;
private static long I_OFFSET;
private String[] table = {"1","2","3","4"};
static {
try {
//通过反射获取UNASFE实例
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
UNSAFE = (Unsafe) field.get(null);
} catch (NoSuchFieldException | IllegalAccessException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
final Person person = new Person();
//数组中存储的对象的对象头大小
int ns = UNSAFE.arrayIndexScale(String[].class); //4
//取数组中第一个元素的起始位置
int base = UNSAFE.arrayBaseOffset(String[].class);
System.out.println(UNSAFE.getObject(person.table,base+1*ns)); //结果为1
}
}案例2：再ConcurrentHashMap用UNSAFE来获取内存当中的数据，来保证在高并发的场景下每个线程拿到的数据都是最新的数据。
package com.company;
import sun.misc.Unsafe;
import java.lang.reflect.Field;
public class Person {
private int i = 0;
private static sun.misc.Unsafe UNSAFE;
private static long I_OFFSET;
static {
try {
//通过反射获取UNASFE实例
Field field = Unsafe.class.getDeclaredField("theUnsafe");
field.setAccessible(true);
UNSAFE = (Unsafe) field.get(null);
//获取Person属性i字段的偏移量
I_OFFSET = UNSAFE.objectFieldOffset(Person.class.getDeclaredField("i"));
} catch (NoSuchFieldException | IllegalAccessException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
final Person person = new Person();
new Thread((Runnable) () -> {
while (true){
//内存中的i（I_OFFSET）与缓存中的i（person.i）做比较，如果相等则加+
boolean b = UNSAFE.compareAndSwapInt(person,I_OFFSET,person.i,person.i+1);
if(b){
System.out.println(UNSAFE.getIntVolatile(person,I_OFFSET));
}
try {
Thread.sleep(500);
}catch (InterruptedException e){
e.printStackTrace();
}
}
}).start();
new Thread((Runnable) () -> {
while (true){
//内存中的i（I_OFFSET）与缓存中的i（person.i）做比较，如果相等则加+
boolean b = UNSAFE.compareAndSwapInt(person,I_OFFSET,person.i,person.i+1);
if(b){
System.out.println(UNSAFE.getIntVolatile(person,I_OFFSET));
}
try {
Thread.sleep(500);
}catch (InterruptedException e){
e.printStackTrace();
}
}
}).start();
}
}
6.ConcurrentHashMap的扩容
ConcurrentHashMap不存在在高并发场景下HashMap在扩容中存在的循环链表的问题，因为ConcurrentHashMap在put时已经加锁。
ConcurrentHashMap的rehash方法和Hashmap中的resize方法基本是一样的，都是先扩容两倍，再通过((newTableLentgh-1) & key.hash)求出节点在新table中的位置，接着用头插法把原table中的节点插入到新table中。不过Segment的rehash方法加入了一点优化判断，就是在遍历原table的时候，会用lastRun和lastIdx来记录每个单链表上的一个节点，如果这个节点及后面的节点在新table中的映射位置相同，这样在插入的时候直接插入lastRun即可，后面的元素也就跟着一起插入了。也就是记录当前循环次数中连续的节点的头节点，如果后面的循环同样出现连续的节点，那么则覆盖前面已经记录的头节点，永远记录最新的头节点。
问题？新扩容的数组的某个下标上也有数据，然后在下一次循环转移原有链表的数据到同样位置的下标的时候会存在冲突？
rehash的方法说明中写道：意思是oldTable中的节点插入到newTable中，节点的位置要么不变，要么都偏移一个2次幂数，这说明不会有其它位置上的单链表的节点会插入到newTable中的lastIdx位置。
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
// 遍历segemnt链表
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // 单节点链表
newTable[idx] = e;
else {
//该位置存在多节点链表
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask; //判断添加到哪个链表中去
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// 克隆剩余节点
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
//添加最新的node节点到新数组中
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
7.ConcurrentHashMap的查询
get方法是没有加锁解锁动作的，但是也能实现并发同步的效果。原因是因为通过UNSAFE进行查询得到内存中的数据，对于其他线程来说是可见的。详情可见：https://www.cnblogs.com/keeya/p/9632958.html。
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
//要查找的key属于segment[u]
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
//segment[u] != null且segment[u]中的table[] != null
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//找到key映射位置上的链表并遍历
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;
//查找成功就返回val
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
//查找失败返回null
return null;
}
8.ConcurrentHashMap的删除
remove操作首先尝试获取锁，获取到了锁后进行table的遍历，查找到指定key位置的节点后删除该节点，并返回原节点的值。节点删除过程同Hashmap。
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;
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;
}
以上是关于JDK1.7中的ConcurrentHashMap的主要内容，如果未能解决你的问题，请参考以下文章
ConcurrentHashMap在jdk1.7和jdk1.8中的不同
Java并发编程总结4——ConcurrentHashMap在jdk1.8中的改进
java并发：jdk1.8中ConcurrentHashMap源码浅析