标签：hash int cap JDK1.7 源码 HashEntry Segment ssize ConcurrenHashMap
世味年来薄似纱，谁令骑马客京华。
小楼一夜听春雨，深巷明朝卖杏花。
矮纸斜行闲作草，晴窗细乳戏分茶。
素衣莫起风尘叹，犹及清明可到家。——陆游《临安春雨初霁》
一、前言
2.1 ConcurrentHashMap的锁分段技术
HashTable容器在竞争激烈的并发环境下表现出效率低下的原因，是因为所有访问HashTable的线程都必须竞争同一把锁，那假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术，首先将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。
二、JDK1.7中ConcurrentHashMap
2.1 初识ConcurrentHashMap
我们通过ConcurrentHashMap的类图来分析ConcurrentHashMap的结构:
JDK1.7中ConcurrentHashMap允许多个修改操作并发进行，其关键在于使用了锁分离技术。它使用了多个锁来控制对hash表的不同部分进行的修改。ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的Hashtable，它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行。
有些方法需要跨段，比如size()和containsValue()，它们可能需要锁定整个表而而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁。这里“按顺序”是很重要的，否则极有可能出现死锁，在ConcurrentHashMap内部，段数组是final的，并且其成员变量实际上也是final的，但是，仅仅是将数组声明为final的并不保证数组成员也是final的，这需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。
2.2 Segment（锁分段）
- 在ConcurrentHashMap中是使用多个哈希表，具体为通过定义一个Segment来封装这个哈希表其中Segment继承于ReentrantLock，故自带lock的功能。即每个Segment其实就是相当于一个HashMap，只是结合使用了ReentrantLock来进行并发控制，实现线程安全。
- Segment定义如下：ConcurrentHashMap的一个静态内部类，继承于ReentrantLock，在内部定义了一个HashEntry数组table，HashEntry是链表的节点定义，其中table使用volatile修饰，保证某个线程对table进行新增链表节点（头结点或者在已经存在的链表新增一个节点）对其他线程可见。
static final class Segment<K,V> extends ReentrantLock implements Serializable {
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
transient volatile HashEntry<K,V>[] table;
...
}- HashEntry的定义如下：包含key，value，key的hash，所在链表的下一个节点next
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
...
}由定义可知，value和next均为使用volatile修饰，当多个线程共享该HashEntry所在的Segment时，其中一个线程对该Segment内部的某个链表节点HashEntry的value或下一个节点next修改能够对其他线程可见。而hash和key均使用final修饰，因为创建一个链表节点HashEntry，是根据key的hash值来确定附加到哈希表数组table的某个链表的，即根据hash计算对应的table数组的下标，故一旦添加后是不可变的。
2.2.3 Segment的哈希表table数组的容量
MIN_SEGMENT_TABLE_CAPACITY：table数组的容量最小量，默认为2
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
2.3 成员变量
//段掩码
final int segmentMask;
//段偏移量
final int segmentShift;
/**
* The segments, each of which is a specialized hash table.
*/
final Segment<K,V>[] segments;
transient Set<K> keySet;
transient Set<Map.Entry<K,V>> entrySet;
transient Collection<V> values;初始化segmentShift和segmentMask。这两个全局变量在定位segment时的哈希算法里需要使用，sshift等于ssize从1向左移位的次数，在默认情况下concurrencyLevel等于16，1需要向左移位移动4次，所以sshift等于4。segmentShift用于定位参与hash运算的位数，segmentShift等于32减sshift，所以等于28，这里之所以用32是因为ConcurrentHashMap里的hash()方法输出的最大数是32位的，后面的测试中我们可以看到这点。segmentMask是哈希运算的掩码，等于ssize减1，即15，掩码的二进制各个位的值都是1。因为ssize的最大长度是65536，所以segmentShift最大值是16，segmentMask最大值是65535，对应的二进制是16位，每个位都是1。
（都是在默认情况下，concurrencyLevel=16,ssize=16,sshift=4）
那么
this.segmentShift = 32 - sshift=28
而this.segmentMask = ssize - 1=15 二进制就是1111;
hash >>> segmentShift) & segmentMask//定位Segment所使用的hash算法
int index = hash & (tab.length - 1);// 定位HashEntry所使用的hash算法
关于segmentShift和segmentMask
segmentMask：段掩码，假如segments数组长度为16，则段掩码为16-1=15；segments长度为32，段掩码为32-1=31。这样得到的所有bit位都为1，可以更好地保证散列的均匀性
segmentShift：2的sshift次方等于ssize，segmentShift=32-sshift。若segments长度为16，segmentShift=32-4=28;若segments长度为32，segmentShift=32-5=27。而计算得出的hash值最大为32位，无符号右移segmentShift，则意味着只保留高几位（其余位是没用的），然后与段掩码segmentMask位运算来定位Segment。
2.4 ConcurrentHashMap构造方法
在ConcurrentHashMap的构造函数定义实际大小：使用ConcurrentHashMap的整体容量initialCapacity除以Segments数组的大小，得到每个Segment内部的table数组的实际大小。
public ConcurrentHashMap() {
//DEFAULT_INITIAL_CAPACITY:16;DEFAULT_LOAD_FACTOR:0.75;DEFAULT_CONCURRENCY_LEVEL:16
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}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;
int ssize = 1;
// ssize：segments数组的大小
// 不能小于concurrencyLevel，默认为16
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
// cap：Segment内部HashEntry数组的大小
// 最小为MIN_SEGMENT_TABLE_CAPACITY，默认为2
// 实际大小根据c来计算，而c是由上面代码，
// 根据initialCapacity / ssize得到，
// 即整体容量大小除以Segment数组的数量，则
// 得到每个Segment内部的table的大小
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]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}这段代码比较长，我们选择代码块逐个分析：
1.代码块一
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}<<运算规则：按二进制形式把所有的数字向左移动对应的位数，高位移出（舍弃），低位的空位补零。
语法格式：
需要移位的数字 << 移位的次数
例如： 3 << 2，则是将数字3左移2位
计算过程：
3 << 2
初始情况下 ssize=1 concurrencyLevel=16
ssize <<= 1;
第一次循环 ssize=1<16 进入循环 ssize左移1位变成2；
第二次循环 ssize=2<16 进入循环 ssize左移1位变成4；
第三次循环 ssize=4<16 进入循环 ssize左移1位变成8；
第四次循环 ssize=8<16 进入循环 ssize左移变成16；此时 ssize变成了16 不小于16 跳出循环。
这块代码的目的是去找到一个最小的大于等于concurrencyLevel的2的幂次方数
大于等于16的最小2的幂次方数就是16呀 所以我们找到ssize就是16
同时 心细一点的同学跟着刚才的循环去计算 得到的sshift是4 也就是找到了2^4等于16的这个4
2.代码块2
我们先注释掉一部分不去管它
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;这块代码就要引出ConcurrentHashMap的底层结构了 要认真理解这块。
//initialCapacity的默认值是16
//ssize上面我们算出来是16
//我们可以得到下面这个写的不太规范的式子
int c=16/16=1这里的c是什么呢 ？ 这里的c先把他理解成就是Segment数组的大小（其实c的含义是不对的 但我们先这么理解） 我们是根据 initialCapacity和ssize计算出来的。
别着急 下面还有完善
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;这个MIN_SEGMENT_TABLE_CAPACITY的**默认值是2（也就是说设计者设计这个Segment数组最小长度就是2） **就是说现在cap=2；c=1；
2不小于1 所以不会进入这个while循环 接着往下，
我们这里算的c=1 比设计者设计的Segment数组最小长度还要小 所以我们按cap=2去初始化
如上图所示 cap指定了每一个Segment可以放几个HashEntry
ssize指定了一个ConcurrentHashMap可以放多少个Segment
现在 我们就可以重新绘制一下这个数据结构图了，
这其实就是默认情况下 ConcurrentHashMap的数据结构了：
这里 我们把刚刚注释掉的代码打开 再去分析一波：
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;假设 我这里指定了initialCapacity是33，
那我们这个时候33/16 计算出来得c就是2 这个时候就会进入if这里 c就会加1 此时c等于了3
while (cap < c)
cap <<= 1;当代码走到这里 cap=2 c=3 2<3，就会进入while当中 cap左移变成了4 这里实质性的改变就是对应的每一个Segment就会有4个HashEntry。
这里说明一下 为什么cap使用了左移 从2变成了4 这是因为设计者要Segment的大小不论是几都应该是2的幂次方数。
此时的数据结构是这样：
2.5 ConcurrentHashMap的put方法
public V put(K key, V value) {
Segment<K,V> s;
//value不能为null，为null抛异常
if (value == null)
throw new NullPointerException();
//hash函数对key的hashCode重新散列，避免差劲的不合理的hashcode，保证散列均匀
int hash = hash(key);
//返回的hash值无符号右移segmentShift位与段掩码进行位运算，定位segment
//(1) j是算出来的Segment数组的下标
int j = (hash >>> segmentShift) & segmentMask;
//(2) 通过Unsafe类去取segments数组第j个位置的元素看是不是null
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
//(3) 如果是null 去生成一个Segment对象
s = ensureSegment(j);
//(4) 去调用生成的Segment对象的put方法
return s.put(key, hash, value, false);
}小问题：
1.(j << SSHIFT) + SBASE就是segments数组中该segment的位置！ 2.(j << SSHIFT) + SBASE：中SBASE值是怎么来的？
从源码看出，put的主要逻辑也就两步：
- 定位segment并确保定位的Segment已初始化
- 调用Segment的put方法。
我们把(3)处的ensureSegment方法展开说明一下 去理解他是怎么生成一个Segment对象并保证线程安全的：
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;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
//这部分代码就是将一开始构造方法生成的ss[0]作为一个原型（雏形）
//利用ss[0]去初始化我们此时的Segment对象
//但是真正初始化在下面一个if之后 这里可以理解是做了一个准备工作
//准备了一些需要的属性 如负载因子啊 cap长度啊
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];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
//这时候又再次判断该位置有没有其他线程进行了初始化
//没有其他线程的话 这时候真正去创建一个Segment对象
//但是这里还没有把Segment对象放到数组对应的位置
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
//这里的这个CAS操作真正对第u个位置进行赋值
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}上述代码用到了双重检查锁（double check）。
由上面代码可知创建segment对象是怎么保证线程安全的：
1.用到了双重检查锁（double check）；
2.赋值的时候 利用了CAS这个原子操作。
CAS这个原子操作是不能被中断的 我们这里简单谈一下CAS干了什么。
CAS就是先获取主物理内存中的值作为期望值 然后我们再去获得此时主物理内存的真实值 如果期望值与真实值一致 我们就进行修改 否则 就一直取值比较 直到成功
再看看 (4) 处代码，去调用生成的Segment对象的put方法：
//(4) 去调用生成的Segment对象的put方法
return s.put(key, hash, value, false);我们去继续学习这个Segment对象的put方法
这里我们先去抽象出一个数据结构 也就是说Segment内部维护的一个个HashEntry整合起来 就好像一个小的HashMap一样 也是数组+链表的形式(Segment内部就像一个小的HashMap)
我们先不去看加锁的逻辑 我们先把中间怎么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;
//取tab数组 下表为index的值作为first
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
//链表的头结点不为空的情况
if (e != null) {
K k;
//遍历当前位置的链表
//判断传入的key 和当前遍历的 key 是否相等，相等则覆盖旧的 value
//这里根HashMap的逻辑很像
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
//为空的情况
//情况1 头结点为空 把key-value放在头结点
//情况2 遍历完整个链表 然后头插法插入
else {
if (node != null)
node.setNext(first);
else
//生成了一个HashEntry对象 记为node
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
//如果超过阈值 就rehash
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
//没有超过阈值 就把刚刚生成的node通过setEntryAt这个方法放进去
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}由于Segment的put方法里需要对共享变量进行写入操作，所以为了线程安全，在操作共享变量时必须加锁。第一个put方法已经定位到Segment，然后这个方法中就是在Segment里进行插入操作。此方法插入操作需要经历两个步骤，第一步判断是否需要对Segment里的HashEntry数组进行扩容，第二步定位添加元素的位置，然后将其放在HashEntry数组里。
a.是否需要扩容
在插入元素前会先判断Segment里的HashEntry数组是否超过容量（threshold），如果超过阈值，则对数组进行扩容。值得一提的是，Segment的扩容判断比HashMap更恰当，因为HashMap是在插入元素后判断元素是否已经到达容量的，如果到达了就进行扩容，但是很有可能扩容之后没有新元素插入，这时HashMap就进行了一次无效的扩容。
b.如何扩容
在扩容的时候，首先会创建一个容量是原来容量两倍的数组，然后将原数组里的元素进行再散列后插入到新的数组里。为了高效，ConcurrentHashMap不会对整个容器进行扩容，而只对某个segment进行扩容。
c.对于ConcurrentHashMap的数据插入，这里要进行两次Hash去定位数据的存储位置。
static class Segment<K,V> extends ReentrantLock implements Serializable {从上Segment的继承体系可以看出，Segment实现了ReentrantLock，也就带有锁的功能，当执行put操作时，会进行第一次key的hash来定位Segment的位置，如果该Segment还没有初始化，即通过CAS操作进行初始化，然后进行第二次hash操作，找到相应的HashEntry的位置，这里会利用继承过来的锁的特性，在将数据插入指定的HashEntry位置时（链表的尾端），会通过继承ReentrantLock的tryLock（）方法尝试去获取锁，如果获取成功就直接插入相应的位置，如果已经有线程获取该Segment的锁，那当前线程会以自旋的方式去继续的调用tryLock（）方法去获取锁，超过指定次数就挂起，等待唤醒。
我们上面初步分析了一下 put添加数据的过程
下面我们重点分析一下加锁的过程
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
lock()——是一个无条件的锁，与synchronize意思差不多。 tryLock()——ReentrantLock类特有的获取锁的方法。 unlock()——ReentrantLock类特有的释放锁的方法。trylock()这个方法 如果能够获取到锁 就会立马返回一个true
trylock()这个方法 如果获取不到锁 就会立马返回一个false
trylock()不会阻塞
lock()这个方法如果获取不到锁 就会一直阻塞在这里进一步了解请移步： Java高级-线程同步lock与unlock使用
这个scanAndLockForPut方法大概干了什么事情 我给大家解释一下:
当trylock()获取不到锁的时候 通过刚刚我们的铺垫我们知道trylock()是不会阻塞的。
那我们不能傻傻的等在这里 我们既然不会阻塞 我们在这个过程中可以准备一些什么事情呀？
这个过程就好比做饭 你在烧水等水开的过程中 可以去准备个凉菜 算是合理安排 提高效率。
我们这里的合理安排就是根据key-value去new一个HashEntry 我们把这个HashEntry记成node。
这里我们就把scanAndLockForPut这个方法做的事情给大家大致说明白了，scanAndLockForPut这个方法本身就设计的非常精妙 由于篇幅的原因就不在展开描述，之后会有更加详细的说明解释。
tryLock() ? null : scanAndLockForPut(key, hash, value);这里不是有个三目运算符吗 trylock()获取不到锁的时候 就会走scanAndLockForPut这个方法准备一个node对象出来 。
我的理解：
如果tyrLock()未获取到锁，会新建一个node对象，并等待锁。等下次获取到了锁，此时名为first的HashEntry对象已经被另一个线程放到tab的响应位置了，所以需要把它设置为node节点的下一个节点，行成链表。
我们理解一下这里的保证线程安全 为什么用了lock
如上图所示 我们根本的目的是把key-value放进去
我们要在链表当中去插入元素 注意是插入元素 这个时候CAS就没有更好的办法了 因为CAS对某一个具体的位置赋 值还是可以的 但是让CAS去插入是不能实现的 所以这个插入时候我们为了保证线程安全 就要去加锁。
这里保证线程安全的方法很实在 就是加了一把锁 让同一时间只有一个线程去put数据。
我们总结一下jdk7 下 ConcurrentHashMap是怎么保证并发安全的：
1. 在进行一些链表的插入数据时 用了ReentrantLock去加了一把锁
2.用了UNSAFE的各种方法 这其中包括了我们最熟悉的CAS 还有UNSAFE类的一些其他方法呀
比如 UNSAFE.putOrderedObject等等
2.6 UNSAFE.getObject() 方法
移步： Unsafe API介绍及其使用
标签：hash,int,cap,JDK1.7,源码,HashEntry,Segment,ssize,ConcurrenHashMap 来源： https://blog.csdn.net/moneywenxue/article/details/111188084
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。