ConcurrentHashMap源码学习笔记jdk1.7&1.8

ConcurrentHashMap源码学习笔记jdk1.7&1.8(一)基于JDK1.7的ConcurrentHashMap1.基本实现:由Segement数组和HashEntry组成,与HashMap相同都是数组+链表的结构,而每一个Segment元素存储的是HashEntry数组

(一)基于JDK1.7的ConcurrentHashMap

           1.基本实现:

                由Segement数组和HashEntry组成,与HashMap相同都是数组+链表的结构,而每一个Segment元素存储的是HashEntry数组+链表,这个和HashMap的数据存储结构一样

                Segement:

                 

    static final class Segment<K,V> extends ReentrantLock implements Serializable {

    private static final long serialVersionUID = 2249069246763182397L;
     // 真正存放数据的桶
      transient volatile HashEntry<K,V>[] table;
 
      transient int count;

      transient int modCount;

       transient int threshold;//扩容阈值

       final float loadFactor;//加载因子

  }

           hashEntry:

static final class HashEntry<K,V> {
        final int hash;
        // key值初始化后不能改变
        final K key;
        //volatile保证读到的数据为最新值
        volatile V value;
        //volatile保证读到的数据为最新的
        volatile HashEntry<K,V> next;
}

      关系如图所示:
ConcurrentHashMap源码学习笔记jdk1.7&1.8

      锁分段技术:HashTable效率低下的原因是多个线程竞争同一把锁,而ConcurrentHashMap把数据分成了一段一段的存储,每一段数据都分给一把锁,当一段数据被一个线程访问时,其他段的数据也能被其他线程所访问。

       2.ConcurrentHashMap实现存储和读取:

         1.put()

         

public V put(K key, V value) {
       Segment<K,V> s;
       if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        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);
}

  先通过key定位到Segement,之后再具体Segement进行具体的添加:

     先进行再散列:减少散列冲突,使元素均匀的分布在不同的Segment上,从而提高容器的存取效率
    通过散列算法定位: segments[(hash>>>segmentShift)&segementMask]

 

 1        final V put(K key, int hash, V value, boolean onlyIfAbsent) {
 2            HashEntry<K,V> node = tryLock() ? null :
 3                scanAndLockForPut(key, hash, value);
 4            V oldValue;
 5            try {
 6                HashEntry<K,V>[] tab = table;
 7                int index = (tab.length - 1) & hash;
 8                HashEntry<K,V> first = entryAt(tab, index);
 9                for (HashEntry<K,V> e = first;;) {
10                    if (e != null) {
11                        K k;
12                        if ((k = e.key) == key ||
13                            (e.hash == hash && key.equals(k))) {
14                            oldValue = e.value;
15                            if (!onlyIfAbsent) {
16                                e.value = value;
17                                ++modCount;
18                            }
19                            break;
20                        }
21                        e = e.next;
22                    }
23                    else {
24                        if (node != null)
25                            node.setNext(first);
26                        else
27                            node = new HashEntry<K,V>(hash, key, value, first);
28                        int c = count + 1;
29                        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
30                            rehash(node);
31                        else
32                            setEntryAt(tab, index, node);
33                        ++modCount;
34                        count = c;
35                        oldValue = null;
36                        break;
37                    }
38                }
39            } finally {
40                unlock();
41            }
42            return oldValue;
43        }

      首先尝试获取锁,如果获取失败则调用scanAndLockForPut()自旋获取锁。

   scanAndLockForPut

    

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
            HashEntry<K,V> first = entryForHash(this, hash);
            HashEntry<K,V> e = first;
            HashEntry<K,V> node = null;
            int retries = -1; // 定位节点时为负数
       //(1)
            while (!tryLock()) {
                HashEntry<K,V> f; // 首先在下面重新检查
                if (retries < 0) {
                    if (e == null) {
                        if (node == null) // 推测性地创建节点
                            node = new HashEntry<K,V>(hash, key, value, null);
                        retries = 0;
                    }
                    else if (key.equals(e.key))
                        retries = 0;
                    else
                        e = e.next;
                }
          //(2)
                else if (++retries > MAX_SCAN_RETRIES) {
                    lock();
                    break;
                }
                else if ((retries & 1) == 0 &&
                         (f = entryForHash(this, hash)) != first) {
                    e = first = f; // 如果Entry改变则重新遍历
                    retries = -1;
                }
            }
            return node;
        }

       

代码(1)尝试自旋获取锁。

代码(2)如果重试的次数达到了 MAX_SCAN_RETRIES 则改为阻塞锁获取,保证能获取成功。

    

 流程:

            1.首先尝试获取锁,如果获取失败说明与其它线程存在竞争,则调用scanAndLockForPut()

            2. 计算HashEntry的数组下标并定位到HashEntry链表的首节点

            3.遍历链表,如果不为空则比较key是否相同,相同则覆盖

            4.为空创建一个新的结点并添加到hash链的头部

            5.判断元素个数是否超过扩容阈值,超过则调用resize扩容

            6.最后释放锁

        2.get()

 

 1    public V get(Object key) {
 2        Segment<K,V> s; // manually integrate access methods to reduce overhead
 3        HashEntry<K,V>[] tab;
 4        int h = hash(key);
 5        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
 6        if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
 7            (tab = s.table) != null) {
 8            for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
 9                     (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
10                 e != null; e = e.next) {
11                K k;
12                if ((k = e.key) == key || (e.hash == h && key.equals(k)))
13                    return e.value;
14            }
15        }
16        return null;
17    }

 

     通过key的hash值,调用UNSAFE.getObjectVolatile方法,由于是volatile版本,可以实现线程之间的可见性(遵循happend-before原则),故可以从最新的segments数组中获取该key所在的segment;
       然后根据key的hash值,获取该segment内部的哈希表table数组的下标,从而获取该key所在的链表的头结点,然后从链表头结点开始遍历该链表,最终如果没有找到,则返回null或者找到对应的链表节点,返回该链表节点的value。

      3.size:

         一开始并不对Segment加锁,而是直接尝试将所有的Segment元素中的count相加,这样执行两次,然后将两次的结果对比,如果两次结果相等则直接返回;而如果两次结果不同,则再将所有Segment加锁(将所有segement的put,remove,clean都锁住),然后再执行统计得到对应的size值

                 

(二)基于JDK1.8的ConcurrentHashMap

1.重要概念: 采用Node+CAS+Synchronized来保证并发安全的实现。

2.put方法

       

public V put(K key, V value) {
    return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode()); //两次hash,减少hash冲突,可以均匀分布
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) { //对这个table进行迭代
        Node<K,V> f; int n, i, fh;
        //这里就是上面构造方法没有进行初始化,在这里进行判断,为null就调用initTable进行初始化,属于懒汉模式初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果i位置没有数据,就直接无锁插入
            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;
            //如果以上条件都不满足,那就要进行加锁操作,也就是存在hash冲突,锁住链表或者红黑树的头结点
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { //表示该节点是链表结构
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            //这里涉及到相同的key进行put就会覆盖原先的value
                            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) { //如果链表的长度大于8时就会进行红黑树的转换
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);//统计size,并且检查是否需要扩容
    return null;
}

流程:

      1.先对key的hashcode进行第二次hash,减少哈希冲突

      2.如果table没有初始化,则调用initTable进行table的初始化

      3.如果table的i位置没有数据,则直接进行无锁插入(调用casTabAt)

      4.判断是否在进行扩容,如果是则协助扩容

      5.以上条件都不满足,则使用synchronized进行加锁操作

     6.如果是链表则进行链表的插入,如果是红黑树则进行红黑树的插入操作

     7.如果链表的个数是否大于阈值,大于则转换为红黑树

     8.最后调用addCount,统计size,检查否需要扩容

initable:

      

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                Thread.yield(); 
                //正在初始化时将sizeCtl设为-1
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                /*
                unsafe.compareAndSwapInt这个方法,这个方法有四个参数,其中第一个参数为需要改变的对象,第二个为偏移量(即之前求出来的valueOffset的值),第三个参数为期待的值,第四个为更新后的值
                */
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;  //DEFAULT_CAPACITY为16
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);   //扩容阈值为新容量的0.75倍
                    }
                } finally {
                    sizeCtl = sc;   //扩容保护
                }
                break;
            }
        }
        return tab;
    }

 

    table的初始化时未加锁的,当腰初始化的时候调用compareAndSwapInt将sizeCtl视为-1,sizeCtl由votaile修饰,保证了可见性,因而别的线程再调用次方法时会执行Thread.yield()

addCount

   

     3.扩容实现:

            1.扩容时机:

                1.调用put后会调用addCount(),内部检查sizeCtl是否需要扩容

                 2.链表转红黑树时如果table容量小于64

                 3.调用putAll()之类一次性加入大量元素

           2.总体思想:分工合作,多个线程一个协作扩容,将表拆分,让每个线程处理自己的区间

            å¨è¿éæå¥å¾çæè¿°

     3.重要属性:

         1. sizeCtl:控制标志符

                当前未初始化:

                      =0 未指定初始化容量

                      >0由指定的初始化容量计算,再寻找最近的2的幂次

                初始化中:

                     -1 table正在初始化

                     -N  表示正有N-1个线程执行扩容操作

                初始化完成:

                     =table.length*0.75

         2.static变量:

// 用于生成每次扩容都唯一的生成戳的数,最小是6。
private static int RESIZE_STAMP_BITS = 16;
 
// 最大的扩容线程的数量
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
 
// 移位量,把生成戳移位后保存在sizeCtl中当做扩容线程计数的基数,相反方向移位后能够反解出生成戳
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

       3.

       4.

       5.transfer方法

      

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        // 每核处理的量小于16,则强制赋值16
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        if (nextTab == null) {            // initiating
            try {
                @SuppressWarnings("unchecked")
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];        //构建一个nextTable对象,其容量为原来容量的两倍
                nextTab = nt;
            } catch (Throwable ex) {      // try to cope with OOME
                sizeCtl = Integer.MAX_VALUE;
                return;
            }
            nextTable = nextTab;
            transferIndex = n;
        }
        int nextn = nextTab.length;
        // 连接点指针,用于标志位(fwd的hash值为-1,fwd.nextTable=nextTab)
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        // 当advance == true时,表明该节点已经处理过了
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            // 控制 --i ,遍历原hash表中的节点
            while (advance) {
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                // 用CAS计算得到的transferIndex
                else if (U.compareAndSwapInt
                        (this, TRANSFERINDEX, nextIndex,
                                nextBound = (nextIndex > stride ?
                                        nextIndex - stride : 0))) {
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            if (i < 0 || i >= n || i + n >= nextn) {
                int sc;
                // 已经完成所有节点复制了
                if (finishing) {
                    nextTable = null;
                    table = nextTab;        // table 指向nextTable
                    sizeCtl = (n << 1) - (n >>> 1);     // sizeCtl阈值为原来的1.5倍
                    return;     // 跳出死循环,
                }
                // CAS 更扩容阈值,在这里面sizectl值减一,说明新加入一个线程参与到扩容操作
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
            // 遍历的节点为null,则放入到ForwardingNode 指针节点
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            // f.hash == -1 表示遍历到了ForwardingNode节点,意味着该节点已经处理过了
            // 这里是控制并发扩容的核心
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                // 节点加锁
                synchronized (f) {
                    // 节点复制工作
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        // fh >= 0 ,表示为链表节点
                        if (fh >= 0) {
                            // 构造两个链表  一个是原链表  另一个是原链表的反序排列
                            int runBit = fh & n;
                            Node<K,V> lastRun = f;
                            for (Node<K,V> p = f.next; p != null; p = p.next) {
                                int b = p.hash & n;
                                if (b != runBit) {
                                    runBit = b;
                                    lastRun = p;
                                }
                            }
                            if (runBit == 0) {
                                ln = lastRun;
                                hn = null;
                            }
                            else {
                                hn = lastRun;
                                ln = null;
                            }
                            for (Node<K,V> p = f; p != lastRun; p = p.next) {
                                int ph = p.hash; K pk = p.key; V pv = p.val;
                                if ((ph & n) == 0)
                                    ln = new Node<K,V>(ph, pk, pv, ln);
                                else
                                    hn = new Node<K,V>(ph, pk, pv, hn);
                            }
                            // 在nextTable i 位置处插上链表
                            setTabAt(nextTab, i, ln);
                            // 在nextTable i + n 位置处插上链表
                            setTabAt(nextTab, i + n, hn);
                            // 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
                            setTabAt(tab, i, fwd);
                            // advance = true 可以执行--i动作,遍历节点
                            advance = true;
                        }
                        // 如果是TreeBin,则按照红黑树进行处理,处理逻辑与上面一致
                        else if (f instanceof TreeBin) {
                            TreeBin<K,V> t = (TreeBin<K,V>)f;
                            TreeNode<K,V> lo = null, loTail = null;
                            TreeNode<K,V> hi = null, hiTail = null;
                            int lc = 0, hc = 0;
                            for (Node<K,V> e = t.first; e != null; e = e.next) {
                                int h = e.hash;
                                TreeNode<K,V> p = new TreeNode<K,V>
                                        (h, e.key, e.val, null, null);
                                if ((h & n) == 0) {
                                    if ((p.prev = loTail) == null)
                                        lo = p;
                                    else
                                        loTail.next = p;
                                    loTail = p;
                                    ++lc;
                                }
                                else {
                                    if ((p.prev = hiTail) == null)
                                        hi = p;
                                    else
                                        hiTail.next = p;
                                    hiTail = p;
                                    ++hc;
                                }
                            }
                            // 扩容后树节点个数若<=6,将树转链表
                            ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                                    (hc != 0) ? new TreeBin<K,V>(lo) : t;
                            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                    (lc != 0) ? new TreeBin<K,V>(hi) : t;
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

4.get方法:

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode()); //计算两次hash
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {//读取首节点的Node元素
        if ((eh = e.hash) == h) { //如果该节点就是首节点就返回
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        //hash值为负值表示正在扩容,这个时候查的是ForwardingNode的find方法来定位到nextTable来
        //查找,查找到就返回
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode,那就往下遍历
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}
  1. 计算hash值,定位到该table索引位置,如果是首节点符合就返回
  2. 如果遇到扩容的时候,会调用标志正在扩容节点ForwardingNode的find方法,查找该节点,匹配就返回
  3. 以上都不符合的话,就往下遍历节点,匹配就返回,否则最后就返回null

5.size操作:元素总数 = baseCount + sum(CounterCell)

public int size() {
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
            (int)n);
}
final long sumCount() {
    CounterCell[] as = counterCells; CounterCell a; //变化的数量
    long sum = baseCount;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

 

今天的文章ConcurrentHashMap源码学习笔记jdk1.7&1.8分享到此就结束了,感谢您的阅读。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/81816.html

(0)
编程小号编程小号

相关推荐

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注