并发编程 – Concurrent 用户指南
轉(zhuǎn)載自?并發(fā)編程 – Concurrent 用戶指南
1. java.util.concurrent – Java 并發(fā)工具包
Java 5 添加了一個(gè)新的包到 Java 平臺(tái),java.util.concurrent 包。這個(gè)包包含有一系列能夠讓 Java 的并發(fā)編程變得更加簡(jiǎn)單輕松的類。在這個(gè)包被添加以前,你需要自己去動(dòng)手實(shí)現(xiàn)自己的相關(guān)工具類。
本文我將帶你一一認(rèn)識(shí) java.util.concurrent 包里的這些類,然后你可以嘗試著如何在項(xiàng)目中使用它們。本文中我將使用 Java 6 版本,我不確定這和 Java 5 版本里的是否有一些差異。我不會(huì)去解釋關(guān)于 Java 并發(fā)的核心問(wèn)題 – 其背后的原理,也就是說(shuō),如果你對(duì)那些東西感興趣,參考《Java 并發(fā)指南》。
半成品
本文很大程度上還是個(gè) “半成品”,所以當(dāng)你發(fā)現(xiàn)一些被漏掉的類或接口時(shí),請(qǐng)耐心等待。在我空閑的時(shí)候會(huì)把它們加進(jìn)來(lái)的。
2. 阻塞隊(duì)列 BlockingQueue
java.util.concurrent 包里的 BlockingQueue 接口表示一個(gè)線程安放入和提取實(shí)例的隊(duì)列。本小節(jié)我將給你演示如何使用這個(gè) BlockingQueue。本節(jié)不會(huì)討論如何在 Java 中實(shí)現(xiàn)一個(gè)你自己的 BlockingQueue。如果你對(duì)那個(gè)感興趣,參考《Java 并發(fā)指南》
BlockingQueue 用法
BlockingQueue 通常用于一個(gè)線程生產(chǎn)對(duì)象,而另外一個(gè)線程消費(fèi)這些對(duì)象的場(chǎng)景。下圖是對(duì)這個(gè)原理的闡述:
一個(gè)線程往里邊放,另外一個(gè)線程從里邊取的一個(gè) BlockingQueue。
一個(gè)線程將會(huì)持續(xù)生產(chǎn)新對(duì)象并將其插入到隊(duì)列之中,直到隊(duì)列達(dá)到它所能容納的臨界點(diǎn)。也就是說(shuō),它是有限的。如果該阻塞隊(duì)列到達(dá)了其臨界點(diǎn),負(fù)責(zé)生產(chǎn)的線程將會(huì)在往里邊插入新對(duì)象時(shí)發(fā)生阻塞。它會(huì)一直處于阻塞之中,直到負(fù)責(zé)消費(fèi)的線程從隊(duì)列中拿走一個(gè)對(duì)象。負(fù)責(zé)消費(fèi)的線程將會(huì)一直從該阻塞隊(duì)列中拿出對(duì)象。如果消費(fèi)線程嘗試去從一個(gè)空的隊(duì)列中提取對(duì)象的話,這個(gè)消費(fèi)線程將會(huì)處于阻塞之中,直到一個(gè)生產(chǎn)線程把一個(gè)對(duì)象丟進(jìn)隊(duì)列。
BlockingQueue 的方法
BlockingQueue 具有 4 組不同的方法用于插入、移除以及對(duì)隊(duì)列中的元素進(jìn)行檢查。如果請(qǐng)求的操作不能得到立即執(zhí)行的話,每個(gè)方法的表現(xiàn)也不同。這些方法如下:
四組不同的行為方式解釋:
拋異常:如果試圖的操作無(wú)法立即執(zhí)行,拋一個(gè)異常。
特定值:如果試圖的操作無(wú)法立即執(zhí)行,返回一個(gè)特定的值(常常是 true / false)。
阻塞:如果試圖的操作無(wú)法立即執(zhí)行,該方法調(diào)用將會(huì)發(fā)生阻塞,直到能夠執(zhí)行。
超時(shí):如果試圖的操作無(wú)法立即執(zhí)行,該方法調(diào)用將會(huì)發(fā)生阻塞,直到能夠執(zhí)行,但等待時(shí)間不會(huì)超過(guò)給定值。返回一個(gè)特定值以告知該操作是否成功(典型的是 true / false)。
無(wú)法向一個(gè) BlockingQueue 中插入 null。如果你試圖插入 null,BlockingQueue 將會(huì)拋出一個(gè) NullPointerException。
可以訪問(wèn)到 BlockingQueue 中的所有元素,而不僅僅是開始和結(jié)束的元素。比如說(shuō),你將一個(gè)對(duì)象放入隊(duì)列之中以等待處理,但你的應(yīng)用想要將其取消掉。那么你可以調(diào)用諸如 remove(o) 方法來(lái)將隊(duì)列之中的特定對(duì)象進(jìn)行移除。但是這么干效率并不高(譯者注:基于隊(duì)列的數(shù)據(jù)結(jié)構(gòu),獲取除開始或結(jié)束位置的其他對(duì)象的效率不會(huì)太高),因此你盡量不要用這一類的方法,除非你確實(shí)不得不那么做。
BlockingQueue 的實(shí)現(xiàn)
BlockingQueue 是個(gè)接口,你需要使用它的實(shí)現(xiàn)之一來(lái)使用 BlockingQueue。java.util.concurrent 具有以下 BlockingQueue 接口的實(shí)現(xiàn)(Java 6):
- ArrayBlockingQueue
- DelayQueue
- LinkedBlockingQueue
- PriorityBlockingQueue
- SynchronousQueue
Java 中使用 BlockingQueue 的例子
這里是一個(gè) Java 中使用 BlockingQueue 的示例。本示例使用的是 BlockingQueue 接口的 ArrayBlockingQueue 實(shí)現(xiàn)。
首先,BlockingQueueExample 類分別在兩個(gè)獨(dú)立的線程中啟動(dòng)了一個(gè) Producer 和 一個(gè) Consumer。
Producer 向一個(gè)共享的 BlockingQueue 中注入字符串,而 Consumer 則會(huì)從中把它們拿出來(lái)。
| 123456789101112131415 | public class BlockingQueueExample {? ????public static void main(String[] args) throws Exception {? ????????BlockingQueue queue = new ArrayBlockingQueue(1024);? ????????Producer producer = new Producer(queue);? ????????Consumer consumer = new Consumer(queue);? ????????new Thread(producer).start();? ????????new Thread(consumer).start();? ????????Thread.sleep(4000);? ????}? } |
以下是 Producer 類。注意它在每次 put() 調(diào)用時(shí)是如何休眠一秒鐘的。這將導(dǎo)致 Consumer 在等待隊(duì)列中對(duì)象的時(shí)候發(fā)生阻塞。
| 1234567891011121314151617181920 | public class Producer implements Runnable{? ????protected BlockingQueue queue = null;? ????public Producer(BlockingQueue queue) {? ????????this.queue = queue;? ????}? ????public void run() {? ????????try {? ????????????queue.put("1");? ????????????Thread.sleep(1000);? ????????????queue.put("2");? ????????????Thread.sleep(1000);? ????????????queue.put("3");? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????}? ????}? } |
以下是 Consumer 類。它只是把對(duì)象從隊(duì)列中抽取出來(lái),然后將它們打印到 System.out。
| 123456789101112131415161718 | public class Consumer implements Runnable{? ????protected BlockingQueue queue = null;? ????public Consumer(BlockingQueue queue) {? ????????this.queue = queue;? ????}? ????public void run() {? ????????try {? ????????????System.out.println(queue.take());? ????????????System.out.println(queue.take());? ????????????System.out.println(queue.take());? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????}? ????}? } |
3. 數(shù)組阻塞隊(duì)列 ArrayBlockingQueue
ArrayBlockingQueue 類實(shí)現(xiàn)了 BlockingQueue 接口。
ArrayBlockingQueue 是一個(gè)有界的阻塞隊(duì)列,其內(nèi)部實(shí)現(xiàn)是將對(duì)象放到一個(gè)數(shù)組里。有界也就意味著,它不能夠存儲(chǔ)無(wú)限多數(shù)量的元素。它有一個(gè)同一時(shí)間能夠存儲(chǔ)元素?cái)?shù)量的上限。你可以在對(duì)其初始化的時(shí)候設(shè)定這個(gè)上限,但之后就無(wú)法對(duì)這個(gè)上限進(jìn)行修改了(譯者注:因?yàn)樗腔跀?shù)組實(shí)現(xiàn)的,也就具有數(shù)組的特性:一旦初始化,大小就無(wú)法修改)。
‘ArrayBlockingQueue 內(nèi)部以 FIFO(先進(jìn)先出)的順序?qū)υ剡M(jìn)行存儲(chǔ)。隊(duì)列中的頭元素在所有元素之中是放入時(shí)間最久的那個(gè),而尾元素則是最短的那個(gè)。
以下是在使用 ArrayBlockingQueue 的時(shí)候?qū)ζ涑跏蓟囊粋€(gè)示例:
| 12345 | BlockingQueue queue = new ArrayBlockingQueue(1024);? queue.put("1");? Object object = queue.take(); |
以下是使用了 Java 泛型的一個(gè) BlockingQueue 示例。注意其中是如何對(duì) String 元素放入和提取的:
| 12345 | BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024);? queue.put("1");? String string = queue.take(); |
4. 延遲隊(duì)列 DelayQueue
DelayQueue 實(shí)現(xiàn)了 BlockingQueue 接口。DelayQueue 對(duì)元素進(jìn)行持有直到一個(gè)特定的延遲到期。注入其中的元素必須實(shí)現(xiàn) java.util.concurrent.Delayed 接口,該接口定義:
| 12345 | public interface Delayed extends Comparable<Delayed< {? ?public long getDelay(TimeUnit timeUnit);? } |
DelayQueue 將會(huì)在每個(gè)元素的 getDelay() 方法返回的值的時(shí)間段之后才釋放掉該元素。如果返回的是 0 或者負(fù)值,延遲將被認(rèn)為過(guò)期,該元素將會(huì)在 DelayQueue 的下一次 take 被調(diào)用的時(shí)候被釋放掉。傳遞給 getDelay 方法的 getDelay 實(shí)例是一個(gè)枚舉類型,它表明了將要延遲的時(shí)間段。
TimeUnit 枚舉將會(huì)取以下值:
| 12345678 | DAYS? HOURS? MINUTES? SECONDS? MILLISECONDS? MICROSECONDS? NANOSECONDS? ` |
正如你所看到的,Delayed 接口也繼承了 java.lang.Comparable 接口,這也就意味著 Delayed 對(duì)象之間可以進(jìn)行對(duì)比。這個(gè)可能在對(duì) DelayQueue 隊(duì)列中的元素進(jìn)行排序時(shí)有用,因此它們可以根據(jù)過(guò)期時(shí)間進(jìn)行有序釋放。以下是使用 DelayQueue 的例子:
| 123456789101112 | public class DelayQueueExample {? ????public static void main(String[] args) {? ????????DelayQueue queue = new DelayQueue();? ????????Delayed element1 = new DelayedElement();? ????????queue.put(element1);? ????????Delayed element2 = queue.take();? ????}? } |
DelayedElement 是我所創(chuàng)建的一個(gè) DelayedElement 接口的實(shí)現(xiàn)類,它不在 Java.util.concurrent 包里。你需要自行創(chuàng)建你自己的 Delayed 接口的實(shí)現(xiàn)以使用 DelayQueue 類。
5. 鏈阻塞隊(duì)列 LinkedBlockingQueue
LinkedBlockingQueue 類實(shí)現(xiàn)了 BlockingQueue 接口。
LinkedBlockingQueue 內(nèi)部以一個(gè)鏈?zhǔn)浇Y(jié)構(gòu)(鏈接節(jié)點(diǎn))對(duì)其元素進(jìn)行存儲(chǔ)。如果需要的話,這一鏈?zhǔn)浇Y(jié)構(gòu)可以選擇一個(gè)上限。如果沒(méi)有定義上限,將使用 Integer.MAX_VALUE 作為上限。
LinkedBlockingQueue 內(nèi)部以 FIFO(先進(jìn)先出)的順序?qū)υ剡M(jìn)行存儲(chǔ)。隊(duì)列中的頭元素在所有元素之中是放入時(shí)間最久的那個(gè),而尾元素則是最短的那個(gè)。
以下是 LinkedBlockingQueue 的初始化和使用示例代碼:
| 123456 | BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();? BlockingQueue<String> bounded?? = new LinkedBlockingQueue<String>(1024);? bounded.put("Value");? String value = bounded.take(); |
6. 具有優(yōu)先級(jí)的阻塞隊(duì)列 PriorityBlockingQueue
PriorityBlockingQueue 類實(shí)現(xiàn)了 BlockingQueue 接口。
PriorityBlockingQueue 是一個(gè)無(wú)界的并發(fā)隊(duì)列。它使用了和類 java.util.PriorityQueue 一樣的排序規(guī)則。你無(wú)法向這個(gè)隊(duì)列中插入 null 值。所有插入到 PriorityBlockingQueue 的元素必須實(shí)現(xiàn) java.lang.Comparable 接口。因此該隊(duì)列中元素的排序就取決于你自己的 Comparable 實(shí)現(xiàn)。注意 PriorityBlockingQueue 對(duì)于具有相等優(yōu)先級(jí)(compare() == 0)的元素并不強(qiáng)制任何特定行為。
同時(shí)注意,如果你從一個(gè) PriorityBlockingQueue 獲得一個(gè) Iterator 的話,該 Iterator 并不能保證它對(duì)元素的遍歷是以優(yōu)先級(jí)為序的。
以下是使用 PriorityBlockingQueue 的示例:
| 123456 | BlockingQueue queue?? = new PriorityBlockingQueue();? ????//String implements java.lang.Comparable? ????queue.put("Value");? ????String value = queue.take(); |
7. 同步隊(duì)列 SynchronousQueue
SynchronousQueue 類實(shí)現(xiàn)了 BlockingQueue 接口。
SynchronousQueue 是一個(gè)特殊的隊(duì)列,它的內(nèi)部同時(shí)只能夠容納單個(gè)元素。如果該隊(duì)列已有一元素的話,試圖向隊(duì)列中插入一個(gè)新元素的線程將會(huì)阻塞,直到另一個(gè)線程將該元素從隊(duì)列中抽走。同樣,如果該隊(duì)列為空,試圖向隊(duì)列中抽取一個(gè)元素的線程將會(huì)阻塞,直到另一個(gè)線程向隊(duì)列中插入了一條新的元素。
據(jù)此,把這個(gè)類稱作一個(gè)隊(duì)列顯然是夸大其詞了。它更多像是一個(gè)匯合點(diǎn)。
8. 阻塞雙端隊(duì)列 BlockingDeque
java.util.concurrent 包里的 BlockingDeque 接口表示一個(gè)線程安放入和提取實(shí)例的雙端隊(duì)列。本小節(jié)我將給你演示如何使用 BlockingDeque。BlockingDeque 類是一個(gè)雙端隊(duì)列,在不能夠插入元素時(shí),它將阻塞住試圖插入元素的線程;在不能夠抽取元素時(shí),它將阻塞住試圖抽取的線程。deque(雙端隊(duì)列) 是 “Double Ended Queue” 的縮寫。因此,雙端隊(duì)列是一個(gè)你可以從任意一端插入或者抽取元素的隊(duì)列。
BlockingDeque 的使用
在線程既是一個(gè)隊(duì)列的生產(chǎn)者又是這個(gè)隊(duì)列的消費(fèi)者的時(shí)候可以使用到 BlockingDeque。如果生產(chǎn)者線程需要在隊(duì)列的兩端都可以插入數(shù)據(jù),消費(fèi)者線程需要在隊(duì)列的兩端都可以移除數(shù)據(jù),這個(gè)時(shí)候也可以使用 BlockingDeque。
一個(gè) BlockingDeque – 線程在雙端隊(duì)列的兩端都可以插入和提取元素。
一個(gè)線程生產(chǎn)元素,并把它們插入到隊(duì)列的任意一端。如果雙端隊(duì)列已滿,插入線程將被阻塞,直到一個(gè)移除線程從該隊(duì)列中移出了一個(gè)元素。如果雙端隊(duì)列為空,移除線程將被阻塞,直到一個(gè)插入線程向該隊(duì)列插入了一個(gè)新元素。
BlockingDeque 的方法
BlockingDeque 具有 4 組不同的方法用于插入、移除以及對(duì)雙端隊(duì)列中的元素進(jìn)行檢查。如果請(qǐng)求的操作不能得到立即執(zhí)行的話,每個(gè)方法的表現(xiàn)也不同。這些方法如下:
四組不同的行為方式解釋:
拋異常:如果試圖的操作無(wú)法立即執(zhí)行,拋一個(gè)異常。
特定值:如果試圖的操作無(wú)法立即執(zhí)行,返回一個(gè)特定的值(常常是 true / false)。
阻塞:如果試圖的操作無(wú)法立即執(zhí)行,該方法調(diào)用將會(huì)發(fā)生阻塞,直到能夠執(zhí)行。
超時(shí):如果試圖的操作無(wú)法立即執(zhí)行,該方法調(diào)用將會(huì)發(fā)生阻塞,直到能夠執(zhí)行,但等待時(shí)間不會(huì)超過(guò)給定值。返回一個(gè)特定值以告知該操作是否成功(典型的是 true / false)。
BlockingDeque 繼承自 BlockingQueue
BlockingDeque 接口繼承自 BlockingQueue 接口。
這就意味著你可以像使用一個(gè) BlockingQueue 那樣使用 BlockingDeque。如果你這么干的話,各種插入方法將會(huì)把新元素添加到雙端隊(duì)列的尾端,而移除方法將會(huì)把雙端隊(duì)列的首端的元素移除。正如 BlockingQueue 接口的插入和移除方法一樣。
以下是 BlockingDeque 對(duì) BlockingQueue 接口的方法的具體內(nèi)部實(shí)現(xiàn):
BlockingDeque 的實(shí)現(xiàn)
既然 BlockingDeque 是一個(gè)接口,那么你想要使用它的話就得使用它的眾多的實(shí)現(xiàn)類的其中一個(gè)。java.util.concurrent 包提供了以下 BlockingDeque 接口的實(shí)現(xiàn)類:
LinkedBlockingDeque
BlockingDeque 代碼示例
以下是如何使用 BlockingDeque 方法的一個(gè)簡(jiǎn)短代碼示例:
| 1234567 | BlockingDeque<String> deque = new LinkedBlockingDeque<String>();? deque.addFirst("1");? deque.addLast("2");? String two = deque.takeLast();? String one = deque.takeFirst(); |
9. 鏈阻塞雙端隊(duì)列 LinkedBlockingDeque
LinkedBlockingDeque 類實(shí)現(xiàn)了 BlockingDeque 接口。
deque(雙端隊(duì)列) 是 “Double Ended Queue” 的縮寫。因此,雙端隊(duì)列是一個(gè)你可以從任意一端插入或者抽取元素的隊(duì)列。(譯者注:唐僧啊,受不了。)LinkedBlockingDeque 是一個(gè)雙端隊(duì)列,在它為空的時(shí)候,一個(gè)試圖從中抽取數(shù)據(jù)的線程將會(huì)阻塞,無(wú)論該線程是試圖從哪一端抽取數(shù)據(jù)。以下是 LinkedBlockingDeque 實(shí)例化以及使用的示例:
| 1234567 | BlockingDeque<String> deque = new LinkedBlockingDeque<String>();? deque.addFirst("1");? deque.addLast("2");? String two = deque.takeLast();? String one = deque.takeFirst(); |
10. 并發(fā) Map(映射) ConcurrentMap
java.util.concurrent.ConcurrentMap
java.util.concurrent.ConcurrentMap 接口表示了一個(gè)能夠?qū)e人的訪問(wèn)(插入和提取)進(jìn)行并發(fā)處理的 java.util.Map。ConcurrentMap 除了從其父接口 java.util.Map 繼承來(lái)的方法之外還有一些額外的原子性方法。
ConcurrentMap 的實(shí)現(xiàn)
既然 ConcurrentMap 是個(gè)接口,你想要使用它的話就得使用它的實(shí)現(xiàn)類之一。java.util.concurrent 包具備 ConcurrentMap 接口的以下實(shí)現(xiàn)類:
- ConcurrentHashMap
ConcurrentHashMap
ConcurrentHashMap 和 java.util.HashTable 類很相似,但 ConcurrentHashMap 能夠提供比 HashTable 更好的并發(fā)性能。在你從中讀取對(duì)象的時(shí)候 ConcurrentHashMap 并不會(huì)把整個(gè) Map 鎖住。
此外,在你向其中寫入對(duì)象的時(shí)候,ConcurrentHashMap 也不會(huì)鎖住整個(gè) Map。它的內(nèi)部只是把 Map 中正在被寫入的部分進(jìn)行鎖定。
另外一個(gè)不同點(diǎn)是,在被遍歷的時(shí)候,即使是 ConcurrentHashMap 被改動(dòng),它也不會(huì)拋 ConcurrentModificationException。盡管 Iterator 的設(shè)計(jì)不是為多個(gè)線程的同時(shí)使用。更多關(guān)于 ConcurrentMap 和 ConcurrentHashMap 的細(xì)節(jié)請(qǐng)參考官方文檔。
ConcurrentMap 例子
以下是如何使用 ConcurrentMap 接口的一個(gè)例子。
本示例使用了 ConcurrentHashMap 實(shí)現(xiàn)類:
| 12345 | ConcurrentMap concurrentMap = new ConcurrentHashMap();? concurrentMap.put("key", "value");? Object value = concurrentMap.get("key"); |
11. 并發(fā)導(dǎo)航映射 ConcurrentNavigableMap
java.util.concurrent.ConcurrentNavigableMap 是一個(gè)支持并發(fā)訪問(wèn)的 java.util.NavigableMap,它還能讓它的子 map 具備并發(fā)訪問(wèn)的能力。所謂的 “子 map” 指的是諸如 headMap(),subMap(),tailMap() 之類的方法返回的 map。
NavigableMap 中的方法不再贅述,本小節(jié)我們來(lái)看一下 ConcurrentNavigableMap 添加的方法。
headMap()
headMap(T toKey) 方法返回一個(gè)包含了小于給定 toKey 的 key 的子 map。如果你對(duì)原始 map 里的元素做了改動(dòng),這些改動(dòng)將影響到子 map 中的元素(譯者注:map 集合持有的其實(shí)只是對(duì)象的引用)。以下示例演示了對(duì) headMap() 方法的使用:
| 1234567 | ConcurrentNavigableMap map = new ConcurrentSkipListMap();? map.put("1", "one");? map.put("2", "two");? map.put("3", "three");? ConcurrentNavigableMap headMap = map.headMap("2"); |
headMap 將指向一個(gè)只含有鍵 “1″ 的 ConcurrentNavigableMap,因?yàn)橹挥羞@一個(gè)鍵小于 “2″。關(guān)于這個(gè)方法及其重載版本具體是怎么工作的細(xì)節(jié)請(qǐng)參考 Java 文檔。
tailMap()
tailMap(T fromKey) 方法返回一個(gè)包含了不小于給定 fromKey 的 key 的子 map。
如果你對(duì)原始 map 里的元素做了改動(dòng),這些改動(dòng)將影響到子 map 中的元素(譯者注:map 集合持有的其實(shí)只是對(duì)象的引用)。
以下示例演示了對(duì) tailMap() 方法的使用:
| 1234567 | ConcurrentNavigableMap map = new ConcurrentSkipListMap();? map.put("1", "one");? map.put("2", "two");? map.put("3", "three");? ConcurrentNavigableMap tailMap = map.tailMap("2"); |
tailMap 將擁有鍵 “2″ 和 “3″,因?yàn)樗鼈儾恍∮诮o定鍵 “2″。關(guān)于這個(gè)方法及其重載版本具體是怎么工作的細(xì)節(jié)請(qǐng)參考 Java 文檔。
subMap()
subMap() 方法返回原始 map 中,鍵介于 from(包含) 和 to (不包含) 之間的子 map。
示例如下:
| 1234567 | ConcurrentNavigableMap map = new ConcurrentSkipListMap();? map.put("1", "one");? map.put("2", "two");? map.put("3", "three");? ConcurrentNavigableMap subMap = map.subMap("2", "3"); |
返回的 submap 只包含鍵 “2″,因?yàn)橹挥兴鼭M足不小于 “2″,比 “3″ 小。
更多方法
ConcurrentNavigableMap 接口還有其他一些方法可供使用,
比如:
- descendingKeySet()
- descendingMap()
- navigableKeySet()
關(guān)于這些方法更多信息參考官方 Java 文檔。
12. 閉鎖 CountDownLatch
java.util.concurrent.CountDownLatch 是一個(gè)并發(fā)構(gòu)造,它允許一個(gè)或多個(gè)線程等待一系列指定操作的完成。
CountDownLatch 以一個(gè)給定的數(shù)量初始化。countDown() 每被調(diào)用一次,這一數(shù)量就減一。通過(guò)調(diào)用 await() 方法之一,線程可以阻塞等待這一數(shù)量到達(dá)零。以下是一個(gè)簡(jiǎn)單示例。
Decrementer 三次調(diào)用 countDown() 之后,等待中的 Waiter 才會(huì)從 await() 調(diào)用中釋放出來(lái)。
| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 | CountDownLatch latch = new CountDownLatch(3);? Waiter????? waiter????? = new Waiter(latch);? Decrementer decrementer = new Decrementer(latch);? new Thread(waiter)???? .start();? new Thread(decrementer).start();? Thread.sleep(4000);? public class Waiter implements Runnable{? ????CountDownLatch latch = null;? ????public Waiter(CountDownLatch latch) {? ????????this.latch = latch;? ????}? ????public void run() {? ????????try {? ????????????latch.await();? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????}? ????????System.out.println("Waiter Released");? ????}? }? public class Decrementer implements Runnable {? ????CountDownLatch latch = null;? ????public Decrementer(CountDownLatch latch) {? ????????this.latch = latch;? ????}? ????public void run() {? ????????try {? ????????????Thread.sleep(1000);? ????????????this.latch.countDown();? ????????????Thread.sleep(1000);? ????????????this.latch.countDown();? ????????????Thread.sleep(1000);? ????????????this.latch.countDown();? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????}? ????}? } |
13. 柵欄 CyclicBarrier
java.util.concurrent.CyclicBarrier 類是一種同步機(jī)制,它能夠?qū)μ幚硪恍┧惴ǖ木€程實(shí)現(xiàn)同步。換句話講,它就是一個(gè)所有線程必須等待的一個(gè)柵欄,直到所有線程都到達(dá)這里,然后所有線程才可以繼續(xù)做其他事情。
圖示如下:
兩個(gè)線程在柵欄旁等待對(duì)方。
通過(guò)調(diào)用 CyclicBarrier 對(duì)象的 await() 方法,兩個(gè)線程可以實(shí)現(xiàn)互相等待。一旦 N 個(gè)線程在等待 CyclicBarrier 達(dá)成,所有線程將被釋放掉去繼續(xù)運(yùn)行。
創(chuàng)建一個(gè) CyclicBarrier
在創(chuàng)建一個(gè) CyclicBarrier 的時(shí)候你需要定義有多少線程在被釋放之前等待柵欄。
創(chuàng)建 CyclicBarrier 示例:
| 1 | CyclicBarrier barrier = new CyclicBarrier(2); |
等待一個(gè) CyclicBarrier
以下演示了如何讓一個(gè)線程等待一個(gè) CyclicBarrier:
| 1 | barrier.await(); |
當(dāng)然,你也可以為等待線程設(shè)定一個(gè)超時(shí)時(shí)間。等待超過(guò)了超時(shí)時(shí)間之后,即便還沒(méi)有達(dá)成 N 個(gè)線程等待 CyclicBarrier 的條件,該線程也會(huì)被釋放出來(lái)。以下是定義超時(shí)時(shí)間示例:
| 1 | barrier.await(10, TimeUnit.SECONDS); |
滿足以下任何條件都可以讓等待 CyclicBarrier 的線程釋放:
- 最后一個(gè)線程也到達(dá) CyclicBarrier(調(diào)用 await())
- 當(dāng)前線程被其他線程打斷(其他線程調(diào)用了這個(gè)線程的 interrupt() 方法)
- 其他等待柵欄的線程被打斷
- 其他等待柵欄的線程因超時(shí)而被釋放
- 外部線程調(diào)用了柵欄的 CyclicBarrier.reset() 方法
CyclicBarrier 行動(dòng)
CyclicBarrier 支持一個(gè)柵欄行動(dòng),柵欄行動(dòng)是一個(gè) Runnable 實(shí)例,一旦最后等待柵欄的線程抵達(dá),該實(shí)例將被執(zhí)行。你可以在 CyclicBarrier 的構(gòu)造方法中將 Runnable 柵欄行動(dòng)傳給它:
| 12 | Runnable barrierAction = ... ; CyclicBarrier barrier = new CyclicBarrier(2, barrierAction); |
CyclicBarrier 示例
以下代碼演示了如何使用 CyclicBarrier:
| 12345678910111213141516171819202122 | Runnable barrier1Action = new Runnable() {? ????public void run() {? ????????System.out.println("BarrierAction 1 executed ");? ????}? };? Runnable barrier2Action = new Runnable() {? ????public void run() {? ????????System.out.println("BarrierAction 2 executed ");? ????}? };? CyclicBarrier barrier1 = new CyclicBarrier(2, barrier1Action);? CyclicBarrier barrier2 = new CyclicBarrier(2, barrier2Action);? CyclicBarrierRunnable barrierRunnable1 =? ????????new CyclicBarrierRunnable(barrier1, barrier2);? CyclicBarrierRunnable barrierRunnable2 =? ????????new CyclicBarrierRunnable(barrier1, barrier2);? new Thread(barrierRunnable1).start();? new Thread(barrierRunnable2).start(); |
CyclicBarrierRunnable 類:
| 1234567891011121314151617181920212223242526272829303132333435 | public class CyclicBarrierRunnable implements Runnable{? ????CyclicBarrier barrier1 = null;? ????CyclicBarrier barrier2 = null;? ????public CyclicBarrierRunnable(? ????????????CyclicBarrier barrier1,? ????????????CyclicBarrier barrier2) {? ????????this.barrier1 = barrier1;? ????????this.barrier2 = barrier2;? ????}? ????public void run() {? ????????try {? ????????????Thread.sleep(1000);? ????????????System.out.println(Thread.currentThread().getName() +? ????????????????????????????????" waiting at barrier 1");? ????????????this.barrier1.await();? ????????????Thread.sleep(1000);? ????????????System.out.println(Thread.currentThread().getName() +? ????????????????????????????????" waiting at barrier 2");? ????????????this.barrier2.await();? ????????????System.out.println(Thread.currentThread().getName() +? ????????????????????????????????" done!");? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????} catch (BrokenBarrierException e) {? ????????????e.printStackTrace();? ????????}? ????}? } |
以上代碼控制臺(tái)輸出如下。注意每個(gè)線程寫入控制臺(tái)的時(shí)序可能會(huì)跟你實(shí)際執(zhí)行不一樣。比如有時(shí) Thread-0 先打印,有時(shí) Thread-1 先打印。
| 12345678 | Thread-0 waiting at barrier 1Thread-1 waiting at barrier 1BarrierAction 1 executedThread-1 waiting at barrier 2Thread-0 waiting at barrier 2BarrierAction 2 executedThread-0 done!Thread-1 done! |
14. 交換機(jī) Exchanger
java.util.concurrent.Exchanger 類表示一種兩個(gè)線程可以進(jìn)行互相交換對(duì)象的會(huì)和點(diǎn)。這種機(jī)制圖示如下:
兩個(gè)線程通過(guò)一個(gè) Exchanger 交換對(duì)象。
交換對(duì)象的動(dòng)作由 Exchanger 的兩個(gè) exchange() 方法的其中一個(gè)完成。
以下是一個(gè)示例:
| 12345678910 | Exchanger exchanger = new Exchanger();? ExchangerRunnable exchangerRunnable1 =? ????????new ExchangerRunnable(exchanger, "A");? ExchangerRunnable exchangerRunnable2 =? ????????new ExchangerRunnable(exchanger, "B");? new Thread(exchangerRunnable1).start();? new Thread(exchangerRunnable2).start(); |
ExchangerRunnable 代碼:
| 12345678910111213141516171819202122232425 | public class ExchangerRunnable implements Runnable{? ????Exchanger exchanger = null;? ????Object??? object??? = null;? ????public ExchangerRunnable(Exchanger exchanger, Object object) {? ????????this.exchanger = exchanger;? ????????this.object = object;? ????}? ????public void run() {? ????????try {? ????????????Object previous = this.object;? ????????????this.object = this.exchanger.exchange(this.object);? ????????????System.out.println(? ????????????????????Thread.currentThread().getName() +? ????????????????????" exchanged " + previous + " for " + this.object? ????????????);? ????????} catch (InterruptedException e) {? ????????????e.printStackTrace();? ????????}? ????}? } |
以上程序輸出:Thread-0 exchanged A for BThread-1 exchanged B for A
15. 信號(hào)量 Semaphore
java.util.concurrent.Semaphore 類是一個(gè)計(jì)數(shù)信號(hào)量。這就意味著它具備兩個(gè)主要方法:
- acquire()
- release()
計(jì)數(shù)信號(hào)量由一個(gè)指定數(shù)量的 “許可” 初始化。每調(diào)用一次 acquire(),一個(gè)許可會(huì)被調(diào)用線程取走。每調(diào)用一次 release(),一個(gè)許可會(huì)被返還給信號(hào)量。因此,在沒(méi)有任何 release() 調(diào)用時(shí),最多有 N 個(gè)線程能夠通過(guò) acquire() 方法,N 是該信號(hào)量初始化時(shí)的許可的指定數(shù)量。這些許可只是一個(gè)簡(jiǎn)單的計(jì)數(shù)器。這里沒(méi)啥奇特的地方。
Semaphore 用法
信號(hào)量主要有兩種用途:
保護(hù)重要部分
如果你將信號(hào)量用于保護(hù)一個(gè)重要部分,試圖進(jìn)入這一部分的代碼通常會(huì)首先嘗試獲得一個(gè)許可,然后才能進(jìn)入重要部分(代碼塊),執(zhí)行完之后,再把許可釋放掉。
比如這樣:
| 12345678 | Semaphore semaphore = new Semaphore(1);? //critical section? semaphore.acquire();? ...? semaphore.release(); |
在線程之間發(fā)送信號(hào)
如果你將一個(gè)信號(hào)量用于在兩個(gè)線程之間傳送信號(hào),通常你應(yīng)該用一個(gè)線程調(diào)用 acquire() 方法,而另一個(gè)線程調(diào)用 release() 方法。如果沒(méi)有可用的許可,acquire() 調(diào)用將會(huì)阻塞,直到一個(gè)許可被另一個(gè)線程釋放出來(lái)。同理,如果無(wú)法往信號(hào)量釋放更多許可時(shí),一個(gè) release() 調(diào)用也會(huì)阻塞。
通過(guò)這個(gè)可以對(duì)多個(gè)線程進(jìn)行協(xié)調(diào)。比如,如果線程 1 將一個(gè)對(duì)象插入到了一個(gè)共享列表(list)之后之后調(diào)用了 acquire(),而線程 2 則在從該列表中獲取一個(gè)對(duì)象之前調(diào)用了 release(),這時(shí)你其實(shí)已經(jīng)創(chuàng)建了一個(gè)阻塞隊(duì)列。信號(hào)量中可用的許可的數(shù)量也就等同于該阻塞隊(duì)列能夠持有的元素個(gè)數(shù)。
公平
沒(méi)有辦法保證線程能夠公平地可從信號(hào)量中獲得許可。也就是說(shuō),無(wú)法擔(dān)保掉第一個(gè)調(diào)用 acquire() 的線程會(huì)是第一個(gè)獲得一個(gè)許可的線程。如果第一個(gè)線程在等待一個(gè)許可時(shí)發(fā)生阻塞,而第二個(gè)線程前來(lái)索要一個(gè)許可的時(shí)候剛好有一個(gè)許可被釋放出來(lái),那么它就可能會(huì)在第一個(gè)線程之前獲得許可。如果你想要強(qiáng)制公平,Semaphore 類有一個(gè)具有一個(gè)布爾類型的參數(shù)的構(gòu)造子,通過(guò)這個(gè)參數(shù)以告知 Semaphore 是否要強(qiáng)制公平。強(qiáng)制公平會(huì)影響到并發(fā)性能,所以除非你確實(shí)需要它否則不要啟用它。
以下是如何在公平模式創(chuàng)建一個(gè) Semaphore 的示例:
| 1 | Semaphore semaphore = new Semaphore(1, true); |
更多方法
java.util.concurrent.Semaphore 類還有很多方法,比如:
- availablePermits()
- acquireUninterruptibly()
- drainPermits()
- hasQueuedThreads()
- getQueuedThreads()
- tryAcquire()
- 等等
這些方法的細(xì)節(jié)請(qǐng)參考 Java 文檔。
16. 執(zhí)行器服務(wù) ExecutorService
java.util.concurrent.ExecutorService 接口表示一個(gè)異步執(zhí)行機(jī)制,使我們能夠在后臺(tái)執(zhí)行任務(wù)。因此一個(gè) ExecutorService 很類似于一個(gè)線程池。實(shí)際上,存在于 java.util.concurrent 包里的 ExecutorService 實(shí)現(xiàn)就是一個(gè)線程池實(shí)現(xiàn)。
ExecutorService 例子
以下是一個(gè)簡(jiǎn)單的 ExecutorService 例子:
| 123456789 | ExecutorService executorService = Executors.newFixedThreadPool(10);? executorService.execute(new Runnable() {? ????public void run() {? ????????System.out.println("Asynchronous task");? ????}? });? executorService.shutdown(); |
首先使用 newFixedThreadPool() 工廠方法創(chuàng)建一個(gè) ExecutorService。這里創(chuàng)建了一個(gè)十個(gè)線程執(zhí)行任務(wù)的線程池。然后,將一個(gè) Runnable 接口的匿名實(shí)現(xiàn)類傳遞給 execute() 方法。這將導(dǎo)致 ExecutorService 中的某個(gè)線程執(zhí)行該 Runnable。
任務(wù)委派
下圖說(shuō)明了一個(gè)線程是如何將一個(gè)任務(wù)委托給一個(gè) ExecutorService 去異步執(zhí)行的:
一個(gè)線程將一個(gè)任務(wù)委派給一個(gè) ExecutorService 去異步執(zhí)行。
一旦該線程將任務(wù)委派給 ExecutorService,該線程將繼續(xù)它自己的執(zhí)行,獨(dú)立于該任務(wù)的執(zhí)行。
ExecutorService 實(shí)現(xiàn)
既然 ExecutorService 是個(gè)接口,如果你想用它的話就得去使用它的實(shí)現(xiàn)類之一。
java.util.concurrent 包提供了 ExecutorService 接口的以下實(shí)現(xiàn)類:
- ThreadPoolExecutor
- ScheduledThreadPoolExecutor
創(chuàng)建一個(gè) ExecutorService
ExecutorService 的創(chuàng)建依賴于你使用的具體實(shí)現(xiàn)。但是你也可以使用 Executors 工廠類來(lái)創(chuàng)建 ExecutorService 實(shí)例。
以下是幾個(gè)創(chuàng)建 ExecutorService 實(shí)例的例子:
| 12345 | ExecutorService executorService1 = Executors.newSingleThreadExecutor();? ExecutorService executorService2 = Executors.newFixedThreadPool(10);? ExecutorService executorService3 = Executors.newScheduledThreadPool(10); |
ExecutorService 使用
有幾種不同的方式來(lái)將任務(wù)委托給 ExecutorService 去執(zhí)行:
- execute(Runnable)
- submit(Runnable)
- submit(Callable)
- invokeAny(…)
- invokeAll(…)
接下來(lái)我們挨個(gè)看一下這些方法。
execute(Runnable)
execute(Runnable) 方法要求一個(gè) java.lang.Runnable 對(duì)象,然后對(duì)它進(jìn)行異步執(zhí)行。以下是使用 ExecutorService 執(zhí)行一個(gè) Runnable 的示例:
| 123456789 | ExecutorService executorService = Executors.newSingleThreadExecutor();? executorService.execute(new Runnable() {? ????public void run() {? ????????System.out.println("Asynchronous task");? ????}? });? executorService.shutdown(); |
沒(méi)有辦法得知被執(zhí)行的 Runnable 的執(zhí)行結(jié)果。如果有需要的話你得使用一個(gè) Callable(以下將做介紹)。
submit(Runnable)
submit(Runnable) 方法也要求一個(gè) Runnable 實(shí)現(xiàn)類,但它返回一個(gè) Future 對(duì)象。這個(gè) Future 對(duì)象可以用來(lái)檢查 Runnable 是否已經(jīng)執(zhí)行完畢。以下是 ExecutorService submit() 示例:
| 1234567 | Future future = executorService.submit(new Runnable() {? ????public void run() {? ????????System.out.println("Asynchronous task");? ????}? });? future.get();? //returns null if the task has finished correctly. |
submit(Callable)
submit(Callable) 方法類似于 submit(Runnable) 方法,除了它所要求的參數(shù)類型之外。Callable 實(shí)例除了它的 call() 方法能夠返回一個(gè)結(jié)果之外和一個(gè) Runnable 很相像。Runnable.run() 不能夠返回一個(gè)結(jié)果。Callable 的結(jié)果可以通過(guò) submit(Callable) 方法返回的 Future 對(duì)象進(jìn)行獲取。
以下是一個(gè) ExecutorService Callable 示例:
| 12345678 | Future future = executorService.submit(new Callable(){? ????public Object call() throws Exception {? ????????System.out.println("Asynchronous Callable");? ????????return "Callable Result";? ????}? });? System.out.println("future.get() = " + future.get()); |
以上代碼輸出:
| 12 | Asynchronous Callablefuture.get() = Callable Result |
invokeAny()
invokeAny() 方法要求一系列的 Callable 或者其子接口的實(shí)例對(duì)象。調(diào)用這個(gè)方法并不會(huì)返回一個(gè) Future,但它返回其中一個(gè) Callable 對(duì)象的結(jié)果。無(wú)法保證返回的是哪個(gè) Callable 的結(jié)果 – 只能表明其中一個(gè)已執(zhí)行結(jié)束。
如果其中一個(gè)任務(wù)執(zhí)行結(jié)束(或者拋了一個(gè)異常),其他 Callable 將被取消。以下是示例代碼:
| 12345678910111213141516171819202122232425 | ExecutorService executorService = Executors.newSingleThreadExecutor();? Set<Callable<String>> callables = new HashSet<Callable<String>>();? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 1";? ????}? });? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 2";? ????}? });? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 3";? ????}? });? String result = executorService.invokeAny(callables);? System.out.println("result = " + result);? executorService.shutdown(); |
上述代碼將會(huì)打印出給定 Callable 集合中的一個(gè)的執(zhí)行結(jié)果。我自己試著執(zhí)行了它幾次,結(jié)果始終在變。有時(shí)是 “Task 1″,有時(shí)是 “Task 2″ 等等。
invokeAll()
invokeAll() 方法將調(diào)用你在集合中傳給 ExecutorService 的所有 Callable 對(duì)象。invokeAll() 返回一系列的 Future 對(duì)象,通過(guò)它們你可以獲取每個(gè) Callable 的執(zhí)行結(jié)果。記住,一個(gè)任務(wù)可能會(huì)由于一個(gè)異常而結(jié)束,因此它可能沒(méi)有 “成功”。
無(wú)法通過(guò)一個(gè) Future 對(duì)象來(lái)告知我們是兩種結(jié)束中的哪一種。以下是一個(gè)代碼示例:
| 123456789101112131415161718192021222324252627 | ExecutorService executorService = Executors.newSingleThreadExecutor();? Set<Callable<String>> callables = new HashSet<Callable<String>>();? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 1";? ????}? });? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 2";? ????}? });? callables.add(new Callable<String>() {? ????public String call() throws Exception {? ????????return "Task 3";? ????}? });? List<Future<String>> futures = executorService.invokeAll(callables);? for(Future<String> future : futures){? ????System.out.println("future.get = " + future.get());? }? executorService.shutdown(); |
ExecutorService 關(guān)閉
使用完 ExecutorService 之后你應(yīng)該將其關(guān)閉,以使其中的線程不再運(yùn)行。
比如,如果你的應(yīng)用是通過(guò)一個(gè) main() 方法啟動(dòng)的,之后 main 方法退出了你的應(yīng)用,如果你的應(yīng)用有一個(gè)活動(dòng)的 ExexutorService 它將還會(huì)保持運(yùn)行。ExecutorService 里的活動(dòng)線程阻止了 JVM 的關(guān)閉。
要終止 ExecutorService 里的線程你需要調(diào)用 ExecutorService 的 shutdown() 方法。ExecutorService 并不會(huì)立即關(guān)閉,但它將不再接受新的任務(wù),而且一旦所有線程都完成了當(dāng)前任務(wù)的時(shí)候,ExecutorService 將會(huì)關(guān)閉。在 shutdown() 被調(diào)用之前所有提交給 ExecutorService 的任務(wù)都被執(zhí)行。如果你想要立即關(guān)閉 ExecutorService,你可以調(diào)用 shutdownNow() 方法。這樣會(huì)立即嘗試停止所有執(zhí)行中的任務(wù),并忽略掉那些已提交但尚未開始處理的任務(wù)。無(wú)法擔(dān)保執(zhí)行任務(wù)的正確執(zhí)行??赡芩鼈儽煌V沽?#xff0c;也可能已經(jīng)執(zhí)行結(jié)束。
17. 線程池執(zhí)行者 ThreadPoolExecutor
java.util.concurrent.ThreadPoolExecutor 是 ExecutorService 接口的一個(gè)實(shí)現(xiàn)。ThreadPoolExecutor 使用其內(nèi)部池中的線程執(zhí)行給定任務(wù)(Callable 或者 Runnable)。
ThreadPoolExecutor 包含的線程池能夠包含不同數(shù)量的線程。池中線程的數(shù)量由以下變量決定:
- corePoolSize
- maximumPoolSize
當(dāng)一個(gè)任務(wù)委托給線程池時(shí),如果池中線程數(shù)量低于 corePoolSize,一個(gè)新的線程將被創(chuàng)建,即使池中可能尚有空閑線程。如果內(nèi)部任務(wù)隊(duì)列已滿,而且有至少 corePoolSize 正在運(yùn)行,但是運(yùn)行線程的數(shù)量低于 maximumPoolSize,一個(gè)新的線程將被創(chuàng)建去執(zhí)行該任務(wù)。
ThreadPoolExecutor 圖解:
一個(gè) ThreadPoolExecutor
創(chuàng)建一個(gè) ThreadPoolExecutor
ThreadPoolExecutor 有若干個(gè)可用構(gòu)造子。比如:
| 123456789101112 | int? corePoolSize? =??? 5;? int? maxPoolSize?? =?? 10;? long keepAliveTime = 5000;? ExecutorService threadPoolExecutor =? ????????new ThreadPoolExecutor(? ????????????????corePoolSize,? ????????????????maxPoolSize,? ????????????????keepAliveTime,? ????????????????TimeUnit.MILLISECONDS,? ????????????????new LinkedBlockingQueue<Runnable>()? ????????????????); |
但是,除非你確實(shí)需要顯式為 ThreadPoolExecutor 定義所有參數(shù),使用 java.util.concurrent.Executors 類中的工廠方法之一會(huì)更加方便,正如?ExecutorService?小節(jié)所述。
18. 定時(shí)執(zhí)行者服務(wù) ScheduledExecutorService
java.util.concurrent.ScheduledExecutorService 是一個(gè) ExecutorService, 它能夠?qū)⑷蝿?wù)延后執(zhí)行,或者間隔固定時(shí)間多次執(zhí)行。 任務(wù)由一個(gè)工作者線程異步執(zhí)行,而不是由提交任務(wù)給 ScheduledExecutorService 的那個(gè)線程執(zhí)行。
ScheduledExecutorService 例子
以下是一個(gè)簡(jiǎn)單的 ScheduledExecutorService 示例:
| 123456789101112 | ScheduledExecutorService scheduledExecutorService =? ????????Executors.newScheduledThreadPool(5);? ScheduledFuture scheduledFuture =? ????scheduledExecutorService.schedule(new Callable() {? ????????public Object call() throws Exception {? ????????????System.out.println("Executed!");? ????????????return "Called!";? ????????}? ????},? ????5,? ????TimeUnit.SECONDS); |
首先一個(gè)內(nèi)置 5 個(gè)線程的 ScheduledExecutorService 被創(chuàng)建。之后一個(gè) Callable 接口的匿名類示例被創(chuàng)建然后傳遞給 schedule() 方法。后邊的倆參數(shù)定義了 Callable 將在 5 秒鐘之后被執(zhí)行。
ScheduledExecutorService 實(shí)現(xiàn)
既然 ScheduledExecutorService 是一個(gè)接口,你要用它的話就得使用 java.util.concurrent 包里對(duì)它的某個(gè)實(shí)現(xiàn)類。ScheduledExecutorService 具有以下實(shí)現(xiàn)類:ScheduledThreadPoolExecutor
創(chuàng)建一個(gè) ScheduledExecutorService
如何創(chuàng)建一個(gè) ScheduledExecutorService 取決于你采用的它的實(shí)現(xiàn)類。但是你也可以使用 Executors 工廠類來(lái)創(chuàng)建一個(gè) ScheduledExecutorService 實(shí)例。比如:
| 123 | ScheduledExecutorService scheduledExecutorService =? ????????Executors.newScheduledThreadPool(5); |
ScheduledExecutorService 使用
一旦你創(chuàng)建了一個(gè) ScheduledExecutorService,你可以通過(guò)調(diào)用它的以下方法:
- schedule (Callable task, long delay, TimeUnit timeunit)
- schedule (Runnable task, long delay, TimeUnit timeunit)
- scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)
- scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)
下面我們就簡(jiǎn)單看一下這些方法。
schedule (Callable task, long delay, TimeUnit timeunit)
這個(gè)方法計(jì)劃指定的 Callable 在給定的延遲之后執(zhí)行。這個(gè)方法返回一個(gè) ScheduledFuture,通過(guò)它你可以在它被執(zhí)行之前對(duì)它進(jìn)行取消,或者在它執(zhí)行之后獲取結(jié)果。以下是一個(gè)示例:
| 12345678910111213141516 | ScheduledExecutorService scheduledExecutorService =? ????????Executors.newScheduledThreadPool(5);? ScheduledFuture scheduledFuture =? ????scheduledExecutorService.schedule(new Callable() {? ????????public Object call() throws Exception {? ????????????System.out.println("Executed!");? ????????????return "Called!";? ????????}? ????},? ????5,? ????TimeUnit.SECONDS);? System.out.println("result = " + scheduledFuture.get());? scheduledExecutorService.shutdown(); |
示例輸出結(jié)果:
| 12 | Executed!result = Called! |
schedule (Runnable task, long delay, TimeUnit timeunit)
除了 Runnable 無(wú)法返回一個(gè)結(jié)果之外,這一方法工作起來(lái)就像以一個(gè) Callable 作為一個(gè)參數(shù)的那個(gè)版本的方法一樣,因此 ScheduledFuture.get() 在任務(wù)執(zhí)行結(jié)束之后返回 null。
scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)
這一方法規(guī)劃一個(gè)任務(wù)將被定期執(zhí)行。該任務(wù)將會(huì)在首個(gè) initialDelay 之后得到執(zhí)行,然后每個(gè) period 時(shí)間之后重復(fù)執(zhí)行。如果給定任務(wù)的執(zhí)行拋出了異常,該任務(wù)將不再執(zhí)行。如果沒(méi)有任何異常的話,這個(gè)任務(wù)將會(huì)持續(xù)循環(huán)執(zhí)行到 ScheduledExecutorService 被關(guān)閉。如果一個(gè)任務(wù)占用了比計(jì)劃的時(shí)間間隔更長(zhǎng)的時(shí)候,下一次執(zhí)行將在當(dāng)前執(zhí)行結(jié)束執(zhí)行才開始。計(jì)劃任務(wù)在同一時(shí)間不會(huì)有多個(gè)線程同時(shí)執(zhí)行。
scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)
除了 period 有不同的解釋之外這個(gè)方法和 scheduleAtFixedRate() 非常像。
scheduleAtFixedRate() 方法中,period 被解釋為前一個(gè)執(zhí)行的開始和下一個(gè)執(zhí)行的開始之間的間隔時(shí)間。而在本方法中,period 則被解釋為前一個(gè)執(zhí)行的結(jié)束和下一個(gè)執(zhí)行的結(jié)束之間的間隔。因此這個(gè)延遲是執(zhí)行結(jié)束之間的間隔,而不是執(zhí)行開始之間的間隔。
ScheduledExecutorService 關(guān)閉
正如 ExecutorService,在你使用結(jié)束之后你需要把 ScheduledExecutorService 關(guān)閉掉。否則他將導(dǎo)致 JVM 繼續(xù)運(yùn)行,即使所有其他線程已經(jīng)全被關(guān)閉。
你可以使用從 ExecutorService 接口繼承來(lái)的 shutdown() 或 shutdownNow() 方法將 ScheduledExecutorService 關(guān)閉。參見 ExecutorService 關(guān)閉部分以獲取更多信息。
19. 使用 ForkJoinPool 進(jìn)行分叉和合并
ForkJoinPool 在 Java 7 中被引入。它和?ExecutorService?很相似,除了一點(diǎn)不同。ForkJoinPool 讓我們可以很方便地把任務(wù)分裂成幾個(gè)更小的任務(wù),這些分裂出來(lái)的任務(wù)也將會(huì)提交給 ForkJoinPool。任務(wù)可以繼續(xù)分割成更小的子任務(wù),只要它還能分割??赡苈犉饋?lái)有些抽象,因此本節(jié)中我們將會(huì)解釋 ForkJoinPool 是如何工作的,還有任務(wù)分割是如何進(jìn)行的。
分叉和合并解釋
在我們開始看 ForkJoinPool 之前我們先來(lái)簡(jiǎn)要解釋一下分叉和合并的原理。
分叉和合并原理包含兩個(gè)遞歸進(jìn)行的步驟。兩個(gè)步驟分別是分叉步驟和合并步驟。
分叉
一個(gè)使用了分叉和合并原理的任務(wù)可以將自己分叉(分割)為更小的子任務(wù),這些子任務(wù)可以被并發(fā)執(zhí)行。如下圖所示:
通過(guò)把自己分割成多個(gè)子任務(wù),每個(gè)子任務(wù)可以由不同的 CPU 并行執(zhí)行,或者被同一個(gè) CPU 上的不同線程執(zhí)行。只有當(dāng)給的任務(wù)過(guò)大,把它分割成幾個(gè)子任務(wù)才有意義。把任務(wù)分割成子任務(wù)有一定開銷,因此對(duì)于小型任務(wù),這個(gè)分割的消耗可能比每個(gè)子任務(wù)并發(fā)執(zhí)行的消耗還要大。
什么時(shí)候把一個(gè)任務(wù)分割成子任務(wù)是有意義的,這個(gè)界限也稱作一個(gè)閥值。這要看每個(gè)任務(wù)對(duì)有意義閥值的決定。很大程度上取決于它要做的工作的種類。
合并
當(dāng)一個(gè)任務(wù)將自己分割成若干子任務(wù)之后,該任務(wù)將進(jìn)入等待所有子任務(wù)的結(jié)束之中。一旦子任務(wù)執(zhí)行結(jié)束,該任務(wù)可以把所有結(jié)果合并到同一個(gè)結(jié)果。圖示如下:
當(dāng)然,并非所有類型的任務(wù)都會(huì)返回一個(gè)結(jié)果。如果這個(gè)任務(wù)并不返回一個(gè)結(jié)果,它只需等待所有子任務(wù)執(zhí)行完畢。也就不需要結(jié)果的合并啦。
ForkJoinPool
ForkJoinPool 是一個(gè)特殊的線程池,它的設(shè)計(jì)是為了更好的配合 分叉-和-合并 任務(wù)分割的工作。ForkJoinPool 也在 java.util.concurrent 包中,其完整類名為 java.util.concurrent.ForkJoinPool。
創(chuàng)建一個(gè) ForkJoinPool
你可以通過(guò)其構(gòu)造子創(chuàng)建一個(gè) ForkJoinPool。作為傳遞給 ForkJoinPool 構(gòu)造子的一個(gè)參數(shù),你可以定義你期望的并行級(jí)別。并行級(jí)別表示你想要傳遞給 ForkJoinPool 的任務(wù)所需的線程或 CPU 數(shù)量。以下是一個(gè) ForkJoinPool 示例:
| 1 | ForkJoinPool forkJoinPool = new ForkJoinPool(4); |
這個(gè)示例創(chuàng)建了一個(gè)并行級(jí)別為 4 的 ForkJoinPool。
提交任務(wù)到 ForkJoinPool
就像提交任務(wù)到 ExecutorService 那樣,把任務(wù)提交到 ForkJoinPool。你可以提交兩種類型的任務(wù)。一種是沒(méi)有任何返回值的(一個(gè) “行動(dòng)”),另一種是有返回值的(一個(gè)”任務(wù)”)。這兩種類型分別由 RecursiveAction 和 RecursiveTask 表示。接下來(lái)介紹如何使用這兩種類型的任務(wù),以及如何對(duì)它們進(jìn)行提交。
RecursiveAction
RecursiveAction 是一種沒(méi)有任何返回值的任務(wù)。它只是做一些工作,比如寫數(shù)據(jù)到磁盤,然后就退出了。一個(gè) RecursiveAction 可以把自己的工作分割成更小的幾塊,這樣它們可以由獨(dú)立的線程或者 CPU 執(zhí)行。你可以通過(guò)繼承來(lái)實(shí)現(xiàn)一個(gè) RecursiveAction。示例如下:
| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647 | import java.util.ArrayList;? import java.util.List;? import java.util.concurrent.RecursiveAction;? public class MyRecursiveAction extends RecursiveAction {? ????private long workLoad = 0;? ????public MyRecursiveAction(long workLoad) {? ????????this.workLoad = workLoad;? ????}? ????@Override?????protected void compute() {? ????????//if work is above threshold, break tasks up into smaller tasks? ????????if(this.workLoad > 16) {? ????????????System.out.println("Splitting workLoad : " + this.workLoad);? ????????????List<MyRecursiveAction> subtasks =? ????????????????new ArrayList<MyRecursiveAction>();? ????????????subtasks.addAll(createSubtasks());? ????????????for(RecursiveAction subtask : subtasks){? ????????????????subtask.fork();? ????????????}? ????????} else {? ????????????System.out.println("Doing workLoad myself: " + this.workLoad);? ????????}? ????}? ????private List<MyRecursiveAction> createSubtasks() {? ????????List<MyRecursiveAction> subtasks =? ????????????new ArrayList<MyRecursiveAction>();? ????????MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2);? ????????MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2);? ????????subtasks.add(subtask1);? ????????subtasks.add(subtask2);? ????????return subtasks;? ????}? } |
例子很簡(jiǎn)單。MyRecursiveAction 將一個(gè)虛構(gòu)的 workLoad 作為參數(shù)傳給自己的構(gòu)造子。如果 workLoad 高于一個(gè)特定閥值,該工作將被分割為幾個(gè)子工作,子工作繼續(xù)分割。如果 workLoad 低于特定閥值,該工作將由 MyRecursiveAction 自己執(zhí)行。你可以這樣規(guī)劃一個(gè) MyRecursiveAction 的執(zhí)行:
| 123 | MyRecursiveAction myRecursiveAction = new MyRecursiveAction(24);? forkJoinPool.invoke(myRecursiveAction); |
RecursiveTask
RecursiveTask 是一種會(huì)返回結(jié)果的任務(wù)。它可以將自己的工作分割為若干更小任務(wù),并將這些子任務(wù)的執(zhí)行結(jié)果合并到一個(gè)集體結(jié)果??梢杂袔讉€(gè)水平的分割和合并。以下是一個(gè) RecursiveTask 示例:
| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152 | import java.util.ArrayList;? import java.util.List;? import java.util.concurrent.RecursiveTask;? public class MyRecursiveTask extends RecursiveTask<Long> {? ????private long workLoad = 0;? ????public MyRecursiveTask(long workLoad) {? ????????this.workLoad = workLoad;? ????}? ????protected Long compute() {? ????????//if work is above threshold, break tasks up into smaller tasks? ????????if(this.workLoad > 16) {? ????????????System.out.println("Splitting workLoad : " + this.workLoad);? ????????????List<MyRecursiveTask> subtasks =? ????????????????new ArrayList<MyRecursiveTask>();? ????????????subtasks.addAll(createSubtasks());? ????????????for(MyRecursiveTask subtask : subtasks){? ????????????????subtask.fork();? ????????????}? ????????????long result = 0;? ????????????for(MyRecursiveTask subtask : subtasks) {? ????????????????result += subtask.join();? ????????????}? ????????????return result;? ????????} else {? ????????????System.out.println("Doing workLoad myself: " + this.workLoad);? ????????????return workLoad * 3;? ????????}? ????}? ????private List<MyRecursiveTask> createSubtasks() {? ????????List<MyRecursiveTask> subtasks =? ????????new ArrayList<MyRecursiveTask>();? ????????MyRecursiveTask subtask1 = new MyRecursiveTask(this.workLoad / 2);? ????????MyRecursiveTask subtask2 = new MyRecursiveTask(this.workLoad / 2);? ????????subtasks.add(subtask1);? ????????subtasks.add(subtask2);? ????????return subtasks;? ????}? } |
除了有一個(gè)結(jié)果返回之外,這個(gè)示例和 RecursiveAction 的例子很像。MyRecursiveTask 類繼承自 RecursiveTask<Long>,這也就意味著它將返回一個(gè) Long 類型的結(jié)果。
MyRecursiveTask 示例也會(huì)將工作分割為子任務(wù),并通過(guò) fork() 方法對(duì)這些子任務(wù)計(jì)劃執(zhí)行。
此外,本示例還通過(guò)調(diào)用每個(gè)子任務(wù)的 join() 方法收集它們返回的結(jié)果。子任務(wù)的結(jié)果隨后被合并到一個(gè)更大的結(jié)果,并最終將其返回。對(duì)于不同級(jí)別的遞歸,這種子任務(wù)的結(jié)果合并可能會(huì)發(fā)生遞歸。
你可以這樣規(guī)劃一個(gè) RecursiveTask:
| 12345 | MyRecursiveTask myRecursiveTask = new MyRecursiveTask(128);? long mergedResult = forkJoinPool.invoke(myRecursiveTask);? System.out.println("mergedResult = " + mergedResult); |
注意是如何通過(guò) ForkJoinPool.invoke() 方法的調(diào)用來(lái)獲取最終執(zhí)行結(jié)果的。
ForkJoinPool 評(píng)論
貌似并非每個(gè)人都對(duì) Java 7 里的 ForkJoinPool 滿意:《一個(gè) Java 分叉-合并 帶來(lái)的災(zāi)禍》。
在你計(jì)劃在自己的項(xiàng)目里使用 ForkJoinPool 之前最好讀一下該篇文章。
20. 鎖 Lock
java.util.concurrent.locks.Lock 是一個(gè)類似于 synchronized 塊的線程同步機(jī)制。但是 Lock 比 synchronized 塊更加靈活、精細(xì)。順便說(shuō)一下,在我的《Java 并發(fā)指南》中我對(duì)如何實(shí)現(xiàn)你自己的鎖進(jìn)行了描述。
Java Lock 例子
既然 Lock 是一個(gè)接口,在你的程序里需要使用它的實(shí)現(xiàn)類之一來(lái)使用它。以下是一個(gè)簡(jiǎn)單示例:
| 1234567 | Lock lock = new ReentrantLock();? lock.lock();? //critical section? lock.unlock(); |
首先創(chuàng)建了一個(gè) Lock 對(duì)象。之后調(diào)用了它的 lock() 方法。這時(shí)候這個(gè) lock 實(shí)例就被鎖住啦。任何其他再過(guò)來(lái)調(diào)用 lock() 方法的線程將會(huì)被阻塞住,直到鎖定 lock 實(shí)例的線程調(diào)用了 unlock() 方法。最后 unlock() 被調(diào)用了,lock 對(duì)象解鎖了,其他線程可以對(duì)它進(jìn)行鎖定了。
Java Lock 實(shí)現(xiàn)
java.util.concurrent.locks 包提供了以下對(duì) Lock 接口的實(shí)現(xiàn)類:
- ReentrantLock
Lock 和 synchronized 代碼塊的主要不同點(diǎn)
一個(gè) Lock 對(duì)象和一個(gè) synchronized 代碼塊之間的主要不同點(diǎn)是:
- synchronized 代碼塊不能夠保證進(jìn)入訪問(wèn)等待的線程的先后順序。
- 你不能夠傳遞任何參數(shù)給一個(gè) synchronized 代碼塊的入口。因此,對(duì)于 synchronized 代碼塊的訪問(wèn)等待設(shè)置超時(shí)時(shí)間是不可能的事情。
- synchronized 塊必須被完整地包含在單個(gè)方法里。而一個(gè) Lock 對(duì)象可以把它的 lock() 和 unlock() 方法的調(diào)用放在不同的方法里。
Lock 的方法
Lock 接口具有以下主要方法:
- lock()
- lockInterruptibly()
- tryLock()
- tryLock(long timeout, TimeUnit timeUnit)
- unlock()
lock() 將 Lock 實(shí)例鎖定。如果該 Lock 實(shí)例已被鎖定,調(diào)用 lock() 方法的線程將會(huì)阻塞,直到 Lock 實(shí)例解鎖。
lockInterruptibly() 方法將會(huì)被調(diào)用線程鎖定,除非該線程被打斷。此外,如果一個(gè)線程在通過(guò)這個(gè)方法來(lái)鎖定 Lock 對(duì)象時(shí)進(jìn)入阻塞等待,而它被打斷了的話,該線程將會(huì)退出這個(gè)方法調(diào)用。
tryLock() 方法試圖立即鎖定 Lock 實(shí)例。如果鎖定成功,它將返回 true,如果 Lock 實(shí)例已被鎖定該方法返回 false。這一方法永不阻塞。tryLock(long timeout, TimeUnit timeUnit) 的工作類似于 tryLock() 方法,除了它在放棄鎖定 Lock 之前等待一個(gè)給定的超時(shí)時(shí)間之外。
unlock() 方法對(duì) Lock 實(shí)例解鎖。一個(gè) Lock 實(shí)現(xiàn)將只允許鎖定了該對(duì)象的線程來(lái)調(diào)用此方法。其他(沒(méi)有鎖定該 Lock 對(duì)象的線程)線程對(duì) unlock() 方法的調(diào)用將會(huì)拋一個(gè)未檢查異常(RuntimeException)。
21. 讀寫鎖 ReadWriteLock
java.util.concurrent.locks.ReadWriteLock 讀寫鎖是一種先進(jìn)的線程鎖機(jī)制。它能夠允許多個(gè)線程在同一時(shí)間對(duì)某特定資源進(jìn)行讀取,但同一時(shí)間內(nèi)只能有一個(gè)線程對(duì)其進(jìn)行寫入。
讀寫鎖的理念在于多個(gè)線程能夠?qū)σ粋€(gè)共享資源進(jìn)行讀取,而不會(huì)導(dǎo)致并發(fā)問(wèn)題。并發(fā)問(wèn)題的發(fā)生場(chǎng)景在于對(duì)一個(gè)共享資源的讀和寫操作的同時(shí)進(jìn)行,或者多個(gè)寫操作并發(fā)進(jìn)行。
本節(jié)只討論 Java 內(nèi)置 ReadWriteLock。如果你想了解 ReadWriteLock 背后的實(shí)現(xiàn)原理,請(qǐng)參考我的《Java 并發(fā)指南》主題中的《讀寫鎖》小節(jié)。
ReadWriteLock 鎖規(guī)則
一個(gè)線程在對(duì)受保護(hù)資源在讀或者寫之前對(duì) ReadWriteLock 鎖定的規(guī)則如下:
- 讀鎖:如果沒(méi)有任何寫操作線程鎖定 ReadWriteLock,并且沒(méi)有任何寫操作線程要求一個(gè)寫鎖(但還沒(méi)有獲得該鎖)。因此,可以有多個(gè)讀操作線程對(duì)該鎖進(jìn)行鎖定。
- 寫鎖:如果沒(méi)有任何讀操作或者寫操作。因此,在寫操作的時(shí)候,只能有一個(gè)線程對(duì)該鎖進(jìn)行鎖定。
ReadWriteLock 實(shí)現(xiàn)
ReadWriteLock 是個(gè)接口,如果你想用它的話就得去使用它的實(shí)現(xiàn)類之一。java.util.concurrent.locks 包提供了 ReadWriteLock 接口的以下實(shí)現(xiàn)類:
- ReentrantReadWriteLock
ReadWriteLock 代碼示例
以下是 ReadWriteLock 的創(chuàng)建以及如何使用它進(jìn)行讀、寫鎖定的簡(jiǎn)單示例代碼:
| 123456789101112131415161718 | ReadWriteLock readWriteLock = new ReentrantReadWriteLock();? readWriteLock.readLock().lock();? ????// multiple readers can enter this section? ????// if not locked for writing, and not writers waiting? ????// to lock for writing.? readWriteLock.readLock().unlock();? readWriteLock.writeLock().lock();? ????// only one writer can enter this section,? ????// and only if no threads are currently reading.? readWriteLock.writeLock().unlock(); |
注意如何使用 ReadWriteLock 對(duì)兩種鎖實(shí)例的持有。一個(gè)對(duì)讀訪問(wèn)進(jìn)行保護(hù),一個(gè)隊(duì)寫訪問(wèn)進(jìn)行保護(hù)。
22. 原子性布爾 AtomicBoolean
AtomicBoolean 類為我們提供了一個(gè)可以用原子方式進(jìn)行讀和寫的布爾值,它還擁有一些先進(jìn)的原子性操作,比如 compareAndSet()。AtomicBoolean 類位于 java.util.concurrent.atomic 包,完整類名是為 java.util.concurrent.atomic.AtomicBoolean。本小節(jié)描述的 AtomicBoolean 是 Java 8 版本里的,而不是它第一次被引入的 Java 5 版本。
AtomicBoolean 背后的設(shè)計(jì)理念在我的《Java 并發(fā)指南》主題的《比較和交換》小節(jié)有解釋。
創(chuàng)建一個(gè) AtomicBoolean
你可以這樣創(chuàng)建一個(gè) AtomicBoolean:
| 1 | AtomicBoolean atomicBoolean = new AtomicBoolean(); |
以上示例新建了一個(gè)默認(rèn)值為 false 的 AtomicBoolean。如果你想要為 AtomicBoolean 實(shí)例設(shè)置一個(gè)顯式的初始值,那么你可以將初始值傳給 AtomicBoolean 的構(gòu)造子:
| 1 | AtomicBoolean atomicBoolean = new AtomicBoolean(true); |
獲取 AtomicBoolean 的值
你可以通過(guò)使用 get() 方法來(lái)獲取一個(gè) AtomicBoolean 的值。示例如下:
| 123 | AtomicBoolean atomicBoolean = new AtomicBoolean(true);? boolean value = atomicBoolean.get(); |
以上代碼執(zhí)行后 value 變量的值將為 true。
設(shè)置 AtomicBoolean 的值
你可以通過(guò)使用 set() 方法來(lái)設(shè)置一個(gè) AtomicBoolean 的值。
示例如下:
| 123 | AtomicBoolean atomicBoolean = new AtomicBoolean(true);? atomicBoolean.set(false); |
以上代碼執(zhí)行后 AtomicBoolean 的值為 false。
交換 AtomicBoolean 的值
你可以通過(guò) getAndSet() 方法來(lái)交換一個(gè) AtomicBoolean 實(shí)例的值。getAndSet() 方法將返回 AtomicBoolean 當(dāng)前的值,并將為 AtomicBoolean 設(shè)置一個(gè)新值。示例如下:
| 123 | AtomicBoolean atomicBoolean = new AtomicBoolean(true);? boolean oldValue = atomicBoolean.getAndSet(false); |
以上代碼執(zhí)行后 oldValue 變量的值為 true,atomicBoolean 實(shí)例將持有 false 值。代碼成功將 AtomicBoolean 當(dāng)前值 ture 交換為 false。
比較并設(shè)置 AtomicBoolean 的值
compareAndSet() 方法允許你對(duì) AtomicBoolean 的當(dāng)前值與一個(gè)期望值進(jìn)行比較,如果當(dāng)前值等于期望值的話,將會(huì)對(duì) AtomicBoolean 設(shè)定一個(gè)新值。compareAndSet() 方法是原子性的,因此在同一時(shí)間之內(nèi)有單個(gè)線程執(zhí)行它。因此 compareAndSet() 方法可被用于一些類似于鎖的同步的簡(jiǎn)單實(shí)現(xiàn)。以下是一個(gè) compareAndSet() 示例:
| 1234567 | AtomicBoolean atomicBoolean = new AtomicBoolean(true);? boolean expectedValue = true;? boolean newValue????? = false;? boolean wasNewValueSet = atomicBoolean.compareAndSet(? ????expectedValue, newValue); |
本示例對(duì) AtomicBoolean 的當(dāng)前值與 true 值進(jìn)行比較,如果相等,將 AtomicBoolean 的值更新為 false。
23. 原子性整型 AtomicInteger
AtomicInteger 類為我們提供了一個(gè)可以進(jìn)行原子性讀和寫操作的 int 變量,它還包含一系列先進(jìn)的原子性操作,比如 compareAndSet()。AtomicInteger 類位于 java.util.concurrent.atomic 包,因此其完整類名為 java.util.concurrent.atomic.AtomicInteger。本小節(jié)描述的 AtomicInteger 是 Java 8 版本里的,而不是它第一次被引入的 Java 5 版本。
AtomicInteger 背后的設(shè)計(jì)理念在我的《Java 并發(fā)指南》主題的《比較和交換》小節(jié)有解釋。
創(chuàng)建一個(gè) AtomicInteger
創(chuàng)建一個(gè) AtomicInteger 示例如下:
| 1 | AtomicInteger atomicInteger = new AtomicInteger(); |
本示例將創(chuàng)建一個(gè)初始值為 0 的 AtomicInteger。如果你想要?jiǎng)?chuàng)建一個(gè)給定初始值的 AtomicInteger,你可以這樣:
| 1 | AtomicInteger atomicInteger = new AtomicInteger(123); |
本示例將 123 作為參數(shù)傳給 AtomicInteger 的構(gòu)造子,它將設(shè)置 AtomicInteger 實(shí)例的初始值為 123。
獲取 AtomicInteger 的值
你可以使用 get() 方法獲取 AtomicInteger 實(shí)例的值。示例如下:
| 123 | AtomicInteger atomicInteger = new AtomicInteger(123);int theValue = atomicInteger.get(); |
設(shè)置 AtomicInteger 的值
你可以通過(guò) set() 方法對(duì) AtomicInteger 的值進(jìn)行重新設(shè)置。以下是 AtomicInteger.set() 示例:
| 123 | AtomicInteger atomicInteger = new AtomicInteger(123);? atomicInteger.set(234); |
以上示例創(chuàng)建了一個(gè)初始值為 123 的 AtomicInteger,而在第二行將其值更新為 234。
比較并設(shè)置 AtomicInteger 的值
AtomicInteger 類也通過(guò)了一個(gè)原子性的 compareAndSet() 方法。這一方法將 AtomicInteger 實(shí)例的當(dāng)前值與期望值進(jìn)行比較,如果二者相等,為 AtomicInteger 實(shí)例設(shè)置一個(gè)新值。AtomicInteger.compareAndSet() 代碼示例:
| 12345 | AtomicInteger atomicInteger = new AtomicInteger(123);? int expectedValue = 123;? int newValue????? = 234;? atomicInteger.compareAndSet(expectedValue, newValue); |
本示例首先新建一個(gè)初始值為 123 的 AtomicInteger 實(shí)例。然后將 AtomicInteger 與期望值 123 進(jìn)行比較,如果相等,將 AtomicInteger 的值更新為 234。
增加 AtomicInteger 值
AtomicInteger 類包含有一些方法,通過(guò)它們你可以增加 AtomicInteger 的值,并獲取其值。這些方法如下:
- addAndGet()
- getAndAdd()
- getAndIncrement()
- incrementAndGet()
第一個(gè) addAndGet() 方法給 AtomicInteger 增加了一個(gè)值,然后返回增加后的值。getAndAdd() 方法為 AtomicInteger 增加了一個(gè)值,但返回的是增加以前的 AtomicInteger 的值。具體使用哪一個(gè)取決于你的應(yīng)用場(chǎng)景。以下是這兩種方法的示例:
| 12345 | AtomicInteger atomicInteger = new AtomicInteger();? System.out.println(atomicInteger.getAndAdd(10));? System.out.println(atomicInteger.addAndGet(10)); |
本示例將打印出 0 和 20。例子中,第二行拿到的是加 10 之前的 AtomicInteger 的值。加 10 之前的值是 0。第三行將 AtomicInteger 的值再加 10,并返回加操作之后的值。該值現(xiàn)在是為 20。你當(dāng)然也可以使用這倆方法為 AtomicInteger 添加負(fù)值。結(jié)果實(shí)際是一個(gè)減法操作。getAndIncrement() 和 incrementAndGet() 方法類似于 getAndAdd() 和 addAndGet(),但每次只將 AtomicInteger 的值加 1。
減小 AtomicInteger 的值
AtomicInteger 類還提供了一些減小 AtomicInteger 的值的原子性方法。這些方法是:
- decrementAndGet()
- getAndDecrement()
decrementAndGet() 將 AtomicInteger 的值減一,并返回減一后的值。getAndDecrement() 也將 AtomicInteger 的值減一,但它返回的是減一之前的值。
24. 原子性長(zhǎng)整型 AtomicLong
AtomicLong 類為我們提供了一個(gè)可以進(jìn)行原子性讀和寫操作的 long 變量,它還包含一系列先進(jìn)的原子性操作,比如 compareAndSet()AtomicLong 類位于 java.util.concurrent.atomic 包,因此其完整類名為 java.util.concurrent.atomic.AtomicLong。本小節(jié)描述的 AtomicLong 是 Java 8 版本里的,而不是它第一次被引入的 Java 5 版本。
AtomicLong 背后的設(shè)計(jì)理念在我的《Java 并發(fā)指南》主題的《比較和交換》小節(jié)有解釋。
創(chuàng)建一個(gè) AtomicLong
創(chuàng)建一個(gè) AtomicLong 如下:
| 1 | AtomicLong atomicLong = new AtomicLong(); |
將創(chuàng)建一個(gè)初始值為 0 的 AtomicLong。如果你想創(chuàng)建一個(gè)指定初始值的 AtomicLong,可以:
| 1 | AtomicLong atomicLong = new AtomicLong(123); |
本示例將 123 作為參數(shù)傳遞給 AtomicLong 的構(gòu)造子,后者將 AtomicLong 實(shí)例的初始值設(shè)置為 123。
獲取 AtomicLong 的值
你可以通過(guò) get() 方法獲取 AtomicLong 的值。AtomicLong.get() 示例:
| 123 | AtomicLong atomicLong = new AtomicLong(123);long theValue = atomicLong.get(); |
設(shè)置 AtomicLong 的值
你可以通過(guò) set() 方法設(shè)置 AtomicLong 實(shí)例的值。一個(gè) AtomicLong.set() 的示例:
| 123 | AtomicLong atomicLong = new AtomicLong(123);? atomicLong.set(234); |
本示例新建了一個(gè)初始值為 123 的 AtomicLong,第二行將其值設(shè)置為 234。
比較并設(shè)置 AtomicLong 的值
AtomicLong 類也有一個(gè)原子性的 compareAndSet() 方法。這一方法將 AtomicLong 實(shí)例的當(dāng)前值與一個(gè)期望值進(jìn)行比較,如果兩種相等,為 AtomicLong 實(shí)例設(shè)置一個(gè)新值。AtomicLong.compareAndSet() 使用示例:
| 12345 | AtomicLong atomicLong = new AtomicLong(123);? long expectedValue = 123;? long newValue????? = 234;? atomicLong.compareAndSet(expectedValue, newValue); |
本示例新建了一個(gè)初始值為 123 的 AtomicLong。然后將 AtomicLong 的當(dāng)前值與期望值 123 進(jìn)行比較,如果相等的話,AtomicLong 的新值將變?yōu)?234。
增加 AtomicLong 值
AtomicLong 具備一些能夠增加 AtomicLong 的值并返回自身值的方法。這些方法如下:
- addAndGet()
- getAndAdd()
- getAndIncrement()
- incrementAndGet()
第一個(gè)方法 addAndGet() 將 AtomicLong 的值加一個(gè)數(shù)字,并返回增加后的值。第二個(gè)方法 getAndAdd() 也將 AtomicLong 的值加一個(gè)數(shù)字,但返回的是增加前的 AtomicLong 的值。具體使用哪一個(gè)取決于你自己的場(chǎng)景。示例如下:
| 12345 | AtomicLong atomicLong = new AtomicLong();? System.out.println(atomicLong.getAndAdd(10));? System.out.println(atomicLong.addAndGet(10)); |
本示例將打印出 0 和 20。例子中,第二行拿到的是加 10 之前的 AtomicLong 的值。加 10 之前的值是 0。第三行將 AtomicLong 的值再加 10,并返回加操作之后的值。該值現(xiàn)在是為 20。你當(dāng)然也可以使用這倆方法為 AtomicLong 添加負(fù)值。結(jié)果實(shí)際是一個(gè)減法操作。getAndIncrement() 和 incrementAndGet() 方法類似于 getAndAdd() 和 addAndGet(),但每次只將 AtomicLong 的值加 1。
減小 AtomicLong 的值
AtomicLong 類還提供了一些減小 AtomicLong 的值的原子性方法。這些方法是:
- decrementAndGet()
- getAndDecrement()
decrementAndGet() 將 AtomicLong 的值減一,并返回減一后的值。getAndDecrement() 也將 AtomicLong 的值減一,但它返回的是減一之前的值。
25. 原子性引用型 AtomicReference
AtomicReference 提供了一個(gè)可以被原子性讀和寫的對(duì)象引用變量。原子性的意思是多個(gè)想要改變同一個(gè) AtomicReference 的線程不會(huì)導(dǎo)致 AtomicReference 處于不一致的狀態(tài)。AtomicReference 還有一個(gè) compareAndSet() 方法,通過(guò)它你可以將當(dāng)前引用于一個(gè)期望值(引用)進(jìn)行比較,如果相等,在該 AtomicReference 對(duì)象內(nèi)部設(shè)置一個(gè)新的引用。
創(chuàng)建一個(gè) AtomicReference
創(chuàng)建 AtomicReference 如下:
| 1 | AtomicReference atomicReference = new AtomicReference(); |
如果你需要使用一個(gè)指定引用創(chuàng)建 AtomicReference,可以:
| 12 | String initialReference = "the initially referenced string";? AtomicReference atomicReference = new AtomicReference(initialReference); |
創(chuàng)建泛型 AtomicReference
你可以使用 Java 泛型來(lái)創(chuàng)建一個(gè)泛型 AtomicReference。示例:
| 12 | AtomicReference<String> atomicStringReference =? ????new AtomicReference<String>(); |
你也可以為泛型 AtomicReference 設(shè)置一個(gè)初始值。示例:
| 123 | String initialReference = "the initially referenced string";? AtomicReference<String> atomicStringReference =? ????new AtomicReference<String>(initialReference); |
獲取 AtomicReference 引用
你可以通過(guò) AtomicReference 的 get() 方法來(lái)獲取保存在 AtomicReference 里的引用。如果你的 AtomicReference 是非泛型的,get() 方法將返回一個(gè) Object 類型的引用。如果是泛型化的,get() 將返回你創(chuàng)建 AtomicReference 時(shí)聲明的那個(gè)類型。先來(lái)看一個(gè)非泛型的 AtomicReference get() 示例:
| 123 | AtomicReference atomicReference = new AtomicReference("first value referenced");? String reference = (String) atomicReference.get(); |
注意如何對(duì) get() 方法返回的引用強(qiáng)制轉(zhuǎn)換為 String。泛型化的 AtomicReference 示例:
| 1234 | AtomicReference<String> atomicReference =?? ?????new AtomicReference<String>("first value referenced");? String reference = atomicReference.get(); |
編譯器知道了引用的類型,所以我們無(wú)需再對(duì) get() 返回的引用進(jìn)行強(qiáng)制轉(zhuǎn)換了。
設(shè)置 AtomicReference 引用
你可以使用 get() 方法對(duì) AtomicReference 里邊保存的引用進(jìn)行設(shè)置。如果你定義的是一個(gè)非泛型 AtomicReference,set() 將會(huì)以一個(gè) Object 引用作為參數(shù)。如果是泛型化的 AtomicReference,set() 方法將只接受你定義給的類型。AtomicReference set() 示例:
| 123 | ?????new AtomicReference();? atomicReference.set("New object referenced"); |
這個(gè)看起來(lái)非泛型和泛型化的沒(méi)啥區(qū)別。真正的區(qū)別在于編譯器將對(duì)你能夠設(shè)置給一個(gè)泛型化的 AtomicReference 參數(shù)類型進(jìn)行限制。
比較并設(shè)置 AtomicReference 引用
AtomicReference 類具備了一個(gè)很有用的方法:compareAndSet()。compareAndSet() 可以將保存在 AtomicReference 里的引用于一個(gè)期望引用進(jìn)行比較,如果兩個(gè)引用是一樣的(并非 equals() 的相等,而是 == 的一樣),將會(huì)給 AtomicReference 實(shí)例設(shè)置一個(gè)新的引用。
如果 compareAndSet() 為 AtomicReference 設(shè)置了一個(gè)新的引用,compareAndSet() 將返回 true。否則 compareAndSet() 返回 false。AtomicReference compareAndSet() 示例:
| 1234567891011 | String initialReference = "initial value referenced";? AtomicReference<String> atomicStringReference =? ????new AtomicReference<String>(initialReference);? String newReference = "new value referenced";? boolean exchanged = atomicStringReference.compareAndSet(initialReference, newReference);? System.out.println("exchanged: " + exchanged);? exchanged = atomicStringReference.compareAndSet(initialReference, newReference);? System.out.println("exchanged: " + exchanged); |
本示例創(chuàng)建了一個(gè)帶有一個(gè)初始引用的泛型化的 AtomicReference。之后兩次調(diào)用 comparesAndSet()來(lái)對(duì)存儲(chǔ)值和期望值進(jìn)行對(duì)比,如果二者一致,為 AtomicReference 設(shè)置一個(gè)新的引用。第一次比較,存儲(chǔ)的引用(initialReference)和期望的引用(initialReference)一致,所以一個(gè)新的引用(newReference)被設(shè)置給 AtomicReference,compareAndSet() 方法返回 true。第二次比較時(shí),存儲(chǔ)的引用(newReference)和期望的引用(initialReference)不一致,因此新的引用沒(méi)有被設(shè)置給 AtomicReference,compareAndSet() 方法返回 false。
創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯,堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)
總結(jié)
以上是生活随笔為你收集整理的并发编程 – Concurrent 用户指南的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇: 微软计划在 Win11 中禁用 NTLM
- 下一篇: 扁扁的鱼叫什么鱼 扁扁的鱼叫什么名字