ConcurrentHashMap原理分析(1.7和1.8)

时间:2021-05-03 19:38:21

  相信大家都已经了解了HashMap是非线程安全的,在JDK 1.7之前,HashMap在涉及到多线程并发的情况,进行put操作时,可能会成环,引起死循环,进而导致CPU的利用率接近100%。虽然可以使用HashTable或者Collections.synchronizedMap(hashMap)使用线程安全的结合类,但它们二者在处理时都使用了锁,当一个元素在进行读写操作时,其余线程必须阻塞等待,那么效率就可想而知了。

1 为什么要使用ConcurrentHashMap

  通过上面的分析,我们可以清楚的了解到使用HashTable或Conllections.synchronizedMap(hashMap)的效率是极低的,那么我们就该庆幸Doug Lea给我们带来了并发安全的ConcurrentHashMap,它的实现依赖于Java的内存模型。

  既然ConcurrentHashMap不同于前两种线程安全的方法,那么它一定是一种高效且安全的HashMap喽。那么为什么它会高效且安全呢?听小编细细道来。

  ConcurrentHashMap的锁分段技术可以有效提升并发访问率。HashTable容器在竞争激烈的并发环境下效率低下是因为所有访问HashTable的线程都必须竞争同一把锁,一旦竞争不上,就只有阻塞等待。但是如果容器中有多把锁,每一把锁用于锁容器其中一部分数据,那么在高并发的环境下,当多线程访问不同段的数据,那么线程间可以竞争多把锁,从而有效地提高并发访问效率,这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地储存,然后给每一段数据分配一把锁,当一个线程占用锁访问其中一个段数据的时候,其它段的数据也能被其他线程访问。

2 ConcurrentHashMap的结构

  首先通过ConcurrentHashMap的类图来简单了解一下ConcurrentHashMap的结构

ConcurrentHashMap原理分析(1.7和1.8)

  可以从图中看出ConcurrentHashMap继承了AbstractMap类,实现了ConcurrentMap、Serializable两个接口。
  1.AbstractMap是Map接口的抽象实现类,实现了Map的一些基本功能,常见的Map都是继承这个类并重写或者实现一些方法以达成自己的特定目的。
  2.ConcurrentMap这个接口继承了Map接口,主要提供了一些线程安全的方法。
  3.Serializable这个接口提供了序列化与反序列化功能(该类可以被串行化存储起来)。

3 JDK 1.7中ConcurrentHashMap的实现

  在JDK 1.7中,ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成的。如下图所示

ConcurrentHashMap原理分析(1.7和1.8)

  Segment是一种可重入锁(ReentrantLock),在ConcurrentHashMap里扮演锁的角色,将一个大的table分割成多个小的table进行加锁。每一个Segment元素存储的是HashEntry数组+链表。

 3.1 ConcurrentHashMap的初始化

  ConcurrentHashMap初始化方法是通过initialCapacity、loadFacotr和concurrencyLevel等几个参数来初始化segment数组、段偏移量segmentShift、段掩码segmentMask和每个segment里的HashEntry数组来实现的。

 3.1.1 初始化segments数组

  下面为初始化segments数组的源码

if(concurrencyLevel > MAX_SEGMENTS) 
	concurrencyLevel = MAX_SEGMENTS;

int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
    ++sshift;
    ssize <<= 1;
}
segmentShift = 32 - sshift;
segmentMask = ssize -1;
this.segments = Segment.newArray(ssize);

  由上面代码可知,ssize用位运算来计算(ssize <<= 1),所以segments数组的大小取值为2的N次方,即为大于或等于concurrencyLevel的最屌的N次方值来作为segment数组的长度。当然concurrencyLevel最大只能用16位的二进制来表示,即65535,这意味着segments数组的长度最大为65536,对应的二进制为16位。

 3.1.2 初始化segmentShift和segmentMask

  这两个全局变量需要在定位segment的时的散列算法里使用,由初始化segments数组的代码中可知,sshift等于ssize从1向左移位的次数,在默认情况下concurrencyLevel等于16,则1需要向左移位移动4次,所以sshift等于4。

  segmentShift用于定位参与散列预算的位数,segmentShift = 32 - sshift,所以默认为28.

  segmentMask是散列运算的掩码,segmentMask = ssize -1,即默认为15,掩码的二进制各个位的值都是1。

  因为ssize的最大长度为65536,所以segmentShift最大值为16,segmentMask最大值为65535,对应的二进制为16位,每个位都是1。

 3.1.3 初始化每个segment

  输入参数initialCapacity是ConcurrentHashMap的初始化容量,loadfactor是每个segment的负载因子,在构造方法中需要通过这两个参数来初始化数组中的每个segment。

if (initialCapacity > MAXIMUM_CAPACITY)
    initialCapacity = MAXIMUM_CAPACITY;

    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;

    int cap = 1;
    while (cap < c)
        cap <<= 1;

    for (int i = 0; i < this.segments.length; ++i)
        this.segments[i] = new Segment<K,V>(cap, loadFacotr);

  上述代码中的cap是segment里HashEntry数组的长度,它等于initialCapacity除以ssize的倍数,如果c大于1,就会取大于等于c的2的N次方值,所以cap不是1就是2的N次方。segment的容量threshold = (int)cap*loadFactor,默认情况下initialCapacity等于16,loadFactor等于0.75,通过计算cap=1,threshold=0。

 3.2 定位Segment

  通过上面的讲述,我们了解到ConcurrentHashMap使用分段所Segment来保护不同段的数据,那么在插入和获取元素的时候,必须先通过散列算法定位到某一个Segment。ConcurrentHashMap首先选用Wang/Jenkins hash的变种算法对元素的hashCode进行一次再散列。

private static int hash(int h) {
	h += (h << 15) ^ 0xffffcd7d; 
	h ^= (h >>> 10);
	h += (h << 3); 
	h ^= (h >>> 6);
	h += (h << 2) + (h << 14); 
	
	return h ^ (h >>> 16);
}

  使用再散列算法,目的为了减少散列冲突,使元素能够均有地分步在不同的Segment上,从而提高容器的存取效率。假如散列的质量差到极点,那么所有的元素都在一个Segment中,不仅存取元素缓慢,分段所也会失去意义。在JDK 1.7中ConcurrentHashMap通过以下散列算法定位segment。

final Segment<K,V> segmentFor(int hash) {
    return segments[(hash >>> segmentShift) & segmentMask];
}

3.3 JDK 1.7中ConcurrentHashMap的操作

 3.3.1 put操作

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

  从上述Segment的继承体系中可以看出,Segment实现了ReentrantLock,也就带有锁的功能。由于put方法需要对共享变量进行写入操作,所以为了线程安全,在操作共享变量是,必须加锁。

  当执行put操作时,首先根据key的hash值定位到Segment,若该Segment还没有初始化,则通过CAS操作进行赋值,然后进行的二次hash操作,找到相应的HashEntry的位置。在加锁时,会通过继承ReentrantLock的tryLock()方法尝试获取锁,若获取成功,就直接在相应的位置插入;若已经有线程获取了该Segment的所,那当前线程会以自旋的方式继续调用tryLock()方法获取所,超过指定次数就挂起,等待唤醒。

  在插入操作需要经过两个步骤,第一步判断是否需要对Segment里的HashEntry数组进行扩容,第二步定位添加元素的位置,然后将其放在HashEntry数组中。

  1. 是否需要扩容?在插入元素前会先判断Segment里的HashEntry数组是否超过容量threshold,如果超过了阈值,则对数组进行扩容。这里的扩容方式与HashMap的扩容方式稍有不同,HashMap是在插入元素之后判断元素是否已经达到容量,如果达到了就进行扩容,但是有可能扩容之后就没有新元素插入,则HashMap就进行了一次无效的扩容。
  2. 如何扩容?扩容时,首先建立一个容量是原来两倍的数组,然后将原数组中的元素再散列后插入到新数组中。为了高效,ConcurrentHashMap不会对整个容器进行扩容,而是只对某个segment进行扩容。

 3.3.2 get操作

  Segment的get操作实现非常简单和高效。先经过一次再散列,然后使用这个散列值通过散列运算定位到Segment,在通过散列算法定位到元素,代码如下。

public V get(Object key) {
	int hash = hash(key.hashCode());
	return segmentFor(hash).get(key, hash);
}

  get操作的高效体现在整个get过程不需要加锁,除非读到的值是空才会加锁重读

  我们知道HashTable容器的get方法是需要加锁的,所有其效率较低,那么ConcurrentHashMap的get操作时如何做到不加锁的呢?

  因为它的get方法里将使用的共享变量都定义成volatile类型。例如用于统计当前Segment大小的count字段和用于存储值的HashEntry的value。定义成volatile的变量,能够在线程之间保持可见性,能够被多线程同时读,并且保证不会读到过期的值(由Java内存模型的happen before原则保证),但是只能被单线程写(有一种情况可以被多线程写,就是写入的值不依赖于原值)。在get操作期间,只需要读取共享变量count和value值,所以不需要加锁。

之所以不会读到过期的值,是根据Java内存模型的happen before原则,对volatile字段的写入操作先于读操作,即使两个线程同时修改和获取volatile变量,get操作也能拿最新的值。(ps:volatile替换锁的经典应用场景)

 3.3.3 size操作

  计算ConcurrentHashMap的元素大小,就必须要统计Segment里的元素的大小后求和。上面说过Segment的全局变量count是一个volatile变量,在并发的场景下,可能会导致计算出来的size值和实际的size值有偏差。因为在计算count值的时候,可能有新数据的插入,导致结果的不准确。那么,最安全的做法就是在统计size的时候把所有Segment的put、remove和clean方法全部锁住,但这种做法显然是非常低效的。 

  JDK 1.7中是如下处理的,先尝试2次通过不锁住Segment的方式来统计各个Segment大小,如果统计的过程中,容器的count发生了变化,则在采用加锁的方式来统计所有Segment的大小。使用modCount变量判断容器是否发生了变化,在put、remove和clean方法里操作元素前都会将变量modCount进行加1操作,那么在统计size前后比较modCount是否发生变化,从而得知容器的大小是否发生变化。

try {
    for (;;) {
        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;        // 进行2次统计
        last = sum; } }
finally {
    if (retries > RETRIES_BEFORE_LOCK) {
        for (int j = 0; j < segments.length; ++j)
            segmentAt(segments, j).unlock();
    }
}

4 JDK 1.8中ConcurrentHashMap的操作

  JDK 1.8中ConcurrentHashMap的实现已经摒弃了Segment的概念,而是直接使用Node数组+链表+红黑树(与HashMap的底层实现相同)的数据结构实现,并发控制使用了synchronized和CAS操作。整体就像是优化过且线程安全的HashMap,虽然在JDK 1.8中还能看到Segment的数据结构,但已经简化了其属性,知识为了兼容旧版本。

ConcurrentHashMap原理分析(1.7和1.8)

  下面就开始我们的源码之旅喽。

4.1 类的成员变量

/**
 * table数组最大的可能值,值必须为1 << 30。需要这个值用于散列控制
 */
private static final int MAXIMUM_CAPACITY = 1 << 30;

/**
 * 默认的table初始值,必须为2的幂数
 */
private static final int DEFAULT_CAPACITY = 16;

/**
 * 数组的最大可能值,需要与toArray()相关方法关联
 */
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

/**
 * 默认的并发级别,为了兼容以前版本遗留下来的。
 */
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;

/**
 * table的负载因子,在构造方法中重写此值只影响表的初始容量。
 * 通常不使用实际的浮点值,使用(n- n>>>2)表达式跟简单
 */
private static final float LOAD_FACTOR = 0.75f;

/**
 * 链表转化为红黑树的阈值,若节点数>8,则转化为红黑树
 */
static final int TREEIFY_THRESHOLD = 8;

/**
 * 红黑树转化为链表的阈值,当节点数小于等于6转化为链表
 */
static final int UNTREEIFY_THRESHOLD = 6;

/**
 * 转化为树的最小容量(否则,如果一个容器中的节点过多,表将被重新调整大小)
 * 它的值至少为4*TREEIFY_THRESHOLD,以避免大小调整和treeification阈值之间的冲突
 */
static final int MIN_TREEIFY_CAPACITY = 64;

/**
 * 每个转化步骤的最小步数。该值至少为DEFAULT_CAPACITY
 */
private static final int MIN_TRANSFER_STRIDE = 16;

/**
 * 用于在sizeCtl中生成分段的位数,32bit的数组至少为6.
 */
private static int RESIZE_STAMP_BITS = 16;

/**
 * help resize的最大线程数,必须满足(1 << (32 - RESIZE_STAMP_BITS)) - 1
 */
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

/**
 * sizeCtl中记录size的偏移量,默认为32-16=16
 */
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

/*
 * forwarding nodes的hash值
 */
static final int MOVED     = -1; 
/*
 * 红黑树的根节点hash值
 */
static final int TREEBIN   = -2; 
/*
 * reservation node的hash值
 */
static final int RESERVED  = -3;
/*
 * 普通节点的散列位
 */
static final int HASH_BITS = 0x7fffffff; 

/**
 * 可使用的CPU的数量
 */
static final int NCPU = Runtime.getRuntime().availableProcessors();


/**
 * 存放节点的table值
 */
transient volatile Node<K,V>[] table;

/**
 * 在没有竞争使用,通过CAS操作更新
 */
private transient volatile long baseCount;

/**
 * 控制标识符,用来控制table的初始化和扩容的操作,不同的值有不同的含义
 * 当为负数时:-1代表正在初始化,-N代表有N-1个线程正在 进行扩容
 * 当为0时,代表当时的table还没有被初始化
 * 当为正数时:表示初始化或下一次进行扩容的大小
 */
private transient volatile int sizeCtl;

4.2 ConcurrentHashMap的基本数据结构

  • Node是ConcurrentHashMap存储结构的基本单元,为链表结构,仅允许对数据的查找,而不允许修改。 
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    // val和next在扩容时都会发生变化,所以加上volatile来保持线程间可见性,禁止重排序
    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;
    }

    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();
    }

    public final boolean equals(Object o) {
        Object k, v, u; Map.Entry<?,?> e;
        return ((o instanceof Map.Entry) &&
                (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                (v = e.getValue()) != null &&
                (k == key || k.equals(key)) &&
                (v == (u = val) || v.equals(u)));
    }

    /**
     * 用于支持map中的get()方法,在子类中重写
     */
    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;
    }
}
  • TreeNode为红黑树的数据存储结构,用于红黑树中存储数据。TreeBin代码篇幅过长,暂时不在这里赘述。
/**
 * 红黑树的存储结构,与TreeBin共同提供红黑树功能
 * @param <K>
 * @param <V>
 */
static final class TreeNode<K,V> extends Node<K,V> {
    TreeNode<K,V> parent;  // red-black tree links
    TreeNode<K,V> left;
    TreeNode<K,V> right;
    TreeNode<K,V> 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);
    }

    /**
     * 根据key值查找,从根节点开始查找,找到相应的treeNode
     */
    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;
                // hash值大于向左子树
                if ((ph = p.hash) > h)
                    p = pl;
                // hash值小于向右子树
                else if (ph < h)
                    p = pr;
                else if ((pk = p.key) == k || (pk != null && 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;
    }
}

4.3 ConcurrentHashMap的构造方法

public ConcurrentHashMap() {
}

/**
 * 根据initialCapacity创建一个新的空的table。
 *
 * @param initialCapacity 
 * @throws IllegalArgumentException 
 */
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    this.sizeCtl = cap;
}

/**
 * 创建一个与指定Map相同的映射
 *
 * @param m the map
 */
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}

/**
 * 根据initialCapacity和loadFactor创建一个新的空映射,调用了三参构造方法
 *
 * @param initialCapacity 
 * @param loadFactor 
 * @throws IllegalArgumentException 
 * @since 1.6
 */
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}

/**
 * 根据给定的initialCapacity和loadFactor和并发线程数量concurrencyLevel
 * 创建一个新的空表
 *
 * @param initialCapacity
 * @param loadFactor 
 * @param concurrencyLevel 
 * @throws IllegalArgumentException 
 */
public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (initialCapacity < concurrencyLevel)   // Use at least as many bins
        initialCapacity = concurrencyLevel;   // as estimated threads
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
        MAXIMUM_CAPACITY : tableSizeFor((int)size);
    this.sizeCtl = cap;
}

4.4 重要方法分析

  ConcurrentHashMap中有大量的方法(源码都比HashMap长了好多),这里主要挑选较为重要的方法进行介绍。

  • 初始化方法initTable()
/**
 * 初始化table,保证初始化时线程安全
 */
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
    	// 当sizeCtl<0时,说明有其他线程正在初始化table,则线程让出CPU,调用yield()方法
        if ((sc = sizeCtl) < 0)
            Thread.yield(); 
        // 尝试使用CAS操作将sizeCtl修改为-1,开始初始化table
        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<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);    // 记录下扩容的大小
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

  • 扩容方法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;
    // 备用table是否没有被初始化
    if (nextTab == null) {            
        try {
            @SuppressWarnings("unchecked")
            // 扩容为原来的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 = n;
    }
    int nextn = nextTab.length;
    // 连接点指针,用于标志位(fwd的hash值为-1,fwd.nextTable=nextTab)
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // advance作为Hash桶操作完成的标志变量
    boolean advance = true;
    // finishing作为扩充完成的标志变量
    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;
                // sizeCtl阈值扩充为原来的1.5倍
                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
            }
        }
        // 若遍历的节点为null,则放入到ForwadingNode标志节点中,表示该桶不在被处理
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // 如果f.hash == MOVED,表示遍历到了ForwadingNode节点,意味着该节点已经处理过了
        // 并发扩容的核心
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
        	// 使用synchronized给头节点加锁,保证线程安全
            synchronized (f) {
            	// 开始节点复制工作
                if (tabAt(tab, i) == f) {
                	// 创建两个节点头,用于拆分原Hash桶中的数据到两个新的Hash桶中
                    Node<K,V> ln, hn;
                    // 判断头结点的hash值是否大于0,若fh>=0,则可能是链表节点
                    if (fh >= 0) {
                    	// 通过fh & n有效地将原Hash桶中的节点分为值为0和值为1的两类
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        // 遍历找到原链表中的最后一段fh & n和runBit相同的链表节点,将其整段插入到新的链表中
                        // lastRun为最后一段fh & n相同的链表节点的头结点
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        // 根据runBit的值判断这段链表该插入到哪个新链表中。
                        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);
                        }
                        // 将新链表分别插入新表中,将标志节点fwd插入原表中,链表数据拷贝完毕
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                    // 如果待处理的Hash桶中的数据位树节点
                    else if (f instanceof TreeBin) {
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        // 创建lo与hi作为新树的两个根节点
                        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);
                            // 通过h & n == 0来将节点分为两类;同时维护树状结构和链表结构
                            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;
                            }
                        }
                        // 如果差分后的新树节点数小于阈值,则转换为链表
                        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;
                    }
                }
            }
        }
    }
}
  • put操作,对当前table进行无条件自循环put成功,可以分成以下步骤
  1. 如果没有初始化就调用initTable()方法来进行初始化过程
  2. 如果没有hash冲突就直接CAS插入
  3. 如果还在进行扩容操作,就先进行扩容
  4. 如果存在hash冲突,就加锁来保证线程安全,这里存在两种情况:一种是链表形式就直接遍历到尾部插入;另一种是红黑树形式,就按红黑树结构插入。
  5. 最后一个,如果链表的数量大于阈值,则先转换为红黑树结构,再一次进入循环
  6. 如果添加成功,则调用addCount()统计size,并且检查是否需要扩容。
static final int spread(int h) {
    return (h ^ (h >>> 16)) & HASH_BITS;
}

public V put(K key, V value) {
    return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
	// ConcurrentHashMap不允许K、V值为null的键值插入Map中
    if (key == null || value == null) throw new NullPointerException();
    // 通过spread()函数,将key的hashCode值进行位运算,使得高位与低位都参与运算
    // 降低hash碰撞的概率
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {         // 自旋,直至成功为止
        Node<K,V> f; int n, i, fh;
        // 如果table未被初始化,则首先初始化table
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 若通过散列得到的位置中没有节点,则不加锁直接将节点通过CAS操作插入
        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
        }
        // 如果发现该桶中有一个节点(forwarding nodes),则帮助扩容
        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;
                        }
                    }
                }
            }
            // 如果链表的长度大于8,则就转换为红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 统计size,并坚持是否需要扩容
    addCount(1L, binCount);
    return null;
}

/**
 * 将指定Map的值插入到当下ConcurrentHashMap中
 * @param m
 */
public void putAll(Map<? extends K, ? extends V> m) {
    tryPresize(m.size());
    for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
        putVal(e.getKey(), e.getValue(), false);
}
  • get操作

ConcurrentHashMap的get操作流程较为简单,可以大体分为三个步骤来描述

  1. 计算hash值,定位到该table索引位置,如果是首节点符合,则返回
  2. 如果遇到扩容的时候,会调用标志正在扩容节点Forwarding Node的find方法,查找该节点,若匹配则返回
  3. 以上都不符合的话,就向下遍历节点,遇到匹配节点就返回,否则最后返回null
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 通过spread函数得到key值的再散列值
    int h = spread(key.hashCode());
    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为负值,表示此时正在扩容。这时查的是Forwarding Node的find方法来定位到nextTable来查找
        // 若查找到就返回
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        // 若既不是首节点也不是forwarding node,则向下遍历
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}
  • size操作
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;
}


总结与思考

  通过上述分析,我们可以发现JDK 1.8版本的ConcurrentHashMap的数据结构已经接近HashMap。相对而言,ConcurrentHashMap只是增加了同步操作来控制并发。从JDK 1.7版本的ReentrantLock+Segment+HashEntry,到JDK 1.8版本中的synchronized+CAS+HashEntry+红黑树,总结以下几点:

  1. JDK 1.8的实现降低锁的颗粒度,JDK 1.7版本的锁的颗粒度是基于Segment,包含多个HashEntry;而JDK 1.8的锁的颗粒度就是HashEntry。
  2. JDK 1.8版本的数据结构变得更加简单,使得操作也更加清晰。使用了synchronized来进行同步,不需要分段锁的概念,也就不再需要Segment这种数据结构,由于颗粒度的降低,实现的复杂度也增加了。
  3. JDK 1,8使用红黑树来优化链表,红黑树的遍历速度是很快的,代替了一定阈值的链表。


  本文参考了方腾飞老师的《聊聊并发(四)深入分析ConcurrentHashMap》和博文ConcurrentHashMap原理分析。本文特别的针对JDK 1.8下的ConcurrentHashMap进行了简单分析,它的实现与JDK 1.7下的实现手法不大相同。希望童靴们可以多了解一些,在面试中也可以多给面试官说一些东西。噢啦。