概述
在之前的文章提到ConcurrentHashMap是一个线程安全的，那么我么看一下ConcurrentHashMap如何进行操作的。
ConcurrentHashMap与HashTable区别？
HashTable
put()源代码
我们来看一下put 操作：
方法体 被 同步锁标记，由于同步锁的特性，其他线程将会排队进行等待处理。
除此之外，对传入的key 值进行了一个判断空值逻辑。【PS：HashMap 是允许key值为空的】
**ConcurrentHashMap **
分段锁技术：ConcurrentHashMap 相比 HashTable 对锁的处理不同的点在于：前者是分段部分数据锁定
每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，后者是全部锁定。【PS：下图 基于JDK 1.7绘制】
从图中可以看出来ConcurrentHashMap的主干是个Segment数组。
由于ConConcurrentHashMap 的主体是由多个Segment 链式组成，因此每个Segment都持有自己的锁。
final Segment<K,V>[] segments;
需要注意的是 Segment 是一种可重入锁（继承ReentrantLock)
那么我简单说一下ReentrantLock 与synchronized有什么区别？
-
synchronized 是一个同步锁 synchronized （）
- 同步锁 当一个线程A 访问 【资源】的代码同步块时，A线程就会持续持有当前锁的状态，如果其他线程B-E 也要访问【资源】时，此时代码同步块将会阻塞，因此B-E线程需要排队等待A线程释放锁的状态后才可以持有【资源】。
-
ReentrantLock 可重入锁 JDK 5 引入的。 顾名思义 可重复进入的锁【Re-entrantLock】，它表示该锁能支持一个线程对资源的重复加锁。同时还支持获取锁的公平和非公平性选择。
-
我们来说说他的两类特性：公平性、非公平性
简单的来说就是ReentrantLock构造对象的时候传入一个值有选择的判断他是否是公平还是不公平。如图例子 -
- 公平性：根据线程请求锁的顺序依次获取锁，当一个线程A 访问 【资源】的期间，线程A 获取锁资源，此时内部存在一个计数器num+1，在访问期间，线程B、C请求 资源时，发现A 线程在持有当前资源，因此在后面线程节点排队（B 处于待唤醒状态），假如此A线程再次请求资源时，不需要再次排队，可以直接再次获取当前资源 （内部计数器+1 num=2） ，当A线程释放所有锁的时候（num=0），此时会唤醒B线程进行获取锁的操作，其他C-E线程就同理。（情况2）
- 非公平性：当A线程已经释放所之后，准备唤醒线程B获取资源时，此时线程M 获取请求，此时会出现竞争，线程B 没有竞争过M线程，测试M获取此线程，因此M会有限获得资源，B继续睡眠。（情况2）
-
synchronized 是一个非公平性锁。通俗来讲暴利等待。
非公平锁总体会比公平要好一些，它是根据每个线程对资源抢占能力来分配的，不需要严格的安装锁的请求顺序接入
ReentrantLock 使用场景
- 定时任务 防止任务重复执行。
- 套字节连接池，如果正在数据通信防止重复接入连接
在了解以上的功能 我们之后我们继续看一下ConcurrentHashMap核心构造方法代码。
// 跟HashMap结构有点类似
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;//负载因子
this.threshold = threshold;//阈值
this.table = tab;//主干数组即HashEntry数组
}
构造方法
从以上代码可以看出ConcurrentHashMap有比较重要的三个参数：
- loadFactor 负载因子 0.75
- threshold 初始 容量 16
- concurrencyLevel 实际上是Segment的实际数量 默认16。
ConcurrentHashMap如何发生ReHash？
ConcurrentLevel 一旦设定的话，就不会改变。ConcurrentHashMap当元素个数大于临界值的时候，就会发生扩容。但是ConcurrentHashMap与其他的HashMap不同的是，它不会对Segmengt 数量增大，只会增加Segmengt 后面的链表容量的大小。即对每个Segmengt 的元素进行的ReHash操作。
我们再看一下核心的ConcurrentHashMap Put ()方法：
1.7(版本)
public V put(K key, V value) {
Segment<K,V> s;
//concurrentHashMap不允许key/value为空
if (value == null)
throw new NullPointerException();
//hash函数对key的hashCode重新散列，避免差劲的不合理的hashcode，保证散列均匀
int hash = hash(key);
//返回的hash值无符号右移segmentShift位与段掩码进行位运算，定位segment
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);
}
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);
//tryLock()是ReentrantLock获取锁一个方法。如果当前线程获取锁成功 返回true,如果别线程获取了锁返回false不成功时会遍历定位到的HashEnry位置的链表（遍历主要是为了使CPU缓存链表），若找不到，则创建HashEntry。tryLock一定次数后（MAX_SCAN_RETRIES变量决定），则lock。若遍历过程中，由于其他线程的操作导致链表头结点变化，则需要重新遍历。
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;//定位HashEntry，可以看到，这个hash值在定位Segment时和在Segment中定位HashEntry都会用到，只不过定位Segment时只用到高几位。
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;
//若c超出阈值threshold，需要扩容并rehash。扩容后的容量是当前容量的2倍。这样可以最大程度避免之前散列好的entry重新散列。扩容并rehash的这个过程是比较消耗资源的。
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
1.8版本
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
1.7 的流程大致：
优先计算出Key 通过Hash函数计算出hash值 现计算出当前key属于哪个Segment 然后调用Segment.put 分段方法Segment.put()
- Put 时候 ，通过Hash函数将即将要put 的元素均匀的放到所需要的Segment 段中，然后调用Segment的put 方法进行添加数据。
- Segment的put 是加锁中完成的。如果当前元素数大于最大临界值的的话将会产生rehash. 先通过 getFirst 找到链表的表头部分，然后遍历链表，调用equals 比配是否存在相同的key ,如果找到的话，则将最新的Key 对应value值。如果没有找到，新增一个HashEntry 它加到整个Segment的头部。
1.8 采用红黑树的Node存储结构，因此在计算过程中做了一些调整优化。大致列了一下：
- 由于是树形存储，计算hash值 进行spread 将散列的较高位散布（XOR）降低 将位数控制在int最大整数之内
- 在添加数据元素过程中，将链表转换成树形结构，同时对树形结构节点元素查找和添加进行结构化的调整。
- 写入数据中增加了CAS 方法 compareAndSwapInt，compareAndSwapLong
我们先看一下Get 方法的源码：
1.7
//计算Segment中元素的数量
transient volatile int count;
***********************************************************
public V get(Object key) {
int hash = hash(key.hashCode());
return segmentFor(hash).get(key, hash);
}
***********************************************************
final Segment<K,V> segmentFor(int hash) {
return segments[(hash >>> segmentShift) & segmentMask];
}
********************************************************
V get(Object key, int hash) {
if (count != 0) { // read-volatile
HashEntry<K,V> e = getFirst(hash);
while (e != null) {
if (e.hash == hash && key.equals(e.key)) {
V v = e.value;
if (v != null)
return v;
return readValueUnderLock(e); // recheck
}
e = e.next;
}
}
return null;
}
1.7 大致流程如下：
1.传入读取Key值，通过Hash函数计算出 对应Segment 的位置。
2.调用segmentFor（int hash）方法，计算确定操作应该在哪一个segment中进行 ，通过右无符号位运算 进行右移操作 计算出偏移值 获得需要操作的Segment位置。
- 确定需要操作的Segment后，再调用 get 方法获取对应的值。通过count 值先判断当前值是否为空。在调用getFirst（）方法获取头节点，然后在Segment内部遍历列表通过equals对比的方式进行比对返回值。
1.8 Get 方法
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
1.8 大致思路和1.7 一致，由于1.8存储结构形态采用了树形数据结构，还是通过get操作通过首先计算key的hash值来确定该元素放在数组的哪个位置，然后进行顺序遍历直到找到确定的数值。
ConcurrentHashMap为什么读的时候不加锁？
- ConcurrentHashMap是分段并发分段进行读取数据的。
- 【jdk1.7】 Segment 里面采用很多计数变量都通过volatile关键字修饰，由于volatile变量 happer-before的特性。导致get 方法能够几乎准确的获取最新的数据并且更新。
再看一下 1.7 jdk ConcurrentHashMapRemove()方法：
V remove(Object key, int hash, Object value) {
lock();
try {
int c = count - 1;
HashEntry<K,V>[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue = null;
if (e != null) {
V v = e.value;
if (value == null || value.equals(v)) {
oldValue = v;
// All entries following removed node can stay
// in list, but all preceding ones need to be
// cloned.
++modCount;
HashEntry<K,V> newFirst = e.next;
for (HashEntry<K,V> p = first; p != e; p = p.next)
newFirst = new HashEntry<K,V>(p.key, p.hash,
newFirst, p.value);
tab[index] = newFirst;
count = c; // write-volatile
}
}
return oldValue;
} finally {
unlock();
}
}
- 调用Segment 的remove 方法，先定位当前要删除的元素C，此时需要把A、B元素全部复制一遍，一个一个接入到D上。
- remove 也是在加锁的情况下进行的。
PS:JDK1.8的删除元素在此就不详细展示，感兴趣的小伙伴可以私底下研究一下。
volatile 变量
volatile 变量 是保证修饰变量具有可见性的变量
我们发现 对于CurrentHashMap而言的话，源码里面又很多地方都用到了这个变量。比如HashEntry 、value 、Segment元素个数Count。
volatile 属于JMM 模型中的一个词语。首先先简单说一下 Java内存模型中的 几个概念：
- 原子性：保证 Java内存模型中原子变量内存操作的。通常有 read、write、load、use、assign、store、lock、unlock等这些。
- 可见性：就是当一个线程对一个变量进行了修改，其他线程即可立即得到这个变量最新的修改数据。
- 有序性：如果在本线程内观察，所有操作都是有序的；如果在一个线程中观察另一个线程，所有操作都是无序的。
- 先行发生：happen-before 先行发生原则是指Java内存模型中定义的两项操作之间的依序关系，如果说操作A先行发生于操作B，其实就是说发生操作B之前.
- 传递性
volatile 变量 与普通变量的不同之处？
- volatile 是有可见性，一定程度的有序性。
- volatile 赋值的时候新值能够立即刷新到主内存中去，每次使用的时候能够立刻从内存中刷新。
做一个简单例子看一下 这个功能
public class VolatileTest{
int a=1;
int b=2;
//赋值操作
public void change(){
a=3;
b=a;
}
//打印操作
public void print(){
System.out.println("b:"+b+",a:"+a);
}
@Test
public void testNorMal(){
VolatileTest vt=new VolatileTest();
for (int i = 0; i < 100000; i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
vt.change();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
vt.print();
}
}).start();
}
}
}
跑了 n 次会出现一条 b=3,a=1 的错误打印记录。这就是因为普通变量相比volatile 不存在可见性。
最后
以上就是结实衬衫为你收集整理的深入理解ConcurrentHashMap原理分析以及线程安全性问题的全部内容，希望文章能够帮你解决深入理解ConcurrentHashMap原理分析以及线程安全性问题所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
发表评论 取消回复