ConcurrentHashMap中有十个提升性能的细节，你都知道吗？

点击上方蓝色“




    
石杉的架构笔记”，选择“设为星标”
回复“PDF”获取独家整理的学习资料！

长按扫描上方 免费领取

一些题外话

如何在高并发下提高系统吞吐是所有后端开发者追求的目标，Java并发的开创者 Doug Lea在 Java 7 ConcurrentHashMap的设计中给出了一些参考答案，本文详细的总结了 ConcurrentHashMap源码中影响并发性能的十个细节，有常见的自旋锁，CAS的使用，也有延迟写内存，volatile语义退化等不常见的技巧，希望对大家的开发设计有所帮助。

把书读薄

1. ConcurrentHashMap 是一个线程安全的 Map 实现，其读取不需要加锁，通过引入 Segment ，可以做到写入的时候加锁力度足够小
2. 由于引入了 Segment ， ConcurrentHashMap 在读取和写入的时候需要需要做两次哈希，但这两次哈希换来的是更细力粒度的锁，也就意味着可以支持更高的并发
3. 每个桶数组中的 key-value 对仍然以链表的形式存放在桶中，这一点和 HashMap 是一致的。

把书读厚

1. ConcurrentHashMap 的哪些操作需要加锁？
2. ConcurrentHashMap 的无锁读是如何实现的？
3. 在多线程的场景下调用 size（） 方法获取 ConcurrentHashMap 的大小有什么挑战？ ConcurrentHashMap 是怎么解决的？
4. 在有 Segment 存在的前提下，应该如何扩容的？

初始化

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (concurrencyLevel > MAX_SEGMENTS)
        concurrencyLevel = MAX_SEGMENTS;
    // Find power-of-two sizes best matching arguments
    int sshift = 0;
    int ssize = 1;
    // 保证ssize是大于concurrencyLevel的最小的2的整数次幂
    while (ssize         ++sshift;
        ssize <<= 1;
    }
    // 寻址需要两次哈希，哈希的高位用于确定segment，低位用户确定桶数组中的元素
    this.segmentShift = 32 - sshift;
    this.segmentMask = ssize - 1;
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize         ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap         cap <<= 1;
    Segment s0 = new Segment(loadFactor, (int)(cap * loadFactor), (HashEntry[])new HashEntry[cap]);
    Segment[] ss = (Segment[])new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

1. 确定了 segments 的数组的大小 ssize ， ssize 根据入参 concurrencyLevel 确定，取大于 concurrencyLevel 的最小的2的整数次幂
2. 确定哈希寻址时的偏移量，这个偏移量在确定元素在 segment 数组中的位置时会用到
3. 初始化 segment 数组中的第一个元素，元素类型为 HashEntry 的数组，这个数组的长度为 initialCapacity / ssize ，即初始化大小除以 segment 数组的大小， segment 数组中的其他元素在后续 put 操作时参考第一个已初始化的实例初始化

static final class HashEntry<K,V> {
    final int hash; 
    final K key;
    volatile V value;
    volatile HashEntry next; 
 
    HashEntry(int hash, K key, V value, HashEntry next) {
        this.hash = hash;
        this.key = key;
        this.value = value;
        this.next = next;
    }
    final void setNext(HashEntry n) {
        UNSAFE.putOrderedObject(this, nextOffset, n);
    }
}

JMM中有一条规定：

对一个变量执行unlock操作之前，必须先把此变量同步到主内存中（执行store和write操作）

哈希

• 确定哈希寻址时的偏移量，这个偏移量在确定元素在 segment 数组中的位置时会用到

this.segmentShift = 32 - sshift;

public V put(K key, V value) {
    Segment s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    // 变量j代表着数据项处于segment数组中的第j项
    int j = (hash >>> segmentShift) & segmentMask;
        // 如果segment[j]为null,则下面的这个方法负责初始化之
        s = ensureSegment(j); 
    return s.put(key, hash, value, false);
}

private Segment ensureSegment(int k) {
    final Segment[] ss = this.segments;
    long u = (k     Segment seg;
    if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
        Segment proto = ss[0]; // use segment 0 as prototype
        int cap = proto.table.length;
        float lf = proto.loadFactor;
        int threshold = (int)(cap * lf);
        HashEntry[] tab = (HashEntry[])new HashEntry[cap];
        if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck 再检查一次是否已经被初始化
            Segment s = new Segment(lf, threshold, tab);
            while ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) // 使用 CAS 确保只被初始化一次
                    break;
            }
        }
    }
    return seg;
}

put 方法

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry node = tryLock() ? null : scanAndLockForPut(key, hash, value); 
    V oldValue;
    try {
        HashEntry[] tab = table;
        int index = (tab.length - 1) & hash;
        HashEntry first = entryAt(tab, index);
        for (HashEntry e = first;;) {
            if (e != null) {
                K k; // 如果找到key相同的数据项，则直接替换
                if ((k = e.key) == key || (e.hash == hash && key.equals(k))) {
                    oldValue = e.value;
                    if




    
 (!onlyIfAbsent) {
                        e.value = value;
                        ++modCount; 
                    }
                    break;
                }
                e = e.next;
            }
            else {
                if (node != null)
                    // node不为空说明已经在自旋等待时初始化了，注意调用的是setNext，不是直接操作next
                    node.setNext(first); 
                else
                    // 否则，在这里新建一个HashEntry
                    node = new HashEntry(hash, key, value, first);
                int c = count + 1; // 先加1
                if (c > threshold && tab.length                     rehash(node);
                else
                    // 将新节点写入，注意这里调用的方法有门道
                    setEntryAt(tab, index, node); 
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

private HashEntry scanAndLockForPut(K key, int hash, V value) {
    HashEntry first = entryForHash(this, hash);
    HashEntry e = first;
    HashEntry node = null;
    int retries = -1; // negative while locating node
    while (!tryLock()) { // 自旋等待
        HashEntry f; // to recheck first below
        if (retries 0) {
            if (e == null) { // 这个桶中还没有写入k-v项
                if (node == null) // speculatively create node 直接创建一个新的节点
                    node = new HashEntry(hash, key, value, null);
                retries = 0;  
            }
            // key值相等，直接跳出去尝试获取锁
            else if (key.equals(e.key))
                retries = 0;
            else // 遍历链表
                e = e.next;
        }
        else if (++retries > MAX_SCAN_RETRIES) {
            // 自旋等待超过一定次数之后只能挂起线程，阻塞等待了
            lock();
            break;
        }
        else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) { 
            // 如果头节点改变了，则重置次数，继续自旋等待
            e = first = f; 
            retries = -1; 
        }
    }
    return node;
}

HashEntry[] tab = table;

transient volatile HashEntry[] table;

嗅探

每个处理器通过嗅探在总线上传播的数据来检查自己缓存的值是不是过期了，当处理器发现自己缓存行对应的内存地址被修改，就会将当前处理器的缓存行设置成无效状态，当处理器对这个数据进行修改操作的时候，会重新从系统内存中把数据读到处理器缓存里

static final  HashEntry entryAt(HashEntry[] tab, int i) {
    return (tab == null) ? null : (HashEntry) UNSAFE.getObjectVolatile(tab, ((long)i <}

static final  void setEntryAt(HashEntry[] tab, int i, HashEntry e) {
    UNSAFE.putOrderedObject(tab, ((long)i <}

对一个变量执行unlock操作之前，必须先把此变量同步到主内存中（执行store和write操作）

node.setNext(first);

final void setNext(HashEntry n) {
    UNSAFE.putOrderedObject(this, nextOffset, n);
}