ConcurrentHashMap并發(fā)容器
　ConcurrentHashMap可以做到讀取數(shù)據(jù)不加鎖，并且其內(nèi)部的結(jié)構(gòu)可以讓其在進(jìn)行寫(xiě)操作的時(shí)候能夠?qū)㈡i的粒度保持地盡量地小，不用對(duì)整個(gè)ConcurrentHashMap加鎖。

ConcurrentHashMap的內(nèi)部結(jié)構(gòu)

　　ConcurrentHashMap為了提高本身的并發(fā)能力，在內(nèi)部采用了一個(gè)叫做Segment的結(jié)構(gòu)，一個(gè)Segment其實(shí)就是一個(gè)類Hash Table的結(jié)構(gòu)，Segment內(nèi)部維護(hù)了一個(gè)鏈表數(shù)組，我們用下面這一幅圖來(lái)看下ConcurrentHashMap的內(nèi)部結(jié)構(gòu)：

從上面的結(jié)構(gòu)我們可以了解到，ConcurrentHashMap定位一個(gè)元素的過(guò)程需要進(jìn)行兩次Hash操作，第一次Hash定位到Segment，第二次Hash定位到元素所在的鏈表的頭部，因此，這一種結(jié)構(gòu)的帶來(lái)的副作用是Hash的過(guò)程要比普通的HashMap要長(zhǎng)，但是帶來(lái)的好處是寫(xiě)操作的時(shí)候可以只對(duì)元素所在的Segment進(jìn)行加鎖即可，不會(huì)影響到其他的Segment，這樣，在最理想的情況下，ConcurrentHashMap可以最高同時(shí)支持Segment數(shù)量大小的寫(xiě)操作（剛好這些寫(xiě)操作都非常平均地分布在所有的Segment上），所以，通過(guò)這一種結(jié)構(gòu)，ConcurrentHashMap的并發(fā)能力可以大大的提高。

Segment

　　我們?cè)賮?lái)具體了解一下Segment的數(shù)據(jù)結(jié)構(gòu)：

static final class Segment<K,V> extends ReentrantLock implements Serializable {
 transient volatile int count;
 transient int modCount;
 transient int threshold;
 transient volatile HashEntry<K,V>[] table;
 final float loadFactor;
}

　　詳細(xì)解釋一下Segment里面的成員變量的意義：

• count：Segment中元素的數(shù)量
• modCount：對(duì)table的大小造成影響的操作的數(shù)量（比如put或者remove操作）
• threshold：閾值，Segment里面元素的數(shù)量超過(guò)這個(gè)值依舊就會(huì)對(duì)Segment進(jìn)行擴(kuò)容
• table：鏈表數(shù)組，數(shù)組中的每一個(gè)元素代表了一個(gè)鏈表的頭部
• loadFactor：負(fù)載因子，用于確定threshold

HashEntry

　　Segment中的元素是以HashEntry的形式存放在鏈表數(shù)組中的，看一下HashEntry的結(jié)構(gòu)：

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

　　可以看到HashEntry的一個(gè)特點(diǎn)，除了value以外，其他的幾個(gè)變量都是final的，這樣做是為了防止鏈表結(jié)構(gòu)被破壞，出現(xiàn)ConcurrentModification的情況。

ConcurrentHashMap的初始化

　　下面我們來(lái)結(jié)合源代碼來(lái)具體分析一下ConcurrentHashMap的實(shí)現(xiàn)，先看下初始化方法：

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;
 while (ssize < concurrencyLevel) {
  ++sshift;
  ssize <<= 1;
 }
 segmentShift = 32 - sshift;
 segmentMask = ssize - 1;
 this.segments = Segment.newArray(ssize);
 
 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, loadFactor);
}

 　　CurrentHashMap的初始化一共有三個(gè)參數(shù)，一個(gè)initialCapacity，表示初始的容量，一個(gè)loadFactor，表示負(fù)載參數(shù)，最后一個(gè)是concurrentLevel，代表ConcurrentHashMap內(nèi)部的Segment的數(shù)量，ConcurrentLevel一經(jīng)指定，不可改變，后續(xù)如果ConcurrentHashMap的元素?cái)?shù)量增加導(dǎo)致ConrruentHashMap需要擴(kuò)容，ConcurrentHashMap不會(huì)增加Segment的數(shù)量，而只會(huì)增加Segment中鏈表數(shù)組的容量大小，這樣的好處是擴(kuò)容過(guò)程不需要對(duì)整個(gè)ConcurrentHashMap做rehash，而只需要對(duì)Segment里面的元素做一次rehash就可以了。

　　整個(gè)ConcurrentHashMap的初始化方法還是非常簡(jiǎn)單的，先是根據(jù)concurrentLevel來(lái)new出Segment，這里Segment的數(shù)量是不大于concurrentLevel的最大的2的指數(shù)，就是說(shuō)Segment的數(shù)量永遠(yuǎn)是2的指數(shù)個(gè)，這樣的好處是方便采用移位操作來(lái)進(jìn)行hash，加快hash的過(guò)程。接下來(lái)就是根據(jù)intialCapacity確定Segment的容量的大小，每一個(gè)Segment的容量大小也是2的指數(shù)，同樣使為了加快hash的過(guò)程。

　　這邊需要特別注意一下兩個(gè)變量，分別是segmentShift和segmentMask，這兩個(gè)變量在后面將會(huì)起到很大的作用，假設(shè)構(gòu)造函數(shù)確定了Segment的數(shù)量是2的n次方，那么segmentShift就等于32減去n，而segmentMask就等于2的n次方減一。

ConcurrentHashMap的get操作

　　前面提到過(guò)ConcurrentHashMap的get操作是不用加鎖的，我們這里看一下其實(shí)現(xiàn)：

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

　　看第三行，segmentFor這個(gè)函數(shù)用于確定操作應(yīng)該在哪一個(gè)segment中進(jìn)行，幾乎對(duì)ConcurrentHashMap的所有操作都需要用到這個(gè)函數(shù)，我們看下這個(gè)函數(shù)的實(shí)現(xiàn)：

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

 　　這個(gè)函數(shù)用了位操作來(lái)確定Segment，根據(jù)傳入的hash值向右無(wú)符號(hào)右移segmentShift位，然后和segmentMask進(jìn)行與操作，結(jié)合我們之前說(shuō)的segmentShift和segmentMask的值，就可以得出以下結(jié)論：假設(shè)Segment的數(shù)量是2的n次方，根據(jù)元素的hash值的高n位就可以確定元素到底在哪一個(gè)Segment中。

　　在確定了需要在哪一個(gè)segment中進(jìn)行操作以后，接下來(lái)的事情就是調(diào)用對(duì)應(yīng)的Segment的get方法：

V get(Object key, int hash) {
 if (count != 0) { // read-volatile
  HashEntry<K,V> e = getFirst(hash);
  while (e != null) {
   if (e.hash == hash && key.equals(e.key)) {
    V v = e.value;
    if (v != null)
     return v;
    return readValueUnderLock(e); // recheck
   }
   e = e.next;
  }
 }
 return null;
}

 　　先看第二行代碼，這里對(duì)count進(jìn)行了一次判斷，其中count表示Segment中元素的數(shù)量，我們可以來(lái)看一下count的定義：

transient volatile int count;

　　 可以看到count是volatile的，實(shí)際上這里里面利用了volatile的語(yǔ)義：

　　對(duì)volatile字段的寫(xiě)入操作happens-before于每一個(gè)后續(xù)的同一個(gè)字段的讀操作。
　　因?yàn)閷?shí)際上put、remove等操作也會(huì)更新count的值，所以當(dāng)競(jìng)爭(zhēng)發(fā)生的時(shí)候，volatile的語(yǔ)義可以保證寫(xiě)操作在讀操作之前，也就保證了寫(xiě)操作對(duì)后續(xù)的讀操作都是可見(jiàn)的，這樣后面get的后續(xù)操作就可以拿到完整的元素內(nèi)容。

　　然后，在第三行，調(diào)用了getFirst()來(lái)取得鏈表的頭部：

HashEntry<K,V> getFirst(int hash) {
 HashEntry<K,V>[] tab = table;
 return tab[hash & (tab.length - 1)];
}

 　　同樣，這里也是用位操作來(lái)確定鏈表的頭部，hash值和HashTable的長(zhǎng)度減一做與操作，最后的結(jié)果就是hash值的低n位，其中n是HashTable的長(zhǎng)度以2為底的結(jié)果。

　　在確定了鏈表的頭部以后，就可以對(duì)整個(gè)鏈表進(jìn)行遍歷，看第4行，取出key對(duì)應(yīng)的value的值，如果拿出的value的值是null，則可能這個(gè)key，value對(duì)正在put的過(guò)程中，如果出現(xiàn)這種情況，那么就加鎖來(lái)保證取出的value是完整的，如果不是null，則直接返回value。

ConcurrentHashMap的put操作

　　看完了get操作，再看下put操作，put操作的前面也是確定Segment的過(guò)程，這里不再贅述，直接看關(guān)鍵的segment的put方法：

V put(K key, int hash, V value, boolean onlyIfAbsent) {
 lock();
 try {
  int c = count;
  if (c++ > threshold) // ensure capacity
   rehash();
  HashEntry<K,V>[] tab = table;
  int index = hash & (tab.length - 1);
  HashEntry<K,V> first = tab[index];
  HashEntry<K,V> e = first;
  while (e != null && (e.hash != hash || !key.equals(e.key)))
   e = e.next;
 
  V oldValue;
  if (e != null) {
   oldValue = e.value;
   if (!onlyIfAbsent)
    e.value = value;
  }
  else {
   oldValue = null;
   ++modCount;
   tab[index] = new HashEntry<K,V>(key, hash, first, value);
   count = c; // write-volatile
  }
  return oldValue;
 } finally {
  unlock();
 }
}

 　　首先對(duì)Segment的put操作是加鎖完成的，然后在第五行，如果Segment中元素的數(shù)量超過(guò)了閾值（由構(gòu)造函數(shù)中的loadFactor算出）這需要進(jìn)行對(duì)Segment擴(kuò)容，并且要進(jìn)行rehash，關(guān)于rehash的過(guò)程大家可以自己去了解，這里不詳細(xì)講了。

　　第8和第9行的操作就是getFirst的過(guò)程，確定鏈表頭部的位置。

　　第11行這里的這個(gè)while循環(huán)是在鏈表中尋找和要put的元素相同key的元素，如果找到，就直接更新更新key的value，如果沒(méi)有找到，則進(jìn)入21行這里，生成一個(gè)新的HashEntry并且把它加到整個(gè)Segment的頭部，然后再更新count的值。

ConcurrentHashMap的remove操作

　　Remove操作的前面一部分和前面的get和put操作一樣，都是定位Segment的過(guò)程，然后再調(diào)用Segment的remove方法：

V remove(Object key, int hash, Object value) {
 lock();
 try {
  int c = count - 1;
  HashEntry<K,V>[] tab = table;
  int index = hash & (tab.length - 1);
  HashEntry<K,V> first = tab[index];
  HashEntry<K,V> e = first;
  while (e != null && (e.hash != hash || !key.equals(e.key)))
   e = e.next;
 
  V oldValue = null;
  if (e != null) {
   V v = e.value;
   if (value == null || value.equals(v)) {
    oldValue = v;
    // All entries following removed node can stay
    // in list, but all preceding ones need to be
    // cloned.
    ++modCount;
    HashEntry<K,V> newFirst = e.next;
    for (HashEntry<K,V> p = first; p != e; p = p.next)
     newFirst = new HashEntry<K,V>(p.key, p.hash,
             newFirst, p.value);
    tab[index] = newFirst;
    count = c; // write-volatile
   }
  }
  return oldValue;
 } finally {
  unlock();
 }
}

 　　首先remove操作也是確定需要?jiǎng)h除的元素的位置，不過(guò)這里刪除元素的方法不是簡(jiǎn)單地把待刪除元素的前面的一個(gè)元素的next指向后面一個(gè)就完事了，我們之前已經(jīng)說(shuō)過(guò)HashEntry中的next是final的，一經(jīng)賦值以后就不可修改，在定位到待刪除元素的位置以后，程序就將待刪除元素前面的那一些元素全部復(fù)制一遍，然后再一個(gè)一個(gè)重新接到鏈表上去，看一下下面這一幅圖來(lái)了解這個(gè)過(guò)程：

　假設(shè)鏈表中原來(lái)的元素如上圖所示，現(xiàn)在要?jiǎng)h除元素3，那么刪除元素3以后的鏈表就如下圖所示：

CopyOnWriteArrayList并發(fā)容器
Copy-On-Write簡(jiǎn)稱COW，是一種用于程序設(shè)計(jì)中的優(yōu)化策略。其基本思路是，從一開(kāi)始大家都在共享同一個(gè)內(nèi)容，當(dāng)某個(gè)人想要修改這個(gè)內(nèi)容的時(shí)候，才會(huì)真正把內(nèi)容Copy出去形成一個(gè)新的內(nèi)容然后再改，這是一種延時(shí)懶惰策略。從JDK1.5開(kāi)始Java并發(fā)包里提供了兩個(gè)使用CopyOnWrite機(jī)制實(shí)現(xiàn)的并發(fā)容器,它們是CopyOnWriteArrayList和CopyOnWriteArraySet。CopyOnWrite容器非常有用，可以在非常多的并發(fā)場(chǎng)景中使用到。

什么是CopyOnWrite容器

　　CopyOnWrite容器即寫(xiě)時(shí)復(fù)制的容器。通俗的理解是當(dāng)我們往一個(gè)容器添加元素的時(shí)候，不直接往當(dāng)前容器添加，而是先將當(dāng)前容器進(jìn)行Copy，復(fù)制出一個(gè)新的容器，然后新的容器里添加元素，添加完元素之后，再將原容器的引用指向新的容器。這樣做的好處是我們可以對(duì)CopyOnWrite容器進(jìn)行并發(fā)的讀，而不需要加鎖，因?yàn)楫?dāng)前容器不會(huì)添加任何元素。所以CopyOnWrite容器也是一種讀寫(xiě)分離的思想，讀和寫(xiě)不同的容器。

CopyOnWriteArrayList的實(shí)現(xiàn)原理

　　在使用CopyOnWriteArrayList之前，我們先閱讀其源碼了解下它是如何實(shí)現(xiàn)的。以下代碼是向CopyOnWriteArrayList中add方法的實(shí)現(xiàn)（向CopyOnWriteArrayList里添加元素），可以發(fā)現(xiàn)在添加的時(shí)候是需要加鎖的，否則多線程寫(xiě)的時(shí)候會(huì)Copy出N個(gè)副本出來(lái)。

/**
  * Appends the specified element to the end of this list.
  *
  * @param e element to be appended to this list
  * @return <tt>true</tt> (as specified by {@link Collection#add})
  */
 public boolean add(E e) {
 final ReentrantLock lock = this.lock;
 lock.lock();
 try {
  Object[] elements = getArray();
  int len = elements.length;
  Object[] newElements = Arrays.copyOf(elements, len + 1);
  newElements[len] = e;
  setArray(newElements);
  return true;
 } finally {
  lock.unlock();
 }
 }

 　　讀的時(shí)候不需要加鎖，如果讀的時(shí)候有多個(gè)線程正在向CopyOnWriteArrayList添加數(shù)據(jù)，讀還是會(huì)讀到舊的數(shù)據(jù)，因?yàn)閷?xiě)的時(shí)候不會(huì)鎖住舊的CopyOnWriteArrayList。

public E get(int index) {
 return get(getArray(), index);
}

 　　JDK中并沒(méi)有提供CopyOnWriteMap，我們可以參考CopyOnWriteArrayList來(lái)實(shí)現(xiàn)一個(gè)，基本代碼如下：

import java.util.Collection;
import java.util.Map;
import java.util.Set;
 
public class CopyOnWriteMap<K, V> implements Map<K, V>, Cloneable {
 private volatile Map<K, V> internalMap;
 
 public CopyOnWriteMap() {
  internalMap = new HashMap<K, V>();
 }
 
 public V put(K key, V value) {
 
  synchronized (this) {
   Map<K, V> newMap = new HashMap<K, V>(internalMap);
   V val = newMap.put(key, value);
   internalMap = newMap;
   return val;
  }
 }
 
 public V get(Object key) {
  return internalMap.get(key);
 }
 
 public void putAll(Map<? extends K, ? extends V> newData) {
  synchronized (this) {
   Map<K, V> newMap = new HashMap<K, V>(internalMap);
   newMap.putAll(newData);
   internalMap = newMap;
  }
 }
}

 　　實(shí)現(xiàn)很簡(jiǎn)單，只要了解了CopyOnWrite機(jī)制，我們可以實(shí)現(xiàn)各種CopyOnWrite容器，并且在不同的應(yīng)用場(chǎng)景中使用。

CopyOnWrite的應(yīng)用場(chǎng)景

　　CopyOnWrite并發(fā)容器用于讀多寫(xiě)少的并發(fā)場(chǎng)景。比如白名單，黑名單，商品類目的訪問(wèn)和更新場(chǎng)景，假如我們有一個(gè)搜索網(wǎng)站，用戶在這個(gè)網(wǎng)站的搜索框中，輸入關(guān)鍵字搜索內(nèi)容，但是某些關(guān)鍵字不允許被搜索。這些不能被搜索的關(guān)鍵字會(huì)被放在一個(gè)黑名單當(dāng)中，黑名單每天晚上更新一次。當(dāng)用戶搜索時(shí)，會(huì)檢查當(dāng)前關(guān)鍵字在不在黑名單當(dāng)中，如果在，則提示不能搜索。實(shí)現(xiàn)代碼如下：

package com.ifeve.book;
 
import java.util.Map;
 
import com.ifeve.book.forkjoin.CopyOnWriteMap;
 
/**
 * 黑名單服務(wù)
 *
 * @author fangtengfei
 *
 */
public class BlackListServiceImpl {
 
 private static CopyOnWriteMap<String, Boolean> blackListMap = new CopyOnWriteMap<String, Boolean>(
   1000);
 
 public static boolean isBlackList(String id) {
  return blackListMap.get(id) == null ? false : true;
 }
 
 public static void addBlackList(String id) {
  blackListMap.put(id, Boolean.TRUE);
 }
 
 /**
  * 批量添加黑名單
  *
  * @param ids
  */
 public static void addBlackList(Map<String,Boolean> ids) {
  blackListMap.putAll(ids);
 }
 
}

 　　代碼很簡(jiǎn)單，但是使用CopyOnWriteMap需要注意兩件事情：

　　1. 減少擴(kuò)容開(kāi)銷。根據(jù)實(shí)際需要，初始化CopyOnWriteMap的大小，避免寫(xiě)時(shí)CopyOnWriteMap擴(kuò)容的開(kāi)銷。

　　2. 使用批量添加。因?yàn)槊看翁砑?，容器每次都?huì)進(jìn)行復(fù)制，所以減少添加次數(shù)，可以減少容器的復(fù)制次數(shù)。如使用上面代碼里的addBlackList方法。

CopyOnWrite的缺點(diǎn)

　　CopyOnWrite容器有很多優(yōu)點(diǎn)，但是同時(shí)也存在兩個(gè)問(wèn)題，即內(nèi)存占用問(wèn)題和數(shù)據(jù)一致性問(wèn)題。所以在開(kāi)發(fā)的時(shí)候需要注意一下。

　　內(nèi)存占用問(wèn)題。因?yàn)镃opyOnWrite的寫(xiě)時(shí)復(fù)制機(jī)制，所以在進(jìn)行寫(xiě)操作的時(shí)候，內(nèi)存里會(huì)同時(shí)駐扎兩個(gè)對(duì)象的內(nèi)存，舊的對(duì)象和新寫(xiě)入的對(duì)象（注意:在復(fù)制的時(shí)候只是復(fù)制容器里的引用，只是在寫(xiě)的時(shí)候會(huì)創(chuàng)建新對(duì)象添加到新容器里，而舊容器的對(duì)象還在使用，所以有兩份對(duì)象內(nèi)存）。如果這些對(duì)象占用的內(nèi)存比較大，比如說(shuō)200M左右，那么再寫(xiě)入100M數(shù)據(jù)進(jìn)去，內(nèi)存就會(huì)占用300M，那么這個(gè)時(shí)候很有可能造成頻繁的Yong GC和Full GC。之前我們系統(tǒng)中使用了一個(gè)服務(wù)由于每晚使用CopyOnWrite機(jī)制更新大對(duì)象，造成了每晚15秒的Full GC，應(yīng)用響應(yīng)時(shí)間也隨之變長(zhǎng)。

　　針對(duì)內(nèi)存占用問(wèn)題，可以通過(guò)壓縮容器中的元素的方法來(lái)減少大對(duì)象的內(nèi)存消耗，比如，如果元素全是10進(jìn)制的數(shù)字，可以考慮把它壓縮成36進(jìn)制或64進(jìn)制。或者不使用CopyOnWrite容器，而使用其他的并發(fā)容器，如ConcurrentHashMap。

　　數(shù)據(jù)一致性問(wèn)題。CopyOnWrite容器只能保證數(shù)據(jù)的最終一致性，不能保證數(shù)據(jù)的實(shí)時(shí)一致性。所以如果你希望寫(xiě)入的的數(shù)據(jù)，馬上能讀到，請(qǐng)不要使用CopyOnWrite容器。

最新評(píng)論