ConcurrentHashMap高并发实现原理

ConcurrentHashMap结构

在JDK7中ConcurrentHashMap是基于细粒度分离锁实现的,其结果大致如下:

在JDK7中ConcurrentHashMap由多个Segment组成,每个Segment都继承了ReentrantLock,是一把重入锁。

JDK8 中完全重写了ConcurrentHashmap,ConcurrentHashmap实现上和原来的分段式存储有很大的区别,本文讲解的JDK8中ConcurrentHashMap与JDK8的HashMap有相通之处,底层依然由“数组”+链表+红黑树实现,利用全新的CAS算法保证线程安全。Segment虽保留,但已经简化属性,仅仅是为了兼容旧版本。

CAS算法:unsafe.compareAndSwapInt(this, valueOffset, expect, update);  CAS(Compare And Swap),意思是如果valueOffset位置包含的值与expect值相同,则更新valueOffset位置的值为update,并返回true,否则不更新,返回false。

CAS作为知名无锁算法,那ConcurrentHashMap就没用锁了么?当然不是,hash值相同的链表的头结点还是会synchronized上锁

重要属性

我们重点关注几个出镜率比较高的属性

    private transient volatile int sizeCtl;

 sizeCtl是控制标识符,volatile关键字保证了其线程可见性,不同的值表示不同的意义。

负数代表正在进行初始化或扩容操作
-1代表正在初始化
-N 表示有N-1个线程正在进行扩容操作
正数或0代表hash表还没有被初始化,这个数值表示初始化或下一次进行扩容的大小,类似于扩容阈值。它的值始终是当前ConcurrentHashMap容量的0.75倍,这与loadfactor是对应的。实际容量>=sizeCtl,则扩容。

    transient volatile Node<K,V>[] table;

盛装Node元素的数组,在第一次插入时被初始化(延迟初始化,并没有在构造函数执行时进行初始化),它的大小总是2的整数次幂 

//链表转树阈值,大于8时    
static final int TREEIFY_THRESHOLD = 8;
//树转链表阈值,小于等于6时(仅在扩容transfer时才可能树转链表)
static final int UNTREEIFY_THRESHOLD = 6;

重要内部类

Node是最核心的内部类,它包装了key-value键值对,所有插入ConcurrentHashMap的数据都包装在这里面。

相关注意的点在源码中用注释说明:

  
static class Node implements Map.Entry {
        final int hash;//final 修饰
        final K key; //final 修饰
        volatile V val; // volatile 保证线程可见性
        volatile Node next; //volatile保证线程可见性
    Node(int hash, K key, V val, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.val = val;
        this.next = next;
    }

    public final K getKey()       { return key; }
    public final V getValue()     { return val; }
    public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
    public final String toString(){ return key + "=" + val; }
   //不允许直接改变value的值
    public final V setValue(V value) {
        throw new UnsupportedOperationException();
    }
   //增加find方法辅助get
    public final boolean equals(Object o) {
        Object k, v, u; Map.Entry<?,?> e;
        return ((o instanceof Map.Entry) &amp;&amp;
                (k = (e = (Map.Entry<?,?>)o).getKey()) != null &amp;&amp;
                (v = e.getValue()) != null &amp;&amp;
                (k == key || k.equals(key)) &amp;&amp;
                (v == (u = val) || v.equals(u)));
    }

TreeNode是另外一个核心的数据结构。当链表长度过长的时候,会转换为TreeNode。但是与HashMap不相同的是,它并不是直接转换为红黑树,而是把这些结点包装成TreeNode放在TreeBin对象中,由TreeBin完成对红黑树的包装。


    /**
     * Nodes for use in TreeBins 链表>8,才可能转为TreeNode
     */ 
    static final class TreeNode extends Node {
        TreeNode parent;  // red-black tree links
        TreeNode left;
        TreeNode right;
        TreeNode prev;    // needed to unlink next upon deletion
        boolean red;
    TreeNode(int hash, K key, V val, Node<K,V> next,
             TreeNode<K,V> parent) {
        super(hash, key, val, next);
        this.parent = parent;
    }

    Node<K,V> find(int h, Object k) {
        return findTreeNode(h, k, null);
    }

    /**
     * Returns the TreeNode (or null if not found) for the given key
     * starting at given root.
     */
    final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
        if (k != null) {
            TreeNode<K,V> p = this;
            do  {
                int ph, dir; K pk; TreeNode<K,V> q;
                TreeNode<K,V> pl = p.left, pr = p.right;
                if ((ph = p.hash) > h)
                    p = pl;
                else if (ph < h)
                    p = pr;
                else if ((pk = p.key) == k || (pk != null &amp;&amp; k.equals(pk)))
                    return p;
                else if (pl == null)
                    p = pr;
                else if (pr == null)
                    p = pl;
                else if ((kc != null ||
                          (kc = comparableClassFor(k)) != null)
                         (dir = compareComparables(kc, k, pk)) != 0)
                    p = (dir < 0) ? pl : pr;
                else if ((q = pr.findTreeNode(h, k, kc)) != null)
                    return q;
                else
                    p = pl;
            } while (p != null);
        }
        return null;
    }
}

TreeNode继承自ConcurrentHashMap的Node类而并非HashMap中的继承自LinkedHashMap.Entry类,也就是说TreeNode带有next指针,这样做的目的是方便基于TreeBin的访问。

TreeBin并不负责包装用户的key、value信息,而是包装的很多TreeNode节点。它代替了TreeNode的根节点,也就是说在实际的ConcurrentHashMap“数组”中,存放的是TreeBin对象,而不是TreeNode对象,这是与HashMap的区别。另外这个类还带有了读写锁。

    
static final class TreeBin extends Node {
        TreeNode root;
        volatile TreeNode first;
        volatile Thread waiter;
        volatile int lockState;
        // values for lockState
        static final int WRITER = 1; // set while holding write lock
        static final int WAITER = 2; // set when waiting for write lock
        static final int READER = 4; // increment value for setting read lock
......
        /**
         * Creates bin with initial set of nodes headed by b.
         * 构造红黑树
         */
        TreeBin(TreeNode b) {
            super(TREEBIN, null, null, null);
            this.first = b;
            TreeNode r = null;
            for (TreeNode x = b, next; x != null; x = next) {
                next = (TreeNode)x.next;
                x.left = x.right = null;
                if (r == null) {
                    x.parent = null;
                    x.red = false;
                    r = x;
                }
                else {
                    K k = x.key;
                    int h = x.hash;
                    Class kc = null;
                    for (TreeNode p = r;;) {
                        int dir, ph;
                        K pk = p.key;
                        if ((ph = p.hash) > h)
                            dir = -1;
                        else if (ph < h)
                            dir = 1;
                        else if ((kc == null &&
                                  (kc = comparableClassFor(k)) == null) ||
                                 (dir = compareComparables(kc, k, pk)) == 0)
                            dir = tieBreakOrder(k, pk);
                            TreeNode xp = p;
                        if ((p = (dir <= 0) ? p.left : p.right) == null) {
                            x.parent = xp;
                            if (dir <= 0)
                                xp.left = x;
                            else
                                xp.right = x;
                            r = balanceInsertion(r, x);
                            break;
                        }
                    }
                }
            }
            this.root = r;
            assert checkInvariants(root);
        }
......
    }

ForwardingNode一个用于连接两个table的节点类,它包含一个nextTable指针,用于指向下一张表。而且这个节点的key value next指针全部为null,它的hash值为-1。生命周期:仅存活于扩容操作且bin不为null时,一定会出现在每个bin的首位。

  
/**
     * A node inserted at head of bins during transfer operations.
     */
    static final class ForwardingNode extends Node {
        final Node[] nextTable;
        ForwardingNode(Node[] tab) {
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }
//从nextTable里进行查询节点,而不是以自身为头结点进行查找
        Node find(int h, Object k) {
            // loop to avoid arbitrarily deep recursion on forwarding nodes
            outer: for (Node[] tab = nextTable;;) {
                Node e; int n;
                if (k == null || tab == null || (n = tab.length) == 0 ||
                    (e = tabAt(tab, (n - 1) & h)) == null)
                    return null;
                for (;;) {
                    int eh; K ek;
                    if ((eh = e.hash) == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                    if (eh < 0) {
                        if (e instanceof ForwardingNode) {
                            tab = ((ForwardingNode)e).nextTable;
                            continue outer;
                        }
                        else
                            return e.find(h, k);
                    }
                    if ((e = e.next) == null)
                        return null;
                }
            }
        }
    }

三个核心方法

ConcurrentHashMap定义了三个原子操作,用于对指定位置的节点进行操作。正是这些原子操作保证了ConcurrentHashMap的线程安全。


//获取i位置上的Node节点  
static final  Node tabAt(Node[] tab, int i) {
        return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }
//利用CAS算法设置i位置上的结点Node,之所以能实现并发是因为它指定了原来这个节点的值是多少
//在CAS算法中会比较内存中的值与你指定的值是否相等,如果相等才接受你的修改,否则拒绝你的修改
//因为当前线程中的值并不是最新的值,这种修改可能会覆盖其他线程的修改结果
    static final  boolean casTabAt(Node[] tab, int i,
                                        Node c, Node v) {
        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    }
//利用volatile方法设置节点位置i的值,仅在上锁区被调用
    static final  void setTabAt(Node[] tab, int i, Node v) {
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }

初始化initTable

对于ConcurrentHashMap来说,调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入第一个元素的时候发生的。

  
/**
     * Initializes table, using the size recorded in sizeCtl.
     */
    private final Node[] initTable() {
        Node[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node[] nt = (Node[])new Node[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

初始化方法主要应用了关键属性sizeCtl 如果这个值小于0,表示其他线程正在进行初始化,就放弃这个操作。在这也可以看出ConcurrentHashMap的初始化只能由一个线程完成。如果获得了初始化权限,就用CAS方法将sizeCtl置为-1,防止其他线程进入。初始化数组后,将sizeCtl的值改为0.75*n。

扩容

当ConcurrentHashMap容量不足的时候,需要对table进行扩容。这个方法的基本思想跟HashMap是很像的,但是由于它是支持并发扩容的,所以要复杂的多。

原因是它支持多线程进行扩容操作,而并没有加锁。我想这样做的目的不仅仅是为了满足concurrent的要求,而是希望利用并发处理去减少扩容带来的时间影响。因为在扩容的时候,总是会涉及到从一个“数组”到另一个“数组”拷贝的操作,如果这个操作能够并发进行,那真真是极好的了

整个扩容操作分两部分:

  1. 构建一个nextTable,它的容量是原来的两倍,这个线程是单线程操作的。单线程的保证是通过RESIZE_STAMP_SHIFT变量经过一次运算来保证的。
  2. 将原来的table数组复制到nextTable中,这里允许多线程操作
/** The next table to use; non-null only while resizing.
     * 一个过渡的table表,只有在扩容时才使用
     */
    private transient volatile Node<K,V>[] nextTable;

我们首先看下单个线程是如何完成的,它的大致思路就是遍历、复制的过程。首先通过运算得到需要遍历的次数i,然后利用tabAt方法获取位置i处的元素。

(1)如果这个位置为空,就在原table的i位置放入ForwardingNode节点,这个也是触发并发扩容的关键点。

(2)如果这个位置是Node节点(fh>=0),如果它是一个链表的头节点,就构造一个反序链表,把它们分别放在nextTable的i和i+n位置。

(3)如果这个节点是TreeBin节点(fh<0),也做一个反序处理,并且判断是否需要untreefi,把处理结果分别放在nextTable的i和i+n位置。

(4)遍历完所有节点后就完成了复制工作。这时让nextTable作为新的table,并且更新sizeCtl为新容量的0.75倍,完成扩容。

 

再看下多线程的情况

如果线程遍历到的节点是forward节点,就向后继续遍历,再加上给节点上锁的机制,就完成了多线程的控制。多线程遍历节点,处理了一个节点,就把对应点的值set为forward,另一个线程看到forward,就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。这个方法设计很值得学习,数据复制的结构图如下:

 

 
/* Moves and/or copies the nodes in each bin to new table. See
     * above for explanation.
     */
    private final void transfer(Node[] tab, Node[] nextTab) {
        int n = tab.length, stride;
        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[] nt = (Node[])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;
        ForwardingNode fwd = new ForwardingNode(nextTab);//构造一个连节点指针,用于标识位
        boolean advance = true;//并发扩容关键变量,如果为true说明这个节点已经处理过
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) {
            Node f; int fh;
//while循环,依次遍历hash表中的节点
            while (advance) {
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                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赋值给table,并清空临时对象nextTable
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);//扩容阈值设置为原来的1.5倍,依然相当于现在容量的0.75
                    return;
                }
//CAS算法更新扩容阈值,在这里sizeCtl减1,说明新加入一个线程参与到扩容操作
                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
                }
            }
//如果遍历到的节点为空,则放入ForwardingNode指针
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
//如果遍历大ForwardingNode节点,说明这个节点节点已经被处理过,直接跳过,这是控制并发扩容的核心
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {//节点上锁
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node ln, hn;
                        if (fh >= 0) {//fh>=0 证明是Node节点
                            int runBit = fh & n;
//以下部分完成构建两个链表,一个是原链表,另一个是原链表的反序链表
                            Node lastRun = f;
                            for (Node 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 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(ph, pk, pv, ln);
                                else
                                    hn = new Node(ph, pk, pv, hn);
                            }
                            setTabAt(nextTab, i, ln);//在nextTable的i位置插入链表
                            setTabAt(nextTab, i + n, hn);//在nextTable的i+n位置插入链表
                            setTabAt(tab, i, fwd);//在table的i位置插入ForwardingNode节点,表示已经处理过该节点
                            advance = true;//设置true,遍历下一个节点
                        }
                        else if (f instanceof TreeBin) {//对TreeBin对象进行处理,和上面相似
                            TreeBin t = (TreeBin)f;
                            TreeNode lo = null, loTail = null;
                            TreeNode hi = null, hiTail = null;
                            int lc = 0, hc = 0;
//构建正反两个链表
                            for (Node e = t.first; e != null; e = e.next) {
                                int h = e.hash;
                                TreeNode p = new TreeNode
                                    (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;
                                }
                            }
//如果扩容后不再需要Tree结构,反向转换为链表
                            ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                                (hc != 0) ? new TreeBin(lo) : t;
                            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                (lc != 0) ? new TreeBin(hi) : t;
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

put方法

前面的所有的介绍其实都为这个方法做铺垫。ConcurrentHashMap最常用的就是put和get两个方法。

现在来介绍put方法,这个put方法依然沿用HashMap的put方法的思想,根据hash值计算这个新插入的点在table中的位置i,如果i位置是空的,直接放进去,否则进行判断,如果i位置是树节点,按照树的方式插入新的节点,否则把i插入到链表的末尾。

ConcurrentHashMap中依然沿用这个思想,有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程,put方法就要复杂一点。 在多线程中可能出现下面情况:

  1. 如果一个或多个线程正在对ConcurrentHashMap进行扩容,当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到,是因为transfer方法中在空结点上插入forward节点,如果检测到需要插入的位置被forward节点占有,就帮助进行扩容;
  2. 如果检查到要插入的节点是非空且不说ForwardingNode节点,就对这个节点进行加锁,保证线程安全。尽管这有些影响性能,但还是比HashTable的synchronized要好很多。

整个流程首先定义不允许key或value为null的情况放入,对于每个放入的值,首先利用spread方法对key的hashCode进行一次hash计算,由此决定这个值在table中的位置。

如果这个位置的值为空,不需要加锁操作,利用 CAS 尝试写入,失败则自旋保证成功。

如果这个位置存在节点,说明发生了hash碰撞,首先判断这个节点的类型,如果是链表节点(fh>=0),则得到的节点就是hash值相同的节点组成的链表的头节点。如果遇到hash值与key值都与新加入节点是一致的情况,则只需要更新value值即可。否则依次向后遍历,直到链表尾插入这个结点。如果加入这个节点以后链表长度大于8,就把这个链表转换成红黑树。如果这个节点的类型已经是树节点的话,直接调用树节点的插入方法进行插入新的值。

  
/** Implementation for put and putIfAbsent */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
//不允许key或value为null
        if (key == null || value == null) throw new NullPointerException();
//计算hash值
        int hash = spread(key.hashCode());
        int binCount = 0;
//死循环 何时插入成功,何时跳出
        for (Node[] tab = table;;) {
            Node f; int n, i, fh;
//如果table为空,初始化table
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
//根据hash值,计算出table里面的位置
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node(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) {//节点上锁,这个节点可以理解为hash值相同节点组成的链表头节点
                    if (tabAt(tab, i) == f) {
                        if (fh >= 0) {//说明是链表节点
                            binCount = 1;
                            for (Node e = f;; ++binCount) {//遍历链表所有节点
                                K ek;
//如果hash值,key值都相同,则修改对应的value
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node pred = e;
//如果遍历到最后的节点,那么就证明新的节点需要插入,,就把它插入到链表尾部
                                if ((e = e.next) == null) {
                                    pred.next = new Node(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        else if (f instanceof TreeBin) {//如果节点是树节点,就按照树 的方式进行插入
                            Node p;
                            binCount = 2;
                            if ((p = ((TreeBin)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);//当前ConcurrentHashMap的元素数量加1
        return null;
    }

get方法

get方法比较简单,给定一个key来确定value的时候,必须满足两个条件  key相同  hash值相同,对于节点可能在链表或树上的情况,需要分别去查找。

  
public V get(Object key) {
        Node[] tab; Node e, p; int n, eh; K ek;
        int h = spread(key.hashCode());//计算hash值
//根据hash值确定节点位置
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            if ((eh = e.hash) == h) {//如果搜索的节点key与传入的key相同且不为null,直接返回这个节点
                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;
    }
# java 

评论

Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×