Java分布式應(yīng)用學(xué)習(xí)筆記04JDK的并發(fā)包的集合總結(jié)

1、Java分布式應(yīng)用學(xué)習(xí)筆記04JDK的并發(fā)包的集合總結(jié)劉巖前言平時(shí)咱們使用的HashMap、ArrayList等等容器集合包都存在線程安全的問題，看過JDK源碼的各位朋友們知道這些實(shí)現(xiàn)類底層，為了性能，都沒有對(duì)這些集合的操作方法做加鎖或者副本傳遞機(jī)制，只有Vector和Stack是線程安全的，大家可以看它們的源碼，底層方法是以在方法上加上synchronized作為代價(jià)的，換句話說是用時(shí)間換取空間的方式。Sun JDK對(duì)多線程并發(fā)環(huán)境下做了很多并發(fā)的解決方案，其類大都在.*下面，此包下的類和java.util.*包下面的集合類，在使用上幾乎沒什么太大分別，想想也是啊！他們都是實(shí)現(xiàn)接口規(guī)范：Li

2、st、Set、Map的。只要接口規(guī)范不變，那么在使用上也不應(yīng)該有何變化，實(shí)現(xiàn)機(jī)制是一個(gè)側(cè)重低概率并發(fā)或者就是單線程環(huán)境下，并發(fā)包則側(cè)重高并發(fā)情況的系統(tǒng)。大家可以看看Tomcat的源代碼，其中.ApplicationContext里面就使用到了并發(fā)包，因?yàn)門omcat作為Web容器一定要接受來自各個(gè)客戶端的request，進(jìn)而分配Web應(yīng)用上下文信息，應(yīng)用參數(shù)key-value值等等。又得滿足并發(fā)的請(qǐng)求、又得滿足性能所需，所以它使用JDK的并發(fā)包。在使用層面上，筆者并不作過多介紹，可以參考非并發(fā)包的使用。至于這些非并發(fā)包的底層實(shí)現(xiàn)方式可以參考筆者的blog關(guān)于Java基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)的基礎(chǔ)知識(shí)，而是

3、介紹一下這些并發(fā)包的底層機(jī)制和性能對(duì)比，在多線程環(huán)境下，用并發(fā)包和不用并發(fā)包的時(shí)間效率對(duì)比，空間資源效率不用比了，肯定單線程那些包消耗的比多線程消耗的小得多，畢竟做任何事都是要付出代價(jià)的。Map的并發(fā)包Map接口在并發(fā)包下的實(shí)現(xiàn)叫做。它實(shí)現(xiàn)了ConcurrentMap接口，而ConcurrentMap接口又是繼承自Map接口的擴(kuò)展。先看看它是如何實(shí)現(xiàn)put操作的。 public V put(K key, V value) if (value = null) throw new NullPointerException(); int hash = hash(key.hashCode(); ret

4、urn segmentFor(hash).put(key, hash, value, false); 首先判斷值是否為空，空值不必要存儲(chǔ)，之后根據(jù)key的哈希值計(jì)算一個(gè)hash值。根據(jù)計(jì)算出的hash值去獲取segment對(duì)象。找到了segment對(duì)象后調(diào)用該對(duì)象的put方法完成操作。Segment是ConcurrentHashMap的內(nèi)部類其底層原理使用一個(gè)transient volatile HashEntry table;進(jìn)行存取?，F(xiàn)在再看segment內(nèi)的put源碼 V put(K key, int hash, V value, boolean onlyIfAbsent) lock()

5、; try int c = count; if (c+ threshold) / ensure capacity rehash(); HashEntry tab = table; int index = hash & (tab.length - 1); HashEntry first = tabindex; HashEntry e = first; while (e != null & (e.hash != hash | !key.equals(e.key) e = e.next; V oldValue; if (e != null) oldValue = e.value; if (!only

6、IfAbsent) e.value = value; else oldValue = null; +modCount; tabindex = new HashEntry(key, hash, first, value); count = c; / write-volatile return oldValue; finally unlock(); 首先是進(jìn)行加鎖操作，之后就是進(jìn)行數(shù)組大小的判斷，如果容量不夠，則需要擴(kuò)充。之后再通過對(duì)hash值的按位與的操作后，得到了這個(gè)key所要放置的位置。有了位置了，再看HashEntry數(shù)組組成的對(duì)象鏈，是否已經(jīng)有key，如果有了，覆蓋value操作，如果沒

7、有，創(chuàng)建一個(gè)新的HashEntry對(duì)象，重新組成HashEntry鏈表，最后進(jìn)行解鎖操作。所以直線我們關(guān)心的在put中會(huì)出現(xiàn)的線程安全問題，看了源碼后是不是就解決了。想想除了put操作會(huì)出現(xiàn)線程不安全的隱患外，我們來看看remove操作。刪除操作代碼原理與put操作類似，也是通過hash值找到那個(gè)segment對(duì)象，之后調(diào)用segment的remove方法去完成真正的操作。真正的操作也是先加鎖，之后迭代HashEntry，直到找到了傳入的hash值相同的。找到了刪之，重新建立鏈表！找不到，over，然后釋放對(duì)象鎖。在ConcurrentHashMap的get、containsKey等等這種不破

8、壞原子性的讀?。╮ead）操作可以說大部分情況下沒有進(jìn)行加鎖操作，即便像get加了鎖操作，也是極其輕量的，僅僅是加鎖了一行很簡單的讀取操作代碼，如下 V readValueUnderLock(HashEntry e) lock(); try return e.value; finally unlock(); 下面我們來看看性能，在此所說的性能僅僅指時(shí)間執(zhí)行效率。使用一般HashMap包的程序如下package threadConcurrent.hashMap;import java.util.HashMap;import java.util.Map;import java.util.concu

9、rrent.ExecutorService;import java.util.concurrent.Executors;/* author liuyan * */public class PubHashMap implements Runnable final static int ThreadSIZE = 2500;final static int elSize = 500;int threadNum;public PubHashMap(int threadNum) this.threadNum = threadNum;Overridepublic void run() Map hashMa

10、p = new HashMap();for (int i = 0; i elSize; i+) hashMap.put(i + + threadNum, i + + threadNum);/* * param args */public static void main(String args) / 啟用線程池ExecutorService exec = Executors.newFixedThreadPool(ThreadSIZE);long startTime = System.currentTimeMillis();for (int index = 0; index = ThreadSI

11、ZE; index+) exec.execute(new PubHashMap(index);long endTime = System.currentTimeMillis();exec.shutdown();System.out.println(消耗時(shí)間： + (endTime - startTime) + ms);啟動(dòng)2500個(gè)線程，每個(gè)線程往HashMap中添加500個(gè)字符串元素。執(zhí)行多次后給出一個(gè)平均時(shí)間吧消耗時(shí)間：1753ms使用并發(fā)包程序如下package threadConcurrent.hashMap;import java.util.Map;import java.util.

12、concurrent.ConcurrentHashMap;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/* author liuyan * */public class PutConcurrentHashMap implements Runnable final static int ThreadSIZE = 2500;final static int elSize = 500;int threadNum;public PutConcurrentHashMap(int thr

13、eadNum) this.threadNum = threadNum;Overridepublic void run() Map concurrentHashMap = new ConcurrentHashMap();for (int i = 0; i elSize; i+) concurrentHashMap.put(i + + threadNum, i + + threadNum);/* * param args */public static void main(String args) / 啟用線程池ExecutorService exec = Executors.newFixedTh

14、readPool(ThreadSIZE);long startTime = System.currentTimeMillis();for (int index = 0; index = ThreadSIZE; index+) exec.execute(new PutConcurrentHashMap(index);long endTime = System.currentTimeMillis();exec.shutdown();System.out.println(消耗時(shí)間： + (endTime - startTime) + ms);也是多次執(zhí)行后，得出一個(gè)平均時(shí)間吧消耗時(shí)間：1869ms時(shí)

15、間消耗上差不多哈。在集合元素越來越多的情況下，在解決線程安全的同時(shí)保證了時(shí)間執(zhí)行熬費(fèi)上幾乎和非線程安全的Map持平。所以在并發(fā)條件下不必自己解決Map的線程安全問題，直接放心使用JDK自己的并發(fā)Map包即可，時(shí)間性能上還能保證。List的并發(fā)包可以在高并發(fā)環(huán)境下使用代替java.util.ArrayList。對(duì)于添加元素的操作，底層并不像Map那么復(fù)雜，就是利用了數(shù)組的copy功能和加鎖機(jī)制public boolean add(E e) final ReentrantLock lock = this.lock;lock.lock();try Object elements = getArray

16、();int len = elements.length;Object newElements = Arrays.copyOf(elements, len + 1);newElementslen = e;setArray(newElements);return true; finally lock.unlock();它是使用ReentrantLock進(jìn)行的加鎖，之后獲得數(shù)組進(jìn)行copy操作，之后數(shù)組個(gè)數(shù)加一。將新元素填充，之后再對(duì)局部變量進(jìn)行一下set操作，最后就是解鎖操作。至于remove操作，和add的原理一樣public E remove(int index) final Reentra

17、ntLock lock = this.lock;lock.lock();try Object elements = getArray();int len = elements.length;Object oldValue = elementsindex;int numMoved = len - index - 1;if (numMoved = 0)setArray(Arrays.copyOf(elements, len - 1);else Object newElements = new Objectlen - 1;System.arraycopy(elements, 0, newElemen

18、ts, 0, index);System.arraycopy(elements, index + 1, newElements, index,numMoved);setArray(newElements);return (E) oldValue; finally lock.unlock();在枷鎖對(duì)兒中間，先找到標(biāo)記下的數(shù)組元素，之后創(chuàng)建一個(gè)新的臨時(shí)數(shù)組，進(jìn)行copy，將要?jiǎng)h除的對(duì)象元素剔除出去！返回被刪除元素對(duì)象。做add操作性能與ArrayList進(jìn)行對(duì)比，線程運(yùn)行400個(gè)，添加元素?cái)?shù)是2000個(gè)。對(duì)比平均之后發(fā)現(xiàn)運(yùn)行的時(shí)間也相差不是很多。并發(fā)情況下，CopyOnWriteArrayLis

19、t比ArrayList略快了那么一點(diǎn)點(diǎn)。get幾乎和ArrayList沒差別，直接從數(shù)組中找第index個(gè)元素。Set的并發(fā)CopyOnWriteArraySet和CopyOnWriteArrayList底層實(shí)現(xiàn)差不多，就是在添加元素的時(shí)候需要對(duì)對(duì)象進(jìn)行唯一性判斷，如果對(duì)象數(shù)組已經(jīng)含有重復(fù)的元素，不進(jìn)行增加處理。在此不再贅述。Queue的并發(fā)隊(duì)列的并發(fā)類是，從類名字上大家估計(jì)就能猜出來了，底層使用的依然是數(shù)組。這個(gè)ArrayBlockingQueue是繼承自原始的java.util.AbstractQueue，所以很多方法在父類里面已經(jīng)有了，只是對(duì)于關(guān)鍵方法入隊(duì)列、出隊(duì)列操作加入了鎖對(duì)兒機(jī)制。

20、入隊(duì)列元素操作源碼如下 public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException if (e = null) throw new NullPointerException();long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try for (;) if (count != items.length) insert(e); return t

21、rue; if (nanos = 0) return false; try nanos = notFull.awaitNanos(nanos); catch (InterruptedException ie) notFull.signal(); / propagate to non-interrupted thread throw ie; finally lock.unlock(); 數(shù)組未滿情況下，執(zhí)行insert操作的時(shí)候，如果數(shù)組滿了，則進(jìn)行等待，單位是納秒。如果超時(shí)或者被喚醒了，那么再次判斷是否數(shù)組已滿，如果線程被打斷直接拋出異常。出隊(duì)列方法和入隊(duì)列差不多，不再贅述。AtomicXXXX的原子類并發(fā)包還提供了支持原子操作的Atomic系列類，我們舉一個(gè)具有代表性的類AtomicInteger類，通常我們使用計(jì)數(shù)器操作的時(shí)候，一般為了避免線程安全的問題，在方法上加鎖操作。有了并發(fā)包下的原子系列類，我們直接使用即可。關(guān)鍵使用代碼片段如下public static int getSum() return sum.incrementAndGet();其自增方法底層片段最關(guān)鍵是這么一句 if (compareAndSet(current, next) return