ConcurentHashMap设计与源码分析

简介

ConcurentHashMap是java.util.concurrent包下的一个线程安全的类,继承自Map类,用于存储具有键(key)、值(value)映射关系的双列集合。其数据结构与HashMap类似,都是使用数组+链表+树(红黑树)的结构实现。

优点

数据结构

ConcurentHashMap底层使用数组加链表的形式存储,K-V通过内部类Node包装,当链表长度大于8时,转换为树节点(TreeNode),超过64时改用红黑树(一种自平衡二叉查找树)

ConcurentHashMap数据结构的实现主要通过Node、TreeNode、TreeBin等内部类实现,其UML图如下:
ConcurentHashMap设计与源码分析

Node

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;

    Node(int hash, K key, V val, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.val = val;
        this.next = next;
    }
}

Node类实现了Entry接口,用于存储节点的hash(哈希值)、key(键)、value(值)以及next(下一个节点的地址)四个属性。

TreeNode

TreeNode继承了Node类,用于存储ConcurentHashMap中的树结构,其构造方法如下:

TreeNode(int hash, K key, V val, Node<K,V> next,
         TreeNode<K,V> parent) {
    super(hash, key, val, next);
    this.parent = parent;
}

在构造方法中,TreeNode调用了Node的构造方法,并指定了该节点的父节点。

TreeBin

TreeBin用于包装TreeNode类,当链表过长时,TreeBin会把TreeNode转换为红黑树。事实上,在ConcurentHashMap的“数组”中(也就是树的根节点)所存储的并不是TreeNode而是TreeBin。TreeBin不存储key/value,TreeBin还维护了一个读写锁,使得读必须等待写操作完成才能进行。

ForwardingNode

ForwardingNode用于标记正在迁移中的Node。在其构造方法会生成一个key、value 和 next 都为 null,且 hash 为 MOVED 的 Node。

static final class ForwardingNode<K,V> extends Node<K,V> {
    final Node<K, V>[] nextTable;

    ForwardingNode(Node<K, V>[] tab) {
        super(MOVED, null, null, null);
        this.nextTable = tab;
    }
}

结构示意图

ConcurentHashMap设计与源码分析

线程安全

1、CAS机制

对节点的修改操作都通过CAS来完成,CAS机制实现了无锁化的修改值的操作,可以大大降低锁代理的性能消耗。

2、volatile关键字

在ConcurentHashMap中,部分变量使用了volatile关键字修饰,保证了变量的可见性和指令的有序性。例如对节点操作进行控制的sizeCtl变量,在Node类中的val、next变量。

3、synchronized

ConcurentHashMap需要使用synchronized对数组中的的非空节点进行加锁操作(空节点可通过CAS直接进行操作,不需要加锁),如put方法及transfer方法。

4、Unsafe类和三个tabAt方法

Java是无法对操作系统底层进行操作的,所以CAS等操作的具体实现都需要Unsafe类以对底层进行操作。而对于节点的取值、设值、修改等操作,ConcurentHashMap基于Unsafe类封装了三个tabAt方法。

/**
 * ((long)i << ASHIFT) + ABASE用于计算出元素的真实地址
 * ASHIFT为每个节点(Node)的偏移量(位数)
 * ABASE为头节点的地址(arrayBaseOffset)
 */
// 获得在i位置上的Node节点
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算法设置i位置上的Node节点
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);
}

方法详解

构造函数

ConcurentHashMap有多个重载的构造方法,可传入三个参数

  • (int) initialCapacity 指ConcurrentHashMap的初始容量
  • (float) loadFactor 加载因子
  • (int) concurrencyLevel 并发度

在Java7中,ConcurentHashMap使用Segment分片的形式实现,Segment之间允许线程进行并发操作,而concurrencyLevel则是用来设置Segment[]数组长度的,concurrencyLevel的最小2次幂便为实际并发度。

而在Java8中,ConcurentHashMap摒弃了Segment,改用CAS加上TreeBin等辅助类实现,并发度concurrencyLevel也就没有实际意义了。

public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    // MAXIMUM_CAPACITY = 1 << 30(2^30 = 1073741824)
    // 如果大小为MAXIMUM_CAPACITY最大总量的一半,那么直接将容量设为MAXIMUM_CAPACITY,否则计算最小幂次方
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
            MAXIMUM_CAPACITY :
            // 1.5 * initialCapacity + 1
            tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    this.sizeCtl = cap;
}

构造函数进行了sizeCtl的赋值,sizeCtl作为控制标识符,不同的数值代表不同的意义

  • -1代表正在初始化
  • -N表示有N-1个线程正在进行扩容操作
  • hash表还没有被初始化时,该数值表示初始化或下一次进行扩容的大小。
  • 初始化之后,它的值始终是当前ConcurrentHashMap容量的0.75倍
    ConcurentHashMap设计与源码分析

initTable初始化

初始化一个容量为sizeCtl的Node数组

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        // sizeCtl小于0,表示有其他线程正在进行初始化操作,把线程挂起
        if ((sc = sizeCtl) < 0)
            // 自旋
            Thread.yield(); // lost initialization race; just spin
        // 利用CAS方法把sizectl的值置为-1,表示本线程正在进行初始化,防止其他线程进入
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // 默认大小16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 初始化Node
                    table = tab = nt;
                    // 设置一个扩容的阈值 相当于0.75*n
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

互斥同步进入阻塞状态需要很大的开销,initTable方法使用了自旋锁,通过Thread.yield()使线程让步,然后忙循环直到sizeCtl满足条件

tableSizeFor函数详解

Returns a power of two table size for the given desired capacity.

返回大于输入参数且最近的2的整数次幂的数

/**
 * 使最高位的1后面的位全变为1,最后再让结果n+1,即得到了2的整数次幂的值
 */
private static final int tableSizeFor(int c) {
    int n = c - 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

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();
    // spread将较高的哈希值扩展为较低的哈希值,并将最高位强制为0
    int hash = spread(key.hashCode());
    // binCount用于记录相应链表的长度
    int binCount = 0;
    // 死循环
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // tab为空,初始化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<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // 存在节点,说明发生了hash碰撞,需要对表进行扩容
        // 如果该位置的节点存在值且为MOVED(-1),说明正在扩容
        else if ((fh = f.hash) == MOVED)
            // helpTransfer方法用于增加线程以协助扩容
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            // 节点上锁
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    // fh > 0 说明这个节点是一个链表的节点 不是树的节点
                    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;
                            // 不存在该key,插入新Node
                            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) {
                // TREEIFY_THRESHOLD = 8
                // 链表长度大于8,转换为树节点
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 将当前ConcurrentHashMap的元素数量+1
    addCount(1L, binCount);
    return null;
}

putVal()方法首先获取到key的hash值,该key所对应位置的值为空,直接放入,若该键对应的位置存在节点,则判断是否该节点为链表节点还是树节点,再使用其相应的方法将Node放入

putVal()方法大概的流程图如下:
ConcurentHashMap设计与源码分析

get方法

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 获取hash值
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
        // 哈希值相等,返回该节点的值
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // TreeBin或ForwardingNode,调用find方法
        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;
}

treeifyBin

Replaces all linked nodes in bin at given index unless table is too small, in which case resizes instead.

将所有索引处的链表节点替换为二叉树,如果表太小则改为调整大小

private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {
        // 链表tab的长度小于64(MIN_TREEIFY_CAPACITY=64)时,进行扩容
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            // 调用tryPresize进行扩容(所传参数n即链表长度 * 2)
            tryPresize(n << 1);
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            synchronized (b) {
                if (tabAt(tab, index) == b) {
                    TreeNode<K,V> hd = null, tl = null;
                    // 遍历链表,建立红黑树
                    for (Node<K,V> e = b; e != null; e = e.next) {
                        TreeNode<K,V> p =
                                new TreeNode<K,V>(e.hash, e.key, e.val,
                                        null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    setTabAt(tab, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

tryPresize

Tries to presize table to accommodate the given number of elements.

尝试调整表的大小以适应给定的元素数量。

private final void tryPresize(int size) {
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
            tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    // 正在扩容
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table; int n;
        // tab未初始化,进行初始化,与initTable类似
        if (tab == null || (n = tab.length) == 0) {
            n = (sc > c) ? sc : c;
            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    if (table == tab) {
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        // 如果扩容大小没有达到阈值,或者超过最大容量
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        // 调用transfer()扩容
        else if (tab == table) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                Node<K,V>[] nt;
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                    break;
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                    (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}

transfer

Moves and/or copies the nodes in each bin to new table.

将树中的节点移动和复制到新表中

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    
    // NCPU为可用的CPU线程数
    // stride可以理解为步长,最小值为16
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range
    // nextTab为null,进行初始化
    if (nextTab == null) {            // initiating
        try {
            // 容量*2的节点数组
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        // transferIndex用于控制迁移的位置
        transferIndex = n;
    }
    int nextn = nextTab.length;
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // advance为true说明这个节点已经处理过
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab
    
    // i 是位置索引,bound 是边界,从后往前移
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        
        // 此处的while循环用作遍历hash原表中的节点
        while (advance) {
            int nextIndex, nextBound;
            
            // 处理从bound到i的节点
            if (--i >= bound || finishing)
                advance = false;

            // transferIndex 小于等于 0,说明原hash表的所有位置都有相应的线程去处理了
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            
            // i指向transferIndex,bound指向(transferIndex-stride)
            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;
            // 迁移完成,nextTab赋值给table
            if (finishing) {
                nextTable = null;
                table = nextTab;
                // 重新计算sizeCtl,得出的值是新数组长度的 0.75 倍
                sizeCtl = (n << 1) - (n >>> 1);
                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
            }
        }
        // 位置 i 为空,放入空节点fwd
        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<K,V> ln, hn;
                    // 链表节点
                    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);
                        }
                        // 在i位置放入一个链表
                        setTabAt(nextTab, i, ln);
                        // 在i+n位置放入另一个链表
                        setTabAt(nextTab, i + n, hn);
                        // 在table的i位置上插入forwardNode节点, 表示已经处理过该节点
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                    // 树结构
                    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;
                            }
                        }
                        // 小于UNTREEIFY_THRESHOLD(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;
                    }
                }
            }
        }
    }
}

还有些事

1、为什么要取最小二次幂

因为HashMap通过对hash值的i = (n - 1) & hash运算实现均匀分布,若n不为2的次幂数,就不能保证均匀分布。

参考文章

1、ConcurrentHashMap总结

2、Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析

3、jdk1.8的HashMap和ConcurrentHashMap

ConcurentHashMap设计与源码分析

上一篇:Solution -「CF 600A」Extract Numbers


下一篇:ASP.NET MVC Model元数据(一)