在上一篇文章中，讲解了HashMapjdk1.7使用头插法，容易形成环形链表，在jdk1.8改为尾插法，虽然解决了环形链表的问题，但是会导致数据丢失的问题，所以我们说HashMap在多线程环境下是不安全的。
所以，不管是jdk1.7还是jdk1.8的HashMap 都存在线程安全的问题。那么在多线程环境下应该这样去保证线程安全呢，有什么办法呢？
上一篇文章中有说到3种解决方案：
- 在多线程环境下用HashTable来解决线程安全的问题,put或者get方法源码如下图：
//put方法
public synchronized V put(K key, V value) {
// Make sure the value is not null
if (value == null) {
throw new NullPointerException();
}
// Makes sure the key is not already in the hashtable.
Entry tab[] = table;
int hash = key.hashCode();
int index = (hash & 0x7FFFFFFF) % tab.length;
@SuppressWarnings("unchecked")
Entry entry = (Entry)tab[index];
for(; entry != null ; entry = entry.next) {
if ((entry.hash == hash) && entry.key.equals(key)) {
V old = entry.value;
entry.value = value;
return old;
}
}
addEntry(hash, key, value, index);
return null;
}
//get方法
@SuppressWarnings("unchecked")
public synchronized V get(Object key) {
Entry tab[] = table;
int hash = key.hashCode();
int index = (hash & 0x7FFFFFFF) % tab.length;
for (Entry e = tab[index] ; e != null ; e = e.next) {
if ((e.hash == hash) && e.key.equals(key)) {
return (V)e.value;
}
}
return null;
}
从上面的源码中可以看出不管是put还是get *** 作都会使用synchronized加锁，由此带来的问题是在这个时间段内只能有一个线程可以 *** 纵HashTable，最不能接收的，就是get方法只是一个读取 *** 作为什么要加锁呢？就连删除也都加上锁，你这是为了安全什么都锁啊，你牛！
如果线程数量大一点的话，HashTable的性能会急剧下降，因为每次 *** 作都需要锁住整个对象，而其他线程在此期间是不能 *** 作的。不仅如此，还会带来额外的上下文切换等开销，所以此时它的吞吐量甚至还不如单线程的情况,因此并不适合高并发场景。
- Collections.synchronizedMap(Map
它和HashTable差不多，要锁住整张表，效率低下，如下图源码：
从源码中可以看出SynchronizedMap在Hashtable对比上改进了很多，synchronized不再放在方法上，而是放在方法内部，作为同步块出现，但仍然是对象级别的同步锁，读和写 *** 作都需要获取锁，本质上，仍然只允许一个线程访问，其他线程被排斥在外。
- 上面的两种方式实现线程安全性都不是很合适，所以剩下ConcurrentHashMap，既然锁住整张表效率低下，那能不能把表分成N份呢，把数据尽量均匀的分布到每个部分中，分别给他们加锁，互相之间并不影响呢？那就使用ConcurrentHashMap实现，因为ConcurrentHashMap将Map分段了，每个段进行加锁，而不是像HashTable,SynchronizedMap都是整个map加锁，这样就可以多线程访问了！
在jdk1.7中ConcurrentHashMap 是基于分段锁的，就是将内部分成不同的 Segment（段），每个段里面是HashEntry数组。
在jdk1.8之后对jdk1.7做了很大的改进，不再采用分段锁的机制了，而采用的是 Synchronized + CAS（Compare and Swap，即比较并替换，实现并发算法时常用到的一种技术） ，把锁的粒度进一步降低，而放弃了 Segment 分段。
下面我们就以 jdk1.8的ConcurrentHashMap中put方法源码进行解析
而put方法中主要有以下几个重点的方法进行讲解：
- initTable方法：初始化表
- helpTransfer方法：帮助扩容
- addCount方法：增加元素个数，并且触发扩容 *** 作
- fullAddCount方法：初始化CounterCell，来记录元素个数
- transfer方法：扩容和迁移元素的
说明一下：由于jdk1.8的HashMap和ConcurrentHashMap的基本属性变量、结构和一些初始化相关的逻辑都差不多，这里不在单独说明介绍。看懂HashMap的伙伴们都知道滴吧，嘻嘻
put方法先从put方法中看一下插入新元素的时候，怎么保证线程安全的吧
public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { //在并发情况下，key 和 value 不支持为空的,为空报异常 if (key == null || value == null) throw new NullPointerException(); 获取key的hash值 int hash = spread(key.hashCode()); // 用来记录所在table数组中的桶的中链表的个数，后面会用于判断是否链表过长需要转红黑树 int binCount = 0; // for循环，直到put成功插入数据才会跳出 for (Node整个put方法的流程：
当添加key、value键值对的时候，首先会去判断保存这些键值对的数组是不是初始化了，如果没有的话就初始化数组，然后通过计算hash值来确定放在数组的哪个位置。如果这个位置为空则直接添加，如果不为空的话，则取出这个节点来，如果取出来的节点的hash值是MOVED(-1)的话，则表示当前正在对这个数组进行扩容，复制到新的数组，则当前线程也去帮助复制。
最后一种情况就是，如果这个节点，不为空，也不在扩容，则通过synchronized来加锁，进行添加 *** 作。然后判断当前取出的节点位置存放的是链表还是树，如果是链表的话，则遍历整个链表，直到取出来的节点的key来个要放的key进行比较，如果key相等，并且key的hash值也相等的话，则说明是同一个key，则覆盖掉value，否则的话则添加到链表的末尾。如果是树的话，则调用putTreeval方法把这个元素添加到树中去，最后在添加完成之后，会判断在该节点处共有多少个节点（注意是添加前的个数），如果达到8个以上了的话，则调用treeifyBin方法来尝试将链表转为树，或者扩容数组。
initTable方法我们看一下当数组为空的时候是怎么初始化的
private final Node
helpTransfer方法当一个线程要对table中元素进行 *** 作的时候，如果检测到节点的hash值为MOVED的时候，就会调用helpTransfer方法，在helpTransfer中再调用transfer方法来帮助完成数组的扩容
final Node
transfer方法transfer这个方法我们一直说是在扩容，要明确的一点是与其说是帮助扩容，其实更准确的说应该是帮助迁移元素，扩容时table容量变为原来的两倍，并把部分元素迁移到其它桶中。该方法发生扩容并且迁移数据。
还有就是扩容是第一次初始化表，只能由一个线程完成，而其他线程帮助迁移元素到新数组当中。
看源码之前先看一下扩容迁移数据的流程图，熟悉一下流程：
下面分析源码：
private final void transfer(Node
以 runBit =0 为例，迁移后的新数组链表示意图：
addCount方法
addCount 方法的主要作用：
对table的长度+1。无论是通过修改 baseCount，还是通过使用CounterCell。当CounterCell被初始化了，就优先使用他，不再使用 baseCount。
检查是否需要扩容，或者是否正在扩容。如果需要扩容，就调用扩容方法，如果正在扩容，就帮助其扩容。
//x为1,check表示链表上的元素个数 private final void addCount(long x, int check) { CounterCell[] as; long b, s; //判断counterCells是否为空. 为空有可能竞争的线程非常少,可通过cas *** 作尝试修改 baseCount 变量，对这个变量进行原子累加 *** 作 //如果cas失败说明存在竞争，这个时候不能再采用baseCount来累加，而是通过counterCells来记录 if ((as = counterCells) != null || !U.compareAndSwapLong(this, baseCOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; //是否冲突标识，标记为true，默认没有冲突 boolean uncontended = true; //条件一：as == null || (m = as.length - 1) < 0 //若数组为空，进fullAddCount方法 //条件二：(a = as[ThreadLocalRandom.getProbe() & m]) == null //表示方法会给当前线程生成一个随机数，然后用随机数与数组长度取模，计算它所在的表格。若当前线程所分配到的表格为空，需要当前线程进入fullAddCount方法 //条件三：!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) //若CounterCell不为空，且线程所在表格不为空，则尝试CAS修改表格对应的value值加1。 //若修改成功，则执行下一步，否则把uncontended值设为fasle，说明产生了竞争，然后进fullAddCount方法 if (as == null || (m = as.length - 1) < 0 || (a = as[ThreadLocalRandom.getProbe() & m]) == null || !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { //执行循环让当前线程一定把1加成功，这个会在稍后讲 fullAddCount(x, uncontended); return; } //这里若是返回了，那后边怎么判断扩容？搞不懂 if (check <= 1) return; //统计容量大小，主要用于下边的逻辑，是否达到了扩容阈值。 s = sumCount(); } //这里用于检查是否需要扩容 if (check >= 0) { //tab：表示table //nt:表示nextTable //n:表示数组的长度 //sc：表示sizeCtl的临时值 NodefullAddCount方法该方法负责初始化CounterCells和更新计数，方法中第二个参数wasUncontended则表示是否不存在竞争，CAS失败之后调用该方法说明存在竞争所以传false。
private final void fullAddCount(long x, boolean wasUncontended) { int h; ///如果当前线程的随机数为0，则强制初始化一个值 if ((h = ThreadLocalRandom.getProbe()) == 0) { ThreadLocalRandom.localInit(); // force initialization// 初始化 //得到probe，用于counterCells数组寻址 h = ThreadLocalRandom.getProbe(); //把wasUncontended 设置为 true 表示不存在竞争 wasUncontended = true; } 冲突标志，若为true，表示可能需要扩容，以减少碰撞冲突 boolean collide = false; // True if last slot nonempty //死循环，表示一定要计数成功 for (;;) { CounterCell[] as; CounterCell a; int n; long v; //如果counterCells不为空，说明已经初始化了 if ((as = counterCells) != null && (n = as.length) > 0) { //当前线程所在的CounterCell对象为空 if ((a = as[(n - 1) & h]) == null) { if (cellsBusy == 0) { // Try to attach new Cell //先创建一个CounterCell对象，把x保存进去 CounterCell r = new CounterCell(x); // Optimistic create //当前线程占用锁 if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean created = false; try { // Recheck under lock CounterCell[] rs; int m, j; //条件三：rs[j = (m - 1) & h] == null //这里又做了一层判断，如果为空才进入 //因为存在一种情况就是上面第一次判断cellsBusy为0时， //可能有一个线程在CAS修改cellsBusy并且初始化完成了， //然后复位cellsBusy //此时另一个线程进入，已经是初始化过了，所以需要再次判断 if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) { //把新创建的对象r的元素放入对应下标的位置 rs[j] = r; //created=true表示创建成功 created = true; } } finally { //手动释放锁 cellsBusy = 0; } //创建成功，且上边的赋值成功，说明加1成功，退出循环 if (created) break; //说明存在竞争该位置已被其他线程放入了CounterCell,继续下次循环 continue; // Slot is now non-empty } } //若cellsBusy=1，说明有其它线程抢锁成功。或者若抢锁的CAS *** 作失败， //都会走到这里，到下面的h = ThreadLocalRandom.advanceProbe(h)中 //重新生成随机数，进行下次循环判断 collide = false; } //如果走到这，说明当前方法在被调用之前已经 CAS 失败过一次， //重新获取线程随机数，并把wasUncontended 设置为true，再循环一次 else if (!wasUncontended) // CAS already known to fail wasUncontended = true; // Continue after rehash //若wasUncontended为true表示不存在竞争，尝试cas修改，成功则退出，否则继续往下走 else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) break; //如果发生过扩容或者数组长度已经达到虚拟机最大可以核心数 //因为长度超限，则说明已经无法扩容，只能认为无碰撞。 else if (counterCells != as || n >= NCPU) collide = false; // At max size or stale //数组长度小于CPU核心数并且collide为false，就把collide改为true， //说明下次循环可能需要扩容 else if (!collide) collide = true; //到了这里，说明竞争激烈，数组容量不够大，需要进行扩容了 else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { try { if (counterCells == as) {// Expand table unless stale //创建一个容量为原来两倍的数组，再通过for循环进行数据迁移。 CounterCell[] rs = new CounterCell[n << 1]; //转移旧数组的值 for (int i = 0; i < n; ++i) rs[i] = as[i]; //把扩容后的对象赋值给counterCells。 counterCells = rs; } } finally { // 释放锁 cellsBusy = 0; } //认为扩容后，下次不会产生冲突了 collide = false; //当次扩容后，就不需要重新生成随机数了 continue; // Retry with expanded table } //上面失败都会到这里来 //这里重新生成一个随机数，下一次循环判断 h = ThreadLocalRandom.advanceProbe(h); } //若cellsBusy为0，说明不存在锁，线程都可以抢锁，若为1，表示已经有线程拿到了锁， //则其它线程不能抢占锁 //上面是counterCells数组已初始化的情况，下面是未初始化情况的处理 else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { //这一步是初始化CounterCells数组 boolean init = false; try { // Initialize table //重新检查下counterCells数组引用是否有变化 if (counterCells == as) { //初始化一个长度为2的数组 CounterCell[] rs = new CounterCell[2]; //根据当前线程的随机数值，计算下标，只有两个结果0或1，并初始化对象 rs[h & 1] = new CounterCell(x); //赋值给counterCells counterCells = rs; //初始化成功的标志 init = true; } } finally { //释放锁 cellsBusy = 0; } //若初始化成功，则说明当前加1的操作也已经完成了，则退出整个循环 if (init) break; } //到这里标志着线程竞争激烈，其它线程占据着counterCells数组, //尝试更新baseCount的值，若成功，也说明加1操作成功，则退出循环。。 else if (U.compareAndSwapLong(this, baseCOUNT, v = baseCount, v + x)) // 更新成功直接返回 break; // Fall back on using base } }最后到这里ConcurrentHashMap的put方法插入最主要的逻辑基本上都讲完了，相对于jdk1.7来说，jdk1.8ConcurrentHashMap代码实现相对复杂，但是锁的粒度降低了，效率也提高了。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）