生活

為什麼我們總是沒有時間把事情做對，卻有時間做完它？

瞭解ConcurrentHashMap

工作中常用到hashMap,但是HashMap在多執行緒高併發場景下並不是執行緒安全的。
所以引入了ConcurrentHashMap，它是HashMap的執行緒安全版本，採用了分段加鎖的方式來保證執行緒安全，同樣在高併發的場景下有較好的效能。

ConcurrentHashMap 組成

ConcurrentHashMap 底層通過分段鎖的形式實現，他的底層一共16個Segment，而每個Segment維護一個HashEntry，在操作HashEntry裡的資料時只需給對應的Segment加速，所以在高併發場景下的效能比較好。
現在先來簡單瞭解一下它的組成：

成員

//Segment內部HashEntry預設的初始化容量
    static final int DEFAULT_INITIAL_CAPACITY = 16;
    //擴容時用到的載入因子
    static final float DEFAULT_LOAD_FACTOR = 0.75f;
    //預設併發數，就是Segment的數量
    static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    //segment內部 hashEntry的最大長度
    static final int MAXIMUM_CAPACITY = 1 << 30;
    //每個分段最小長度
    static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
    //分段最大容量
    static final int MAX_SEGMENTS = 1 << 16; // slightly conservative
    //預設自旋次數。超過這個次數就加鎖
    static final int RETRIES_BEFORE_LOCK = 2;
    
    final int segmentMask;

   
    final int segmentShift;
	
	// segment陣列
    final Segment<K,V>[] segments;

    transient Set<K> keySet;
    transient Set<Map.Entry<K,V>> entrySet;
    transient Collection<V> values;

 

Segment

//hashentey實際儲存資料的地方
 transient volatile HashEntry<K,V>[] table;
//保持的數量
        transient int count;
     //修改的次數
        transient int modCount;
	//擴容 rehash閾值
        transient int threshold;
        //載入因子
        final float loadFactor;

注意segment其實就是一個鎖，它繼承自ReentrantLock,在執行某些操作時時互斥的。

HashEntry

HashEntry時ConcurrentHashMap下粒度最小的資料結構：

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

原始碼深入

建立

//其他構造器最終都呼叫這個構造器
 @SuppressWarnings("unchecked")
    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是2的次數，方便後面的位運算，只要移位就行
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }
        this.segmentShift = 32 - sshift;
        this.segmentMask = ssize - 1;
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        int c = initialCapacity / ssize;
        if (c * ssize < initialCapacity)
            ++c;
        int cap = MIN_SEGMENT_TABLE_CAPACITY;
        //cap 跟ssize同理
        while (cap < c)
            cap <<= 1;
        // create segments and segments[0]
//在初始化時至初始化第一個segment，後面等用到再初始化
        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;
    }

    
    public ConcurrentHashMap(int initialCapacity, float loadFactor) {
        this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL);
    }

  
    public ConcurrentHashMap(int initialCapacity) {
        this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
    }


    public ConcurrentHashMap() {
        this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
    }

 
    public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
        this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1,
                      DEFAULT_INITIAL_CAPACITY),
             DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
        putAll(m);
    }

put

 public V put(K key, V value) {
        Segment<K,V> s;
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key);
        //hash得到segment的陣列索引
        int j = (hash >>> segmentShift) & segmentMask;
        if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
             (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
             //如果還沒有就初始化
            s = ensureSegment(j);
            //呼叫segment的put方法
        return s.put(key, hash, value, false);
    }

下面看下segment裡的put方法
 final V put(K key, int hash, V value, boolean onlyIfAbsent) {
 //先嚐試上鎖，上鎖成功 node為null，否則進scanAndLockForPut 掃描預熱並上鎖返回對應的HashEntry
            HashEntry<K,V> node = tryLock() ? null :
                scanAndLockForPut(key, hash, value);
            V oldValue;
            try {
                HashEntry<K,V>[] tab = table;
                int index = (tab.length - 1) & hash;
                HashEntry<K,V> first = entryAt(tab, index);
                //拿到對應連結串列的第一個HashEntry
                for (HashEntry<K,V> e = first;;) {
                //取出來的結果不空
                    if (e != null) {
                        K k;
			//如果找到對應的key，且hash一樣
                        if ((k = e.key) == key ||
                            (e.hash == hash && key.equals(k))) {
                            oldValue = e.value;
                            //如果時onlyIfAbsent，就不設定值，只返回舊值，否則要覆蓋的
                            if (!onlyIfAbsent) {
                                e.value = value;
                                ++modCount;
                            }
                            //找到就break
                            break;
                        }
                        //沒找到就繼續找下一個
                        e = e.next;
                    }
                    else {
                
                //吧新節點放在第一個節點位置
                        if (node != null)
                            node.setNext(first);
                        else
                            node = new HashEntry<K,V>(hash, key, value, first);
                        int c = count + 1;
                        //容量超過閾值就要擴容
                        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                            rehash(node);
                        else
                            setEntryAt(tab, index, node);
                        ++modCount;
                        count = c;
                        oldValue = null;
                        break;
                    }
                }
            } finally {
                unlock();
            }
            return oldValue;
        }
來看下 scanAndLockForPut ，這個方法主要時在等待的過程中預熱一下資料

 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) {
                    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;
                    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; // re-traverse if entry changed
                    retries = -1;
                }
            }
            return node;
        }

rehash

在put的過程中，如果size超過閾值則會rehash，來看下rehash的原始碼實現：

private void rehash(HashEntry<K,V> node) {
            HashEntry<K,V>[] oldTable = table;
            int oldCapacity = oldTable.length;
            int newCapacity = oldCapacity << 1;
            threshold = (int)(newCapacity * loadFactor);
            HashEntry<K,V>[] newTable =
                (HashEntry<K,V>[]) new HashEntry[newCapacity];
            int sizeMask = newCapacity - 1;
            for (int i = 0; i < oldCapacity ; i++) {
                HashEntry<K,V> e = oldTable[i];
                if (e != null) {
                    HashEntry<K,V> next = e.next;
                    int idx = e.hash & sizeMask;
                    if (next == null)   //  Single node on list
                        newTable[idx] = e; //納悶點 1
                    else { // Reuse consecutive sequence at same slot
                        HashEntry<K,V> lastRun = e;
                        int lastIdx = idx;
                        for (HashEntry<K,V> last = next;
                             last != null;
                             last = last.next) {
                            int k = last.hash & sizeMask;
                            if (k != lastIdx) {
                                lastIdx = k;
                                lastRun = last;
                            }
                        }
                        //rehash需要遍歷所有hashentery的資料重新分配到指定indexde hashentry下，
                        //這裡做了一個效率的優化，找到一個節點，其之後的資料都hash到一個 hashentery上，就
                        //直接把這一串放到那個對應的hashentry裡
                        newTable[lastIdx] = lastRun;//納悶點2
                        // Clone remaining nodes
                      //在這個點之前的hash到對應的entry存起來
                        for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                            V v = p.value;
                            int h = p.hash;
                            int k = h & sizeMask;
                            HashEntry<K,V> n = newTable[k];
                            newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                        }
                    }
                }
            }
            int nodeIndex = node.hash & sizeMask; // add the new node
            node.setNext(newTable[nodeIndex]);
            newTable[nodeIndex] = node;
            table = newTable;
        }

剛開始對 newTable[idx] = e;和newTable[lastIdx] = lastRun;比較納悶，起初認為後續滿足這個條件的會覆蓋原先的資料。
後來看了其他部落格瞭解到，因為擴容是擴容為原先的2倍，所以原先index上的資料 要麼還在這裡，要麼就是2*index，所以不存在其他HashEntry覆蓋原先的情況。

ensureSegment

這是put時 初始化 還未初始化的segment
  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;
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
            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的引數建立一個新的segment
                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;
    }

get

GET方法一目瞭然
 public V get(Object key) {
        Segment<K,V> s; // manually integrate access methods to reduce overhead
        HashEntry<K,V>[] tab;
        int h = hash(key);
        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

//在檢視size的時候，需要關注到每一個segment，如果直接全部上鎖，對效能不好，
//因此採用的方式時自旋兩次，判斷兩次的modCount是否一致，如果一致說明沒有修改可以返回size
//如果兩次的modCount不一致，則需要加鎖
 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 (;;) {
                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;
    }

區別

1.7與1.8的差異如下，1.8的程式碼比較多，就不貼了。
感興趣的自己看程式碼
1、1.7 使用segment分段鎖，每個鎖維護一個Node陣列
1.8 給Node加鎖，鎖粒度更小，併發效能更佳,進一步減少了併發衝突
2、1.7使用的是 segment+陣列+連結串列
1.8 使用的是陣列+連結串列+紅黑樹，在雜湊衝突較多的情況下，有較好的查詢效能