学习ConcurrentHashMap并发写机制

时间 2020-03-16

标签学习 concurrenthashmap 并发机制栏目 Java 繁體版

原文原文链接

1. 前言

上篇文章讲了 Unsafe 类中 CAS 的实现，实际上是在为这篇文章打基础。不太熟悉的小伙伴请移步Unsafe 中 CAS 的实现。本篇文章主要基于 OpenJDK8 来作源码解析。java

2. 源码

ConcurrentHashMap 基于 HashMap 实现。数组

JDK1.7 和 JDK1.8 做为并发容器在实现上是有差异的。JDK1.7 经过 Segment 分段锁实现，而 JDK1.8 经过 CAS+synchronized 实现。安全

2.1 ConcurrentHashMap 几个重要方法

在 ConcurrentHashMap 中使用了 unSafe 方法，经过直接操做内存的方式来保证并发处理的安全性，使用的是硬件的安全机制。网络

private static final sun.misc.Unsafe U;
    private static final long SIZECTL;
    private static final long TRANSFERINDEX;
    private static final long BASECOUNT;
    private static final long CELLSBUSY;
    private static final long CELLVALUE;
    private static final long ABASE;
    private static final int ASHIFT;

    static {
        try {
            U = sun.misc.Unsafe.getUnsafe();
            Class<?> k = ConcurrentHashMap.class;
            SIZECTL = U.objectFieldOffset
                (k.getDeclaredField("sizeCtl"));
            TRANSFERINDEX = U.objectFieldOffset
                (k.getDeclaredField("transferIndex"));
            BASECOUNT = U.objectFieldOffset
                (k.getDeclaredField("baseCount"));
            CELLSBUSY = U.objectFieldOffset
                (k.getDeclaredField("cellsBusy"));
            Class<?> ck = CounterCell.class;
            CELLVALUE = U.objectFieldOffset
                (ck.getDeclaredField("value"));
            Class<?> ak = Node[].class;
            ABASE = U.arrayBaseOffset(ak);
            int scale = U.arrayIndexScale(ak);
            if ((scale & (scale - 1)) != 0)
                throw new Error("data type scale not a power of two");
            ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
        return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }

    // CAS 将Node插入bucket
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                        Node<K,V> c, Node<K,V> v) {
        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    }

    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }

2.2 put()流程

仍是老规矩，先上流程图帮助阅读源码。并发

主体源码以下:spa

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());
        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) { // 若是bucket==null，即没有hash冲突，CAS插入
                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);
            // 不然，存在hash冲突
            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;
                                // 遍历过程当中出现相同key直接覆盖
                                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;
    }

put 操做过程以下:线程

若是没有初始化就先调用 initTable（）方法来进行初始化过程
若是没有 hash 冲突就直接 CAS 插入
若是还在进行扩容操做就先进行扩容
若是存在 hash 冲突，就加锁来保证线程安全，这里有两种状况，一种是链表形式就直接遍历到尾端插入，一种是红黑树就按照红黑树结构插入
最后一个若是该链表的数量大于阈值 8，就要先转换成黑红树的结构，break 再一次进入循环
若是添加成功就调用 addCount（）方法统计 size，而且检查是否须要扩容

2.3 ConcurrentHashMap 的存储结构

下边的示意图来自网络设计

3. 结语

本文只分析了 ConcurrentHashMap 的 put() 方法，并无分析 get()、扩容、删除节点等方法。主要目的是初步了解 ConcurrentMap 确保并发写的设计思路。至此，本篇文章结束，感谢你们的阅读！欢迎你们关注公众号【当我赶上你】。code