第 1 部分

http://www.ibm.com/developerworks/cn/java/j-5things4.html

Concurrent Collections 是 Java? 5 的巨大附加產(chǎn)品，但是在關(guān)于注釋和泛型的爭執(zhí)中很多 Java 開發(fā)人員忽視了它們。此外（或者更老實(shí)地說），許多開發(fā)人員避免使用這個(gè)數(shù)據(jù)包，因?yàn)樗麄冋J(rèn)為它一定很復(fù)雜，就像它所要解決的問題一樣。

事實(shí)上，java.util.concurrent?包含許多類，能夠有效解決普通的并發(fā)問題，無需復(fù)雜工序。閱讀本文，了解?java.util.concurrent?類，比如?CopyOnWriteArrayList?和BlockingQueue?如何幫助您解決多線程編程的棘手問題。

1. TimeUnit

盡管本質(zhì)上?不是 Collections 類，但?java.util.concurrent.TimeUnit?枚舉讓代碼更易讀懂。使用?TimeUnit?將使用您的方法或 API 的開發(fā)人員從毫秒的 “暴政” 中解放出來。

TimeUnit?包括所有時(shí)間單位，從?MILLISECONDS?和?MICROSECONDS?到?DAYS?和?HOURS，這就意味著它能夠處理一個(gè)開發(fā)人員所需的幾乎所有的時(shí)間范圍類型。同時(shí)，因?yàn)樵诹信e上聲明了轉(zhuǎn)換方法，在時(shí)間加快時(shí)，將?HOURS?轉(zhuǎn)換回?MILLISECONDS?甚至變得更容易。

回頁首

2. CopyOnWriteArrayList

創(chuàng)建數(shù)組的全新副本是過于昂貴的操作，無論是從時(shí)間上，還是從內(nèi)存開銷上，因此在通常使用中很少考慮；開發(fā)人員往往求助于使用同步的ArrayList。然而，這也是一個(gè)成本較高的選擇，因?yàn)槊慨?dāng)您跨集合內(nèi)容進(jìn)行迭代時(shí)，您就不得不同步所有操作，包括讀和寫，以此保證一致性。

這又讓成本結(jié)構(gòu)回到這樣一個(gè)場(chǎng)景：需多讀者都在讀取?ArrayList，但是幾乎沒人會(huì)去修改它。

CopyOnWriteArrayList?是個(gè)巧妙的小寶貝，能解決這一問題。它的 Javadoc 將?CopyOnWriteArrayList?定義為一個(gè) “ArrayList?的線程安全變體，在這個(gè)變體中所有易變操作（添加，設(shè)置等）可以通過復(fù)制全新的數(shù)組來實(shí)現(xiàn)”。

集合從內(nèi)部將它的內(nèi)容復(fù)制到一個(gè)沒有修改的新數(shù)組，這樣讀者訪問數(shù)組內(nèi)容時(shí)就不會(huì)產(chǎn)生同步成本（因?yàn)樗麄儚膩聿皇窃谝鬃償?shù)據(jù)上操作）。

本質(zhì)上講，CopyOnWriteArrayList?很適合處理?ArrayList?經(jīng)常讓我們失敗的這種場(chǎng)景：讀取頻繁，但很少有寫操作的集合，例如 JavaBean 事件的?Listeners。

回頁首

3. BlockingQueue

BlockingQueue?接口表示它是一個(gè)?Queue，意思是它的項(xiàng)以先入先出（FIFO）順序存儲(chǔ)。在特定順序插入的項(xiàng)以相同的順序檢索 — 但是需要附加保證，從空隊(duì)列檢索一個(gè)項(xiàng)的任何嘗試都會(huì)阻塞調(diào)用線程，直到這個(gè)項(xiàng)準(zhǔn)備好被檢索。同理，想要將一個(gè)項(xiàng)插入到滿隊(duì)列的嘗試也會(huì)導(dǎo)致阻塞調(diào)用線程，直到隊(duì)列的存儲(chǔ)空間可用。

BlockingQueue?干凈利落地解決了如何將一個(gè)線程收集的項(xiàng)“傳遞”給另一線程用于處理的問題，無需考慮同步問題。Java Tutorial 的 Guarded Blocks 試用版就是一個(gè)很好的例子。它構(gòu)建一個(gè)單插槽綁定的緩存，當(dāng)新的項(xiàng)可用，而且插槽也準(zhǔn)備好接受新的項(xiàng)時(shí)，使用手動(dòng)同步和wait()/notifyAll()?在線程之間發(fā)信。（詳見?Guarded Blocks 實(shí)現(xiàn)。）

盡管 Guarded Blocks 教程中的代碼有效，但是它耗時(shí)久，混亂，而且也并非完全直觀。退回到 Java 平臺(tái)較早的時(shí)候，沒錯(cuò)，Java 開發(fā)人員不得不糾纏于這種代碼；但現(xiàn)在是 2010 年 — 情況難道沒有改善？

清單 1 顯示了 Guarded Blocks 代碼的重寫版，其中我使用了一個(gè)?ArrayBlockingQueue，而不是手寫的?Drop。

清單 1. BlockingQueue

import java.util.*; import java.util.concurrent.*;class Producerimplements Runnable {private BlockingQueue<String> drop;List<String> messages = Arrays.asList("Mares eat oats","Does eat oats","Little lambs eat ivy","Wouldn't you eat ivy too?");public Producer(BlockingQueue<String> d) { this.drop = d; }public void run(){try{for (String s : messages)drop.put(s);drop.put("DONE");}catch (InterruptedException intEx){System.out.println("Interrupted! " + "Last one out, turn out the lights!");}} }class Consumerimplements Runnable {private BlockingQueue<String> drop;public Consumer(BlockingQueue<String> d) { this.drop = d; }public void run(){try{String msg = null;while (!((msg = drop.take()).equals("DONE")))System.out.println(msg);}catch (InterruptedException intEx){System.out.println("Interrupted! " + "Last one out, turn out the lights!");}} }public class ABQApp {public static void main(String[] args){BlockingQueue<String> drop = new ArrayBlockingQueue(1, true);(new Thread(new Producer(drop))).start();(new Thread(new Consumer(drop))).start();} }

ArrayBlockingQueue?還體現(xiàn)了“公平” — 意思是它為讀取器和編寫器提供線程先入先出訪問。這種替代方法是一個(gè)更有效，但又冒窮盡部分線程風(fēng)險(xiǎn)的政策。（即，允許一些讀取器在其他讀取器鎖定時(shí)運(yùn)行效率更高，但是您可能會(huì)有讀取器線程的流持續(xù)不斷的風(fēng)險(xiǎn)，導(dǎo)致編寫器無法進(jìn)行工作。）

注意 Bug！

順便說一句，如果您注意到 Guarded Blocks 包含一個(gè)重大 bug，那么您是對(duì)的 — 如果開發(fā)人員在?main()?中的Drop?實(shí)例上同步，會(huì)出現(xiàn)什么情況呢？

BlockingQueue?還支持接收時(shí)間參數(shù)的方法，時(shí)間參數(shù)表明線程在返回信號(hào)故障以插入或者檢索有關(guān)項(xiàng)之前需要阻塞的時(shí)間。這么做會(huì)避免非綁定的等待，這對(duì)一個(gè)生產(chǎn)系統(tǒng)是致命的，因?yàn)橐粋€(gè)非綁定的等待會(huì)很容易導(dǎo)致需要重啟的系統(tǒng)掛起。

回頁首

4. ConcurrentMap

Map?有一個(gè)微妙的并發(fā) bug，這個(gè) bug 將許多不知情的 Java 開發(fā)人員引入歧途。ConcurrentMap?是最容易的解決方案。

當(dāng)一個(gè)?Map?被從多個(gè)線程訪問時(shí)，通常使用?containsKey()?或者?get()?來查看給定鍵是否在存儲(chǔ)鍵/值對(duì)之前出現(xiàn)。但是即使有一個(gè)同步的Map，線程還是可以在這個(gè)過程中潛入，然后奪取對(duì)?Map?的控制權(quán)。問題是，在對(duì)?put()?的調(diào)用中，鎖在?get()?開始時(shí)獲取，然后在可以再次獲取鎖之前釋放。它的結(jié)果是個(gè)競爭條件：這是兩個(gè)線程之間的競爭，結(jié)果也會(huì)因誰先運(yùn)行而不同。

如果兩個(gè)線程幾乎同時(shí)調(diào)用一個(gè)方法，兩者都會(huì)進(jìn)行測(cè)試，調(diào)用 put，在處理中丟失第一線程的值。幸運(yùn)的是，ConcurrentMap?接口支持許多附加方法，它們?cè)O(shè)計(jì)用于在一個(gè)鎖下進(jìn)行兩個(gè)任務(wù)：putIfAbsent()，例如，首先進(jìn)行測(cè)試，然后僅當(dāng)鍵沒有存儲(chǔ)在?Map?中時(shí)進(jìn)行 put。

回頁首

5. SynchronousQueues

根據(jù) Javadoc，SynchronousQueue?是個(gè)有趣的東西：

這是一個(gè)阻塞隊(duì)列，其中，每個(gè)插入操作必須等待另一個(gè)線程的對(duì)應(yīng)移除操作，反之亦然。一個(gè)同步隊(duì)列不具有任何內(nèi)部容量，甚至不具有 1 的容量。

本質(zhì)上講，SynchronousQueue?是之前提過的?BlockingQueue?的又一實(shí)現(xiàn)。它給我們提供了在線程之間交換單一元素的極輕量級(jí)方法，使用ArrayBlockingQueue?使用的阻塞語義。在清單 2 中，我重寫了?清單 1?的代碼，使用?SynchronousQueue?替代?ArrayBlockingQueue：

清單 2. SynchronousQueue

import java.util.*; import java.util.concurrent.*;class Producerimplements Runnable {private BlockingQueue<String> drop;List<String> messages = Arrays.asList("Mares eat oats","Does eat oats","Little lambs eat ivy","Wouldn't you eat ivy too?");public Producer(BlockingQueue<String> d) { this.drop = d; }public void run(){try{for (String s : messages)drop.put(s);drop.put("DONE");}catch (InterruptedException intEx){System.out.println("Interrupted! " + "Last one out, turn out the lights!");}} }class Consumerimplements Runnable {private BlockingQueue<String> drop;public Consumer(BlockingQueue<String> d) { this.drop = d; }public void run(){try{String msg = null;while (!((msg = drop.take()).equals("DONE")))System.out.println(msg);}catch (InterruptedException intEx){System.out.println("Interrupted! " + "Last one out, turn out the lights!");}} }public class SynQApp {public static void main(String[] args){BlockingQueue<String> drop = new SynchronousQueue<String>();(new Thread(new Producer(drop))).start();(new Thread(new Consumer(drop))).start();} }

實(shí)現(xiàn)代碼看起來幾乎相同，但是應(yīng)用程序有額外獲益：SynchronousQueue?允許在隊(duì)列進(jìn)行一個(gè)插入，只要有一個(gè)線程等著使用它。

在實(shí)踐中，SynchronousQueue?類似于 Ada 和 CSP 等語言中可用的 “會(huì)合通道”。這些通道有時(shí)在其他環(huán)境中也稱為 “連接”，這樣的環(huán)境包括 .NET （見?參考資料）。

回頁首

結(jié)束語

當(dāng) Java 運(yùn)行時(shí)知識(shí)庫提供便利、預(yù)置的并發(fā)性時(shí)，為什么還要苦苦掙扎，試圖將并發(fā)性導(dǎo)入到您的 Collections 類？本系列的下一篇文章將會(huì)進(jìn)一步探討?java.util.concurrent?名稱空間的內(nèi)容。

第 2 部分

http://www.ibm.com/developerworks/cn/java/j-5things5.html

并發(fā) Collections 提供了線程安全、經(jīng)過良好調(diào)優(yōu)的數(shù)據(jù)結(jié)構(gòu)，簡化了并發(fā)編程。然而，在一些情形下，開發(fā)人員需要更進(jìn)一步，思考如何調(diào)節(jié)和/或限制線程執(zhí)行。由于?java.util.concurrent?的總體目標(biāo)是簡化多線程編程，您可能希望該包包含同步實(shí)用程序，而它確實(shí)包含。

本文是?第 1 部分?的延續(xù)，將介紹幾個(gè)比核心語言原語（監(jiān)視器）更高級(jí)的同步結(jié)構(gòu)，但它們還未包含在 Collection 類中。一旦您了解了這些鎖和門的用途，使用它們將非常直觀。

關(guān)于本系列

您覺得自己懂 Java 編程？事實(shí)是，大多數(shù)開發(fā)人員都只領(lǐng)會(huì)到了 Java 平臺(tái)的皮毛，所學(xué)也只夠應(yīng)付工作。在本系列?中，Ted Neward 深度挖掘 Java 平臺(tái)的核心功能，揭示一些鮮為人知的事實(shí)，幫助您解決最棘手的編程困難。

1. Semaphore

在一些企業(yè)系統(tǒng)中，開發(fā)人員經(jīng)常需要限制未處理的特定資源請(qǐng)求（線程/操作）數(shù)量，事實(shí)上，限制有時(shí)候能夠提高系統(tǒng)的吞吐量，因?yàn)樗鼈儨p少了對(duì)特定資源的爭用。盡管完全可以手動(dòng)編寫限制代碼，但使用 Semaphore 類可以更輕松地完成此任務(wù)，它將幫您執(zhí)行限制，如清單 1 所示：

清單 1. 使用 Semaphore 執(zhí)行限制

import java.util.*;import java.util.concurrent.*;public class SemApp {public static void main(String[] args){Runnable limitedCall = new Runnable() {final Random rand = new Random();final Semaphore available = new Semaphore(3);int count = 0;public void run(){int time = rand.nextInt(15);int num = count++;try{available.acquire();System.out.println("Executing " + "long-running action for " + time + " seconds... #" + num);Thread.sleep(time * 1000);System.out.println("Done with #" + num + "!");available.release();}catch (InterruptedException intEx){intEx.printStackTrace();}}};for (int i=0; i<10; i++)new Thread(limitedCall).start();} }

即使本例中的 10 個(gè)線程都在運(yùn)行（您可以對(duì)運(yùn)行?SemApp?的 Java 進(jìn)程執(zhí)行?jstack?來驗(yàn)證），但只有 3 個(gè)線程是活躍的。在一個(gè)信號(hào)計(jì)數(shù)器釋放之前，其他 7 個(gè)線程都處于空閑狀態(tài)。（實(shí)際上，Semaphore?類支持一次獲取和釋放多個(gè)?permit，但這不適用于本場(chǎng)景。）

回頁首

2. CountDownLatch

如果?Semaphore?是允許一次進(jìn)入一個(gè)（這可能會(huì)勾起一些流行夜總會(huì)的保安的記憶）線程的并發(fā)性類，那么?CountDownLatch?就像是賽馬場(chǎng)的起跑門柵。此類持有所有空閑線程，直到滿足特定條件，這時(shí)它將會(huì)一次釋放所有這些線程。

清單 2. CountDownLatch：讓我們?nèi)ベ愸R吧！

import java.util.*; import java.util.concurrent.*;class Race {private Random rand = new Random();private int distance = rand.nextInt(250);private CountDownLatch start;private CountDownLatch finish;private List<String> horses = new ArrayList<String>();public Race(String... names){this.horses.addAll(Arrays.asList(names));}public void run()throws InterruptedException{System.out.println("And the horses are stepping up to the gate...");final CountDownLatch start = new CountDownLatch(1);final CountDownLatch finish = new CountDownLatch(horses.size());final List<String> places = Collections.synchronizedList(new ArrayList<String>());for (final String h : horses){new Thread(new Runnable() {public void run() {try{System.out.println(h + " stepping up to the gate...");start.await();int traveled = 0;while (traveled < distance){// In a 0-2 second period of time....Thread.sleep(rand.nextInt(3) * 1000);// ... a horse travels 0-14 lengthstraveled += rand.nextInt(15);System.out.println(h + " advanced to " + traveled + "!");}finish.countDown();System.out.println(h + " crossed the finish!");places.add(h);}catch (InterruptedException intEx){System.out.println("ABORTING RACE!!!");intEx.printStackTrace();}}}).start();}System.out.println("And... they're off!");start.countDown(); finish.await();System.out.println("And we have our winners!");System.out.println(places.get(0) + " took the gold...");System.out.println(places.get(1) + " got the silver...");System.out.println("and " + places.get(2) + " took home the bronze.");} }public class CDLApp {public static void main(String[] args)throws InterruptedException, java.io.IOException{System.out.println("Prepping...");Race r = new Race("Beverly Takes a Bath","RockerHorse","Phineas","Ferb","Tin Cup","I'm Faster Than a Monkey","Glue Factory Reject");System.out.println("It's a race of " + r.getDistance() + " lengths");System.out.println("Press Enter to run the race....");System.in.read();r.run();} }

注意，在?清單 2?中，CountDownLatch?有兩個(gè)用途：首先，它同時(shí)釋放所有線程，模擬馬賽的起點(diǎn)，但隨后會(huì)設(shè)置一個(gè)門閂模擬馬賽的終點(diǎn)。這樣，“主” 線程就可以輸出結(jié)果。 為了讓馬賽有更多的輸出注釋，可以在賽場(chǎng)的 “轉(zhuǎn)彎處” 和 “半程” 點(diǎn)，比如賽馬跨過跑道的四分之一、二分之一和四分之三線時(shí)，添加?CountDownLatch。

回頁首

3. Executor

清單 1?和?清單 2?中的示例都存在一個(gè)重要的缺陷，它們要求您直接創(chuàng)建?Thread?對(duì)象。這可以解決一些問題，因?yàn)樵谝恍?JVM 中，創(chuàng)建Thread?是一項(xiàng)重量型的操作，重用現(xiàn)有?Thread?比創(chuàng)建新線程要容易得多。而在另一些 JVM 中，情況正好相反：Thread?是輕量型的，可以在需要時(shí)很容易地新建一個(gè)線程。當(dāng)然，如果 Murphy 擁有自己的解決辦法（他通常都會(huì)擁有），那么您無論使用哪種方法對(duì)于您最終將部署的平臺(tái)都是不對(duì)的。

JSR-166 專家組（參見?參考資料）在一定程度上預(yù)測(cè)到了這一情形。Java 開發(fā)人員無需直接創(chuàng)建?Thread，他們引入了?Executor?接口，這是對(duì)創(chuàng)建新線程的一種抽象。如清單 3 所示，Executor?使您不必親自對(duì)?Thread?對(duì)象執(zhí)行?new?就能夠創(chuàng)建新線程：

清單 3. Executor

Executor exec = getAnExecutorFromSomeplace(); exec.execute(new Runnable() { ... });

使用?Executor?的主要缺陷與我們?cè)谒泄S中遇到的一樣：工廠必須來自某個(gè)位置。不幸的是，與 CLR 不同，JVM 沒有附帶一個(gè)標(biāo)準(zhǔn)的 VM 級(jí)線程池。

Executor?類實(shí)際上?充當(dāng)著一個(gè)提供?Executor?實(shí)現(xiàn)實(shí)例的共同位置，但它只有?new?方法（例如用于創(chuàng)建新線程池）；它沒有預(yù)先創(chuàng)建實(shí)例。所以您可以自行決定是否希望在代碼中創(chuàng)建和使用?Executor?實(shí)例。（或者在某些情況下，您將能夠使用所選的容器/平臺(tái)提供的實(shí)例。）

ExecutorService 隨時(shí)可以使用

盡管不必?fù)?dān)心?Thread?來自何處，但?Executor?接口缺乏 Java 開發(fā)人員可能期望的某種功能，比如結(jié)束一個(gè)用于生成結(jié)果的線程并以非阻塞方式等待結(jié)果可用。（這是桌面應(yīng)用程序的一個(gè)常見需求，用戶將執(zhí)行需要訪問數(shù)據(jù)庫的 UI 操作，然后如果該操作花費(fèi)了很長時(shí)間，可能希望在它完成之前取消它。）

對(duì)于此問題，JSR-166 專家創(chuàng)建了一個(gè)更加有用的抽象（ExecutorService?接口），它將線程啟動(dòng)工廠建模為一個(gè)可集中控制的服務(wù)。例如，無需每執(zhí)行一項(xiàng)任務(wù)就調(diào)用一次?execute()，ExecutorService?可以接受一組任務(wù)并返回一個(gè)表示每項(xiàng)任務(wù)的未來結(jié)果的未來列表。

回頁首

4. ScheduledExecutorServices

盡管?ExecutorService?接口非常有用，但某些任務(wù)仍需要以計(jì)劃方式執(zhí)行，比如以確定的時(shí)間間隔或在特定時(shí)間執(zhí)行給定的任務(wù)。這就是ScheduledExecutorService?的應(yīng)用范圍，它擴(kuò)展了?ExecutorService。

如果您的目標(biāo)是創(chuàng)建一個(gè)每隔 5 秒跳一次的 “心跳” 命令，使用?ScheduledExecutorService?可以輕松實(shí)現(xiàn)，如清單 4 所示：

清單 4. ScheduledExecutorService 模擬心跳

import java.util.concurrent.*;public class Ping {public static void main(String[] args){ScheduledExecutorService ses =Executors.newScheduledThreadPool(1);Runnable pinger = new Runnable() {public void run() {System.out.println("PING!");}};ses.scheduleAtFixedRate(pinger, 5, 5, TimeUnit.SECONDS);} }

這項(xiàng)功能怎么樣？不用過于擔(dān)心線程，不用過于擔(dān)心用戶希望取消心跳時(shí)會(huì)發(fā)生什么，也不用明確地將線程標(biāo)記為前臺(tái)或后臺(tái)；只需將所有的計(jì)劃細(xì)節(jié)留給?ScheduledExecutorService。

順便說一下，如果用戶希望取消心跳，scheduleAtFixedRate?調(diào)用將返回一個(gè)?ScheduledFuture?實(shí)例，它不僅封裝了結(jié)果（如果有），還擁有一個(gè)?cancel?方法來關(guān)閉計(jì)劃的操作。

回頁首

5. Timeout 方法

為阻塞操作設(shè)置一個(gè)具體的超時(shí)值（以避免死鎖）的能力是?java.util.concurrent?庫相比起早期并發(fā)特性的一大進(jìn)步，比如監(jiān)控鎖定。

這些方法幾乎總是包含一個(gè)?int/TimeUnit?對(duì)，指示這些方法應(yīng)該等待多長時(shí)間才釋放控制權(quán)并將其返回給程序。它需要開發(fā)人員執(zhí)行更多工作 — 如果沒有獲取鎖，您將如何重新獲取？ — 但結(jié)果幾乎總是正確的：更少的死鎖和更加適合生產(chǎn)的代碼。（關(guān)于編寫生產(chǎn)就緒代碼的更多信息，請(qǐng)參見?參考資料?中 Michael Nygard 編寫的?Release It!。）

回頁首

結(jié)束語

java.util.concurrent?包還包含了其他許多好用的實(shí)用程序，它們很好地?cái)U(kuò)展到了 Collections 之外，尤其是在?.locks?和?.atomic?包中。深入研究，您還將發(fā)現(xiàn)一些有用的控制結(jié)構(gòu)，比如?CyclicBarrier?等。

與 Java 平臺(tái)的許多其他方面一樣，您無需費(fèi)勁地查找可能非常有用的基礎(chǔ)架構(gòu)代碼。在編寫多線程代碼時(shí)，請(qǐng)記住本文討論的實(shí)用程序和?上一篇文章?中討論的實(shí)用程序。

轉(zhuǎn)載于:https://www.cnblogs.com/davidwang456/p/3865117.html

總結(jié)

以上是生活随笔為你收集整理的关于 java.util.concurrent 您不知道的 5 件事--转的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。