Java 并发集合类（未完待续）

作者: 兴厚 | 来源:发表于2020-04-16 22:20 被阅读0次

基于 openJDK 11。 IDEA 2020.1 ， idea设置里面 Debugger ==> stepping ==>> Do not step into the classed, 取消对java.* javax.* 类的debug限制，就能debug源码了。默认这些类的源码debug信息被去除了。

Map

HashMap

首先是一个 Node 的对象数组，每个 Node 对象里面保留有下一个节点的指针, Node结构

// 这个结构也是红黑树的节点结构
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;  // hash值
    final K key;     
    V value;
    Node<K,V> next;  // 链表使用
    
    // .....
}

重点看两处，resize 和 hash 冲突的时候处理。首先看 hash 冲突处理

final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
               boolean evict) {
    Node<K,V>[] tab; Node<K,V> p; int n, i;
    if ((tab = table) == null || (n = tab.length) == 0)
        n = (tab = resize()).length;
    if ((p = tab[i = (n - 1) & hash]) == null)
        tab[i] = newNode(hash, key, value, null);
    else {
        Node<K,V> e; K k;
        if (p.hash == hash && ((k = p.key) == key || (key != null && key.equals(k))))
            e = p; // 这种情况表明 key 重复，直接到最后的处理
        else if (p instanceof TreeNode)
            // 表明这个节点属于红黑树的节点，走红黑树的插入方式，这里就不看了
            e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value); 
        else {
            for (int binCount = 0; ; ++binCount) {
                if ((e = p.next) == null) {
                    // 每次往尾部插入数据
                    p.next = newNode(hash, key, value, null);
                    // 当链表节点数太多的时候，就会走红黑树数据结构的数据插入，接下来看 treeifybin 方法
                    if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
                        treeifyBin(tab, hash);
                    break;
                }
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    break;
                p = e;
            }
        }
        if (e != null) { // existing mapping for key
            V oldValue = e.value;
            if (!onlyIfAbsent || oldValue == null)
                e.value = value;
            afterNodeAccess(e);
            return oldValue;
        }
    }
    ++modCount;
    if (++size > threshold)
        resize();
    afterNodeInsertion(evict);
    return null;
}


 final void treeifyBin(Node<K,V>[] tab, int hash) {
     int n, index; Node<K,V> e;
     if (tab == null || (n = tab.length) < MIN_TREEIFY_CAPACITY)
         // 如果冲突是由于hash表太小了，就重新扩容
         resize();
     else if ((e = tab[index = (n - 1) & hash]) != null) {
         TreeNode<K,V> hd = null, tl = null;
         do {
             // 这个就是将普通的Node转换成红黑树的 TreeNode
             TreeNode<K,V> p = replacementTreeNode(e, null);
             // 双向链表赋值
             if (tl == null)
·                 hd = p;
             else {
                 p.prev = tl;
                 tl.next = p;
             }
             tl = p;
         } while ((e = e.next) != null);
         if ((tab[index] = hd) != null)
             // 开始整理成红黑树结构
             hd.treeify(tab);
     }
 }

这里总结下来就几点

单向链表 -> 红黑树
链表插入都是尾节点插入

接下来看 resize 操作

final Node<K,V>[] resize() {
    Node<K,V>[] oldTab = table;
    int oldCap = (oldTab == null) ? 0 : oldTab.length;
    int oldThr = threshold;
    int newCap, newThr = 0;
    // 这里省略一些不重要的判断，赋值 .......
    threshold = newThr;
    @SuppressWarnings({"rawtypes","unchecked"})
    Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
    table = newTab;
    if (oldTab != null) {
        for (int j = 0; j < oldCap; ++j) {
            Node<K,V> e;
            if ((e = oldTab[j]) != null) {
                oldTab[j] = null;
                if (e.next == null)
                    // 普通节点重新计算数组下标
                    newTab[e.hash & (newCap - 1)] = e;
                else if (e instanceof TreeNode)
                    // 红黑树节点迁移
                    ((TreeNode<K,V>)e).split(this, newTab, j, oldCap);
                else { // preserve order
                    // 单向链表节点保护数据顺序迁移
                    Node<K,V> loHead = null, loTail = null;
                    Node<K,V> hiHead = null, hiTail = null;
                    Node<K,V> next;
                    do {
                        next = e.next;
                        // 这一步位运算是为了当数组扩大时，还能保证命中的下标一致。只能说，妙不可言。
                        // 计算的时候时 （n - 1）& hash ，这里迁移的时候，hash & n. 
                        if ((e.hash & oldCap) == 0) {
                            if (loTail == null)
                                loHead = e;
                            else
                                loTail.next = e;
                            loTail = e;
                        }
                        else {
                            if (hiTail == null)
                                hiHead = e;
                            else
                                hiTail.next = e;
                            hiTail = e;
                        }
                    } while ((e = next) != null);
                    if (loTail != null) {
                        loTail.next = null;
                        newTab[j] = loHead;
                    }
                    if (hiTail != null) {
                        hiTail.next = null;
                        newTab[j + oldCap] = hiHead;
                    }
                }
            }
        }
    }
    return newTab;
}

resize 的时候会进行 reindex 操作（重新定位数组下标），妙不可言，更多设计的精妙之处参考这篇博文.

我们再来看看在高并发的情况下出现经典状况的原因吧:

有些key的值丢失
死循环问题 (只出现在 jdk7 及以前的版本，自jdk8 以来就不会有这种情况了)

问题1：问题出现的代码块（在putVal方法中）

for (int binCount = 0; ; ++binCount) {
    if ((e = p.next) == null) {
    // WARNING: 每次往尾部插入数据， 这里就会出现后面的覆盖前面的值的情况。导致数据丢失
    p.next = newNode(hash, key, value, null);
    if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
    treeifyBin(tab, hash);
    break;
    }
    if (e.hash == hash &&
    ((k = e.key) == key || (key != null && key.equals(k))))
    break;
    p = e;
}

问题2：常见于 jdk7 及更低的版本（resize 的过程）

 void transfer(Entry[] newTable) {
     Entry[] src = table;
     int newCapacity = newTable.length;
     for (int j = 0; j < src.length; j++) {
         Entry<K,V> e = src[j];
         // 看到这段代码我是有点懵逼的，他把原先的链表遍历，然后逆序插入新的节点。我觉得太煞笔了。
         // 1. 每个节点都重新计算 index 位置时没必要的，应为节点的hash是一样的， hash & (n-1) 的值必然也是一样。
         // 2. 为什么需要遍历每个节点，直接拿到原先的头，然后赋值给newTable[i] 不就行了吗 ？？？
         // 言归正传， 这里的代码在多线程情况下肯定时会出现死循环的（逆序插入必死）。当然这是 jdk7 的代码，在jdk8 + 就不是这样实现了。
         if (e != null) {
             src[j] = null;
             do {
                 Entry<K,V> next = e.next;
                 int i = indexFor(e.hash, newCapacity);
                 e.next = newTable[i];
                 newTable[i] = e;
                 e = next;
             } while (e != null);
         }
     }
 }

// 看看 jdk11 的 resize（）实现
Node<K,V> loHead = null, loTail = null;
Node<K,V> hiHead = null, hiTail = null;
Node<K,V> next;
do {
    next = e.next;
    // 这一步位运算是为了当数组扩大时，还能保证命中的下标一致。只能说，妙不可言。
    // 计算的时候时 （n - 1）& hash ，这里迁移的时候，hash & n => (j + oldCap)
    if ((e.hash & oldCap) == 0) {
        if (loTail == null)
            loHead = e;
        else
            loTail.next = e;
        loTail = e;
    }
    // 这里也会遍历所有的节点，我觉得不应该只需要拿到头节点计算hash值不就可以了吗，然后只迁移头节点不就可以了吗？？？？？？ 不过好在这里不是逆序迁移数据了（汗）.
    else {
        if (hiTail == null)
            hiHead = e;
        else
            hiTail.next = e;
        hiTail = e;
    }
} while ((e = next) != null);
if (loTail != null) {
    loTail.next = null;
    newTab[j] = loHead;
}
if (hiTail != null) {
    hiTail.next = null;
    newTab[j + oldCap] = hiHead;
}

关于更详细的画图解释，这里就不画图了 (让各位看官失望了)。参考这篇文章。

我有时间会把我的猜想实现方式试验一下的，实验完毕我再贴上代码（挖坑） .......................

ConcurrentHashMap

这个map以我现在的能力有些代码还是没怎么看懂，回头得看看 hotspot 源码再看看底层原理（挖坑+1），当然网上也有很多解析CAS 的文章，还是可以学习学习的，这里就只根据方法名和注释来理解一下把。

依然是上面的两个经典方法. 先说说 Unsafe 类源码的一些要点吧。

jdk 11 使用 compareAndSetxxx 代替了 jdk8 的compareAndSwap 方法。
compareAndSetxxx 等方法的实现原理一句话解释就是——判断当前线程能否执行的依据是拿当前线程的值去和主内存的值去比较，如果相等，则更新数据，更新成功则返回true，否则 false。这里没有做 ABA 问题的检测（我觉得都是加法的操作确实没必要做ABA浪费性能）。这些方法都是接受4个参数，如下 :
```
// 1.比较的对象，2.对象的移位（找到具体值在对象储存内存中的位置），3.比较值，4.赋予的新值
// 比较的就是 3 的值和 2定位到对象内存中的值是否相等，相等则更新返回。
compareAndSetObject(Object o, long offset,Object expected,Object x)
```

hash 冲突的时候，synchorized 锁住了定位到的头节点进行链表添加操作，这就没啥分析的必要了（得看synchorized 的实现原理），这里得解释一下，在jdk8 以后， synchorized 实现方式改变了，有锁升级的过程，不像Lock的实现类，一上来就锁总线，让线程休眠阻塞，其实 map 这种最多算是 CPU 密集型任务，所以它的并发应该就应该是让线程空转等待（CAS），而不是jdk7的线程休眠阻塞。ConcurrentHashMap 里为了保证线程的可见性，大部分变量都加了 volatile 修饰符，这个也会转成 cpu 的一个指令执行，效率很高。这里有一个关键性的变量得说明一下:

/**
 * Table initialization and resizing control.  When negative, the
 * table is being initialized or resized: -1 for initialization,
 * else -(1 + the number of active resizing threads).  Otherwise,
 * when table is null, holds the initial table size to use upon
 * creation, or 0 for default. After initialization, holds the
 * next element count value upon which to resize the table.
*/
private transient volatile int sizeCtl;

翻译一下就是，为了保证内部的储存结构 table 的有些操作只能让单线程进行，通过这个变量控制，为-1 表示table正在扩容， -（N + 1）表示有 N 个线程在执行 resize 操作，还有些为正值的操作，还没怎么看明白。具体的代码这里就补贴了，有一篇讲的不错的，分享一下

Java 并发集合类（未完待续）

Map

HashMap

ConcurrentHashMap

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 并发集合类 （未完待续）

Map

HashMap

ConcurrentHashMap

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java 并发集合类（未完待续）