源码分析——ConcurrentHashMap

前言

上一篇文章我讲了一下HashMap的相关源码实现,而且咱们知道它是线程不安全的,在并发环境中使用时,HashMap在扩容的时候有可能会生成一个环形链表,从而致使get造成死循环超时。那这篇咱们就来介绍一下并发环境下使用的HashMap——ConcurrentHashMap,下面是它的类关系图。java


JDK1.7中的实现

JDK1.7 中的ConcurrentHashMap采用了分段锁的设计,先来看一下它的数据结构。ConcurrentHashMap中含有一个Segment数组。每一个Segment中又含有一个HashEntry数组。

Segment是一种可重入锁,在ConcurrentHashMap里扮演锁的角色;HashEntry则用于存储键值对数据。node

一个ConcurrentHashMap里包含一个Segment数组。Segment的结构和HashMap相似,是一种数组和链表结构。一个Segment里包含一个HashEntry数组,每个HashEntry是一个链表结构的元素,每一个Segment守护着一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时,必须首先得到与它对应的Segment锁。算法

ConcurrentHashMap经过使用分段锁技术,将数据分红一段一段的存储,而后给每一段数据配一把锁,当一个线程占用锁访问其中一个段数据的时候,其余段的数据也能被其余线程访问,可以实现真正的并发访问。数组


来看看上述Segment结构的定义:

static final class Segment extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;
    static final int MAX_SCAN_RETRIES =
        Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
    transient volatile HashEntry[] table;
    transient int count;
    transient int modCount;
    transient int threshold;
    final float loadFactor;
    ... ...
}复制代码

1. 存储结构

static final class HashEntry<K,V> {
    final int hash;
    final K key;
    volatile V value;
    volatile HashEntry<K,V> next;
}
复制代码
ConcurrentHashMap 和 HashMap 实现上相似,最主要的差异是 ConcurrentHashMap 采用了分段锁(Segment),每一个分段锁维护着几个桶(HashEntry),多个线程能够同时访问不一样分段锁上的桶,从而使其并发度更高(并发度就是 Segment 的个数)。
Segment 继承自 ReentrantLock。

static final class Segment<K,V> extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;

    static final int MAX_SCAN_RETRIES =
        Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

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

    transient int count;

    transient int modCount;

    transient int threshold;

    final float loadFactor;
}
final Segment<K,V>[] segments;复制代码
不看下面的方法,能够看到几个熟悉的字段。HashEntry(哈希数组),threshold(扩容阈值),loadFactor(负载因子)表示segment是一个完整的HashMap。
接下来咱们看看ConcurrentHashMap的构造函数

public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)复制代码
三个参数分别表明了:
  • 初始容量:初始容量表示全部的segment数组中,一共含有多少个hashentry。若initialCapacity不为2的幂,会取一个大于initialCapacity的2的幂。
  • 负载因子:默认0.75。
  • 并发级别:能够同时容许多少个线程并发。concurrencyLevel为多少,就有多少个segment,固然也会取一个大于等于这个值的2的幂。
默认的并发级别为 16,也就是说默认建立 16 个 Segment。

static final int DEFAULT_CONCURRENCY_LEVEL = 16;复制代码


接下来咱们看一下ConcurrentHashMap中的几个关键函数,get,put,rehash(扩容), size方法,看看他是如何实现并发的。
安全

2. get 操做


get实现过程:
  1. 根据key,计算出hashCode;
  2. 根据步骤1计算出的hashCode定位segment,若是segment不为null && segment.table也不为null,跳转到步骤3,不然,返回null,该key所对应的value不存在;
  3. 根据hashCode定位table中对应的hashEntry,遍历hashEntry,若是key存在,返回key对应的value;
  4. 步骤3结束仍未找到key所对应的value,返回null,该key对应的value不存在。
比起Hashtable,ConcurrentHashMap的get操做高效之处在于整个get操做不须要加锁。若是不加锁,ConcurrentHashMap的get操做是如何作到线程安全的呢?缘由是volatile,全部的value都定义成了volatile类型,volatile能够保证线程之间的可见性,这也是用volatile替换锁的经典应用场景。

3. put操做

ConcurrentHashMap提供两个方法put和putIfAbsent来完成put操做,它们之间的区别在于put方法作插入时key存在会更新key所对应的value,而putIfAbsent不会更新。


put实现过程:
  1. 参数校验,value不能为null,为null时抛出空指针异常;
  2. 计算key的hashCode;
  3. 定位segment,若是segment不存在,建立新的segment;
  4. 调用segment的put方法在对应的segment作插入操做。
putIfAbsent实现过程:


putIfAbsent的执行过程与put方法是一致的,除了最后调用的segment的put方法参数onlyIfAbsent传参不同。

segment的put方法实现bash

segment的put方法是整个put操做的核心,它实现了在segment的HashEntry数组中作插入(segment的HashEntry数组采用拉链法来处理冲突)。


segment put实现过程:
1. 获取锁,保证put操做的线程安全;
2. 定位到HashEntry数组中具体的HashEntry;
3. 遍历HashEntry链表,倘若待插入key已存在:
  • 须要更新key所对应value(!onlyIfAbsent),更新oldValue=newValue,跳转到步骤5;
  • 不然,直接跳转到步骤5;
4. 遍历完HashEntry链表,key不存在,插入HashEntry节点,oldValue=null,跳转到步骤5;
5. 释放锁,返回oldValue。

步骤4在作插入的时候实际上经历了两个步骤:
  • 第一:HashEntry数组扩容;
是否须要扩容
在插入元素前会先判断Segment的HashEntry数组是否超过threshold,若是超过阀值,则须要对HashEntry数组扩容;
如何扩容
在扩容的时候,首先建立一个容量是原来容量两倍的数组,将原数组的元素再散列后插入到新的数组里。为了高效,ConcurrentHashMap只对某个Segment进行扩容,不会对整个容器扩容。
  • 第二:定位添加元素对应的位置,而后将其放到HashEntry数组中。

4. size 操做

每一个 Segment 维护了一个 count 变量来统计该 Segment 中的键值对个数。

/** * The number of elements. Accessed only either within locks * or among other volatile reads that maintain visibility. */
transient int count;复制代码
在执行 size 操做时,须要遍历全部 Segment 而后把 count 累计起来。
ConcurrentHashMap 在执行 size 操做时先尝试不加锁,若是连续两次不加锁操做获得的结果一致,那么能够认为这个结果是正确的。
尝试次数使用 RETRIES_BEFORE_LOCK 定义,该值为 2,retries 初始值为 -1,所以尝试次数为 3。
若是尝试的次数超过 3 次,就须要对每一个 Segment 加锁。

static final int RETRIES_BEFORE_LOCK = 2;
public int size() {
    // Try a few times to get accurate count. On failure due to
    // continuous async changes in table, resort to locking.
    final Segment<K,V>[] segments = this.segments;
    int size;
    boolean overflow; // true if size overflows 32 bits
    long sum;         // sum of modCounts
    long last = 0L;   // previous sum
    int retries = -1; // first iteration isn't retry
    try {
        for (;;) {
            // 超过尝试次数,则对每一个 Segment 加锁
            if (retries++ == RETRIES_BEFORE_LOCK) {
                for (int j = 0; j < segments.length; ++j)
                    ensureSegment(j).lock(); // force creation
            }
            sum = 0L;
            size = 0;
            overflow = false;
            for (int j = 0; j < segments.length; ++j) {
                Segment<K,V> seg = segmentAt(segments, j);
                if (seg != null) {
                    sum += seg.modCount;
                    int c = seg.count;
                    if (c < 0 || (size += c) < 0)
                        overflow = true;
                }
            }
            // 连续两次获得的结果一致,则认为这个结果是正确的
            if (sum == last)
                break;
            last = sum;
        }
    } finally {
        if (retries > RETRIES_BEFORE_LOCK) {
            for (int j = 0; j < segments.length; ++j)
                segmentAt(segments, j).unlock();
        }
    }
    return overflow ? Integer.MAX_VALUE : size;
}复制代码

因为在累加count的操做的过程当中以前累加过的count发生变化的概率很是小,因此ConcurrentHashMap先尝试2次不锁住Segment的方式来统计每一个Segment的大小,若是在统计的过程当中Segment的count发生了变化,这时候再加锁统计Segment的count。

ConcurrentHashMap如何判断统计过程当中Segment的cout发生了变化?数据结构

Segment使用变量modCount来表示Segment大小是否发生变化,在put/remove/clean操做里都会将modCount加1,那么在统计size的先后只须要比较modCount是否发生了变化,若是发生变化,Segment的大小确定发生了变化。

JDK 1.8 的改动

JDK 1.7 使用分段锁机制来实现并发更新操做,核心类为 Segment,它继承自重入锁 ReentrantLock,并发度与 Segment 数量相等。
JDK 1.8 使用了 CAS 操做来支持更高的并发度,在 CAS 操做失败时使用内置锁 synchronized。
1.7中Segment[]最大是16,也就是最大支持16个并发。1.8改为Node[],课并行的数量远远大于16。
而且 JDK 1.8 的实现也在链表过长时会转换为红黑树,这点与HashMap1.8的实现是同样的。

数据结构采用数组 + 链表 + 红黑树的方式实现。当链表中(bucket)的节点个数超过8个时,会转换成红黑树的数据结构存储,这样设计的目的是为了提升同一个链表冲突过大状况下的读取效率。并发

Java8中主要作了以下优化:

  1. 将Segment抛弃掉了,直接采用Node(继承自Map.Entry)做为table元素。
  2. 修改时,再也不采用ReentrantLock加锁,直接用内置synchronized加锁,Java8的内置锁比以前版本优化了不少,相较ReentrantLock,性能不并差。
  3. size方法优化,增长CounterCell内部类,用于并行计算每一个bucket的元素数量。

JDK1.8中,出现了较大的改动。没有使用段锁,改为了Node数组 + 链表 + 红黑树的方式。
其中有个重要的变量:sizeCtl
  • 负数表示正在进行初始化或者扩容,-1表示正在初始化,-N表示有N - 1个线程正在扩容
  • 正数0,表示尚未被初始化。其余正数表示下一次扩容的大小。
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;
}复制代码
还有两个数据结构TreeNode、TreeBin,用来当链表的大小超过阈值的时候,将链表变做红黑树。

CAS 操做

这一版本大量使用了CAS操做。所谓的CAS就是,比较内存对应的区域的值,和指望值是否是相等,若是相等,就设置一个新的值进去。
通常是这样使用,先获取对象中的某个域的值,并以这个值为指望值去调用CAS算法。

ConcurrentHashMap中有三个核心的CAS操做async

  • tabAt:得到数组中位置i上的节点
  • casTabAt:设置数组位置i上的节点
  • setTabAt:利用volatile设置位置i上的节点。

//获取索引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节点(将c和table[i]比较,相同则插入v)
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);  
}  
//利用volatile设置节点位置i的值,仅在上锁区被调用 
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {  
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);  
}  复制代码

initTable() 方法

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        //若是一个线程发现sizeCtl<0,意味着另外的线程
        //执行CAS操做成功,当前线程只须要让出cpu时间片
        if ((sc = sizeCtl) < 0)
            Thread.yield();
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            //CAS方法把sizectl置为-1,表示本线程正在进行初始化
            try {
                if ((tab = table) == null || tab.length == 0) {
                    //DEFAULT_CAPACITY 默认初始容量是 16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    //初始化数组,长度为 16 或初始化时提供的长度
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    //将这个数组赋值给 table,table 是 volatile 的
                    table = tab = nt;
                    //若是 n 为 16 的话,那么这里 sc = 12
                    //其实就是 0.75 * n
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}复制代码
在put方法调用时,回去判断table是否是为null,若为null就调用initTable去初始化。

调用initTable会判断sizeCtl的值,若值为-1则表示正在初始化,会调用yield()去等待。函数

若值为0,这时先调用CAS算法去设置为-1,再初始化。

因此执行第一次put操做的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1,有且只有一个线程可以修改为功,其它线程经过Thread.yield()让出CPU时间片等待table初始化完成。

综上所述,能够知道初始化是单线程操做。

put()方法

假设table已经初始化完成,put操做采用CAS+synchronized实现并发插入或更新操做,具体实现以下。

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) {
    //不容许key、value为空
    if (key == null || value == null) throw new NullPointerException();
    //返回 (h ^ (h >>> 16)) & HASH_BITS;
    int hash = spread(key.hashCode());
    int binCount = 0;
    //循环,直到插入成功
    for (Node[] tab = table;;) {
        Node f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)
            //table为空,初始化table
            tab = initTable();
        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)// MOVED=-1;
            //检测到正在扩容,则帮助其扩容
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            //上锁(hash值相同的链表的头节点)
            synchronized (f) {
                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;
                                //仅putIfAbsent()方法中onlyIfAbsent为true
                                if (!onlyIfAbsent)
                                //putIfAbsent()包含key则返回get,不然put并返回
                                    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) {
                //判断是否要将链表转换为红黑树,临界值和HashMap同样也是8
                if (binCount >= TREEIFY_THRESHOLD)
                //若length<64,直接tryPresize,两倍table.length;不转树
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}复制代码

1. hash算法

static final int spread(int h) {
    return (h ^ (h >>> 16)) & HASH_BITS;
}
复制代码
2. table中定位索引位置,n是table的大小

int index = (n - 1) & hash复制代码
3. 获取table中对应索引的元素f。
采用Unsafe.getObjectVolatile来获取。在java内存模型中,咱们已经知道每一个线程都有一个工做内存,里面存储着table的副本,虽然table是volatile修饰的,但不能保证线程每次都拿到table中的最新元素,Unsafe.getObjectVolatile能够直接获取指定内存的数据,保证了每次拿到数据都是最新的。

4. 若是f为null,说明table中这个位置第一次插入元素,利用Unsafe.compareAndSwapObject方法插入Node节点。

  • 若是CAS成功,说明Node节点已经插入,break跳出,随后addCount(1L, binCount)方法会检查当前容量是否须要进行扩容。
  • 若是CAS失败,说明有其它线程提早插入了节点,自旋从新尝试在这个位置插入节点。
5. 若是f的hash值为-1,说明当前f是ForwardingNode节点,意味有其它线程正在扩容,则一块儿进行扩容操做。

6. 其他状况把新的Node节点按链表或红黑树的方式插入到合适的位置,这个过程采用同步内置锁实现并发,代码如上。

在节点f上进行同步,节点插入以前,再次利用tabAt(tab, i) == f判断,防止被其它线程修改。

  1. 若是f.hash >= 0,说明f是链表结构的头结点,遍历链表,若是找到对应的node节点,则修改value,不然在链表尾部加入节点。
  2. 若是f是TreeBin类型节点,说明f是红黑树根节点,则在树结构上遍历元素,更新或增长节点。
  3. 若是链表中节点数binCount >= TREEIFY_THRESHOLD(默认是8),则把链表转化为红黑树结构。

链表转红黑树: treeifyBin()

treeifyBin 不必定就会进行红黑树转换,也多是仅仅作数组扩容。咱们仍是看源码吧。

private final void treeifyBin(Node[] tab, int index) {
    Node b; int n, sc;
    if (tab != null) {
        // MIN_TREEIFY_CAPACITY 为 64
        // 因此,若是数组长度小于 64 的时候,其实也就是 32 或者 16 或者更小的时候,会进行数组扩容
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            // 后面咱们再详细分析这个方法
            tryPresize(n << 1);
        // b 是头结点
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            // 加锁
            synchronized (b) {
                if (tabAt(tab, index) == b) {
                    // 下面就是遍历链表,创建一颗红黑树
                    TreeNode hd = null, tl = null;
                    for (Node e = b; e != null; e = e.next) {
                        TreeNode p =
                            new TreeNode(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(hd));
                }
            }
        }
    }
}复制代码

扩容:tryPresize()

若是说 Java8 ConcurrentHashMap 的源码不简单,那么说的就是扩容操做和迁移操做。

这里的扩容也是作翻倍扩容,扩容后数组容量为原来的 2 倍。

// 首先要说明的是,方法参数 size 传进来的时候就已经翻了倍了
private final void tryPresize(int size) {
    // c:size 的 1.5 倍,再加 1,再往上取最近的 2 的 n 次方。
    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;
        // 这个 if 分支和以前说的初始化数组的代码基本上是同样的
        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); // 0.75 * n
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        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;
                // 2. 用 CAS 将 sizeCtl 加 1,而后执行 transfer 方法
                // 此时 nextTab 不为 null
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)
            // 调用 transfer 方法,此时 nextTab 参数为 null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}复制代码
这个方法的核心在于 sizeCtl 值的操做,首先将其设置为一个负数,而后执行 transfer(tab, null),再下一个循环将 sizeCtl 加 1,并执行 transfer(tab, nt),以后多是继续 sizeCtl 加 1,并执行 transfer(tab, nt)。

至于transfer()方法的源码这里我就不分析了,它的大概功能就是将原来的 tab 数组的元素迁移到新的 nextTab 数组中。

get()方法

get方法不用加锁。利用CAS操做,能够达到无锁的访问。

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {//tabAt(i),获取索引i处Node
        // 判断头结点是否就是咱们须要的节点
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 若是头结点的 hash<0,说明正在扩容,或者该位置是红黑树
        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;
}复制代码

Node<K,V> find(int h, Object k) {
    Node<K,V> e = this;
    if (k != null) {
        do {
            K ek;
            if (e.hash == h &&
                ((ek = e.key) == k || (ek != null && k.equals(ek))))
                return e;
        } while ((e = e.next) != null);
    }
    return null;
}复制代码
get() 执行过程:
1. 计算 hash 值
2. 根据 hash 值找到数组对应位置: (n – 1) & h
3. 根据该位置处结点性质进行相应查找
  • 若是该位置为 null,那么直接返回 null 就能够了
  • 若是该位置处的节点恰好就是咱们须要的,返回该节点的值便可
  • 若是该位置节点的 hash 值小于 0,说明正在扩容,或者是红黑树
  • 若是以上 3 条都不知足,那就是链表,进行遍历比对便可

小结

到这里我就基本把ConcurrentHashMap在 JDK 1.7和1.8中的实现大概捋了一遍,并详细分析了几个重要的方法实现:初始化、put、get。在 JDK1.8中ConcurrentHashMap发生了较大的变化,经过使用CAS+synchronized的实现取代了原先 1.7 中的Segment分段锁机制,从而支持更高的并发量。

这只是我对ConcurrentHashMap的第二次学习,若想更好地理解掌握ConcurrentHashMap的实现之精妙,我的以为还需之后再多看几回,相信每次都会有新的收获。

相关文章
相关标签/搜索