在之前的文章提到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 不存在可见性。
今天的文章concurrenthashmap1.7原理_深入理解MVC分享到此就结束了,感谢您的阅读。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/74046.html