ConcurrentHashMap实现原理

Java识堂

发布于 2019-08-13 09:55:20

3990

发布于 2019-08-13 09:55:20

文章被收录于专栏：Java识堂

前言

在这篇文章中对属性介绍的比较多：HashMap实现原理

HashMap不是线程安全的，在多线程环境下可以使用Hashtable和ConcurrentHashMap，Hashtable实现线程安全的方式是用synchronized修饰方法，如get和put方法都是用synchronized修饰的，使用的是对象锁，这样会导致线程1get元素（或者put元素）时，线程2不能get元素和put元素，在竞争激烈的时候会出现严重的性能问题

简介

Hashtable出现性能问题的原因是所有访问Hashtable的线程都在竞争一把锁，假如容器中有多把锁，每一把锁用于锁容器的中的一部分数据，那么多线程访问容器里不同数据段的数据时，线程之间就不会存在锁竞争，从而可以有效提高并发访问效率，这就是ConcurrentHashMap使用的锁分段技术

ConcurrentHashMap的主要结构如下

假设我们有三个键值对，dnf:1，cf:2，lol:3，每次放值会进行2次hash，即先确定放在哪个Segment中，再确定放在哪个HashEntry中。

假设三个键值对同时进行放，1=hash1(dnf)，知道了放在应该放在segments[1]处，接着获取到segments[1]的锁，再进行hash，2=hash2(dnf)，即放在hashentrys[2]处，放完对segments[1]解锁

3=hash1(cf)，放在segments[3]处，获取到segments[3]的锁，0=hash2(cf)，放在hashentrys[0]，放完对segments[3]解锁

1=hash1(lol)，放在segments[1]处，因为此时segments[1]的锁已经被put key为dnf的线程获取，所以会阻塞的获取锁，直到锁被put key为dnf的这一线程释放，获取到锁后，2=hash2(lol)，放在hashentrys[2]处，因为已经有值了，采用头插法，放在链表的头节点

3个线程操作完，结果如下

get方法也是进行两次hash即可，get方法不用上锁，get方法只读不写，用volatile修饰即可

源码

基于jdk1.7.0_80

关注点

结论

是否允许为null

key和value均不允许为null

是否有序

无序

是否线程安全

线程安全

先看ConcurrentHashMap类的属性

// segments数组的初始容量
static final int DEFAULT_INITIAL_CAPACITY = 16;

// 负载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;

// 默认并发数
static final int DEFAULT_CONCURRENCY_LEVEL = 16;

// 最大容量是2的30次方 
static final int MAXIMUM_CAPACITY = 1 << 30;

// 最小段数，必须是2的倍数
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;

// segments数组的最大大小，必须是2的倍数
static final int MAX_SEGMENTS = 1 << 16; // slightly conservative

// 在size和containsValue方法中使用
// 在采用加锁方法之前， 最多尝试的次数
static final int RETRIES_BEFORE_LOCK = 2;

// 段掩码（和segmentShift配合使用定位Segment）
final int segmentMask;

// 段偏移量
final int segmentShift;

final Segment<K,V>[] segments;

这里说一下RETRIES_BEFORE_LOCK，由于多线程的缘故，调用size和containsValue方法有可能得不到准确的结果，不加锁尝试RETRIES_BEFORE_LOCK+1次还得不到准确的结果，直接上锁

接着看Segment内部类

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

    // 链表数组，数组中的每一个元素存放了一个链表的头部
    transient volatile HashEntry<K,V>[] table;

    // 元素数量
    transient int count;

    // 段被修改的次数（如执行put或者remove）
    transient int modCount;

    // 进行扩容的阀值
    transient int threshold;

    // 负载因子
    final float loadFactor;

    Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
        this.loadFactor = lf;
        this.threshold = threshold;
        this.table = tab;
    }
}

HashEntry内部类

static final class HashEntry<K,V> {
    final int hash;
    final K key;
    volatile V value;
    volatile HashEntry<K,V> next;

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

}

注意这里value和next用volatile修饰保证了可见性

构造方法

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    // MAX_SEGMENTS = 1 << 16，最大并发数为1 << 16
    if (concurrencyLevel > MAX_SEGMENTS)
        concurrencyLevel = MAX_SEGMENTS;
    // 2的sshift次方为ssize
    int sshift = 0;
    // ssize为segments的数组长度
    int ssize = 1;
    // 当concurrencyLevel为16时，ssize为16
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
    this.segmentShift = 32 - sshift;
    this.segmentMask = ssize - 1;
    // MAXIMUM_CAPACITY = 1 << 30，最大容量为1 << 30
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    // 计算Segment中HashEntry数组长度
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    // MIN_SEGMENT_TABLE_CAPACITY = 2
    // 计算Segment下HashEntry数组的长度
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // 创建Segment数组，并且设置segments[0]
    Segment<K,V> s0 =
        new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
                         (HashEntry<K,V>[])new HashEntry[cap]);
    Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

默认情况下concurrencyLevel=16，这样就会导致segments的数组长度也是16，每个Segment里面的HashEntry数组的大小为2

接着看put操作

// ConcurrentHashMap
public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    // 对元素的hashCode进行一次再散列，减少散列冲突
    int hash = hash(key);
    // 定位Segment
    int j = (hash >>> segmentShift) & segmentMask;
    // 如果Segment不存在，则调用ensureSegment方法
    if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}

根据索引去segments数组中获取Segment，如果已经存在了则返回，否则创建并自旋插入

// ConcurrentHashMap
private Segment<K,V> ensureSegment(int k) {
    final Segment<K,V>[] ss = this.segments;
    long u = (k << SSHIFT) + SBASE; // raw offset
    Segment<K,V> seg;
    // 该索引处还没有Segment
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
        // 这里能直接赋值的原因是ss[0]在构造函数中已经初始化了
        Segment<K,V> proto = ss[0]; // use segment 0 as prototype
        int cap = proto.table.length;
        float lf = proto.loadFactor;
        int threshold = (int)(cap * lf);
        HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
        // 再次检查
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
            == null) { // recheck
            Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
            // 自旋插入，成功则退出
            while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                   == null) {
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                    break;
            }
        }
    }
    return seg;
}

ConcurrentHashMap将put操作代理给Segment

将value插入定位到的Segment的HashEntry数组，如果key已经存在，则返回oldValue，否则返回null 注意看最后一个参数，put方法调用的是s.put(key, hash, value, false)，即key相等的时候，put会用newValue替换oldValue 而putIfAbsent方法调用的是s.put(key, hash, value, true)，即key相等的时候，put不会用newValue替换oldValue

// Segment
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 尝试直接获取锁，获取到锁node为null，否则调用scanAndLockForPut方法
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        // 获取在tab数组中的位置
        int index = (tab.length - 1) & hash;
        // 得到链表的头节点
        HashEntry<K,V> first = entryAt(tab, index);
        // 遍历链表
        for (HashEntry<K,V> e = first;;) {
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        e.value = value;
                        ++modCount;
                    }
                    break;
                }
                e = e.next;
            }
            // 遍历到链表尾部，没有重复的key，则新插入
            else {  
                if (node != null)
                    // 头插法，将node节点设为链表头节点
                    node.setNext(first);
                else
                    // 为null，则新建一个节点
                    node = new HashEntry<K,V>(hash, key, value, first);
                int c = count + 1;
                // 若c超过阈值则扩容，并且数组长度小于MAXIMUM_CAPACITY = 1 << 30
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    // 扩容并进行重新hash
                    rehash(node);
                else
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

获取锁失败才会调用这个方法，说明锁被其他线程所占有

// Segment
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
    // 获取链表头结点
    HashEntry<K,V> first = entryForHash(this, hash);
    HashEntry<K,V> e = first;
    HashEntry<K,V> node = null;
    int retries = -1; // negative while locating node
    // 不断尝试获取锁
    while (!tryLock()) {
        HashEntry<K,V> f; // to recheck first below
        if (retries < 0) {
            // 链表的头结点为null，或者遍历到链表的尾部
            if (e == null) {
                // 这里加条件是因为，有可能已经初始化node节点了
                // 结果由于头结点改变重新遍历链表
                if (node == null) // speculatively create node
                    node = new HashEntry<K,V>(hash, key, value, null);
                retries = 0;
            }
            // 找到相同key的节点
            else if (key.equals(e.key))
                retries = 0;
            // 没有找到key对应的节点，指向下一个节点
            else
                e = e.next;
        }
        // 可用处理器数量大于1，MAX_SCAN_RETRIES=64，否则为1
        else if (++retries > MAX_SCAN_RETRIES) {
            // 调用ReentrantLock中NonfairSync的lock()方法
            // 执行过程中有可能不阻塞获取到锁，也有可能被阻塞
            // 而不是之前的一直尝试直接获取锁
            lock();
            break;
        }
        // 链表的头结点发生变化，更新头结点，并重置retries值为-1
        else if ((retries & 1) == 0 &&
                 (f = entryForHash(this, hash)) != first) {
            e = first = f; // re-traverse if entry changed
            retries = -1;
        }
    }
    return node;
}

get方法并没有上锁，而是利用了volatile的可见性

public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    // 定位Segment
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
        (tab = s.table) != null) {
        for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
             e != null; e = e.next) {
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                return e.value;
        }
    }
    return null;
}

最后看size方法

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 (;;) {
            // RETRIES_BEFORE_LOCK = 2 不上锁求值尝试3次，值不一样，直接上锁
            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;
                }
            }
            // 只要有连续2次值相等，段没有被修改，退出
            if (sum == last)
                break;
            last = sum;
        }
    } finally {
        // 如果执行到加锁，则解锁
        if (retries > RETRIES_BEFORE_LOCK) {
            for (int j = 0; j < segments.length; ++j)
                segmentAt(segments, j).unlock();
        }
    }
    // 如果size超过了Integer.MAX_VALUE则返回Integer.MAX_VALUE
    return overflow ? Integer.MAX_VALUE : size;
}

在计算ConcurrentHashMap的size时，因为并发操作的缘故，还有可能一直插入数据，可能导致计算返回的 size和实际的size有相差（在return size的时候插入了多个数据），因此会分为如下2步来进行

尝试不加锁的模式计算2（RETRIES_BEFORE_LOCK）+1次，其中有连续两次计算的总的modCount相等则直接返回size
尝试完3次后，如果没有连续两次计算的结果相等，则对segments加锁求size

这里为什么会超过Integer.MAX_VALUE呢？因为ConcurrentHashMap最多有（MAX_SEGMENTS = 2^16）个Segment，而每个Segment允许的最大容量为（MAXIMUM_CAPACITY = 2 ^30），则最大值为（2^46），int最大值为（2^ 31 - 1）

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2018-05-13，如有侵权请联系 cloudcommunity@tencent.com 删除

http