日韩av黄I国产麻豆传媒I国产91av视频在线观看I日韩一区二区三区在线看I美女国产在线I麻豆视频国产在线观看I成人黄色短片

歡迎訪問(wèn) 生活随笔!

生活随笔

當(dāng)前位置: 首頁(yè) >

Java并发编程-并发工具包java.util.concurrent使用指南

發(fā)布時(shí)間:2025/4/16 57 豆豆
生活随笔 收集整理的這篇文章主要介紹了 Java并发编程-并发工具包java.util.concurrent使用指南 小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

譯序

本指南根據(jù) Jakob Jenkov 最新博客翻譯,請(qǐng)隨時(shí)關(guān)注博客更新
本指南已做成中英文對(duì)照閱讀版的 pdf 文檔,有興趣的朋友可以去 Java并發(fā)工具包java.util.concurrent用戶指南中英文對(duì)照閱讀版 進(jìn)行下載。

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ā)指南》

2.1 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ì)列。

2.2 BlockingQueue的方法

BlockingQueue 具有 4 組不同的方法用于插入、移除以及對(duì)隊(duì)列中的元素進(jìn)行檢查。如果請(qǐng)求的操作不能得到立即執(zhí)行的話,每個(gè)方法的表現(xiàn)也不同。這些方法如下:

操作拋異常特定值阻塞超時(shí)
插入add(o)offer(o)put(o)offer(o, timeout, timeunit)
移除remove(o)poll(o)take(o)poll(timeout, timeunit)
檢查element(o)peek(o)不可用不可用

四組不同的行為方式解釋:

  • 拋異常:如果試圖的操作無(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 中的所有元素,而不僅僅是開(kāi)始和結(jié)束的元素。比如說(shuō),你將一個(gè)對(duì)象放入隊(duì)列之中以等待處理,但你的應(yīng)用想要將其取消掉。那么你可以調(diào)用諸如 remove(o) 方法來(lái)將隊(duì)列之中的特定對(duì)象進(jìn)行移除。但是這么干效率并不高(譯者注:基于隊(duì)列的數(shù)據(jù)結(jié)構(gòu),獲取除開(kāi)始或結(jié)束位置的其他對(duì)象的效率不會(huì)太高),因此你盡量不要用這一類的方法,除非你確實(shí)不得不那么做。

2.3 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

2.4 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)。

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ā)生阻塞。

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

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è)示例:

BlockingQueue queue = new ArrayBlockingQueue(1024); queue.put("1"); Object object = queue.take();

以下是使用了 Java 泛型的一個(gè) BlockingQueue 示例。注意其中是如何對(duì) String 元素放入和提取的:

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 接口,該接口定義:

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ì)取以下值:

  • 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 的例子:

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 的初始化和使用示例代碼:

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 的示例:

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” 的縮寫(xiě)。因此,雙端隊(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。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)也不同。這些方法如下:

操作拋異常特定值阻塞超時(shí)
插入addFirst(o)offerFirst(o)putFirst(o)offerFirst(o, timeout, timeunit)
移除removeFirst(o)pollFirst(o)takeFirst(o)pollFirst(timeout, timeunit)
檢查getFirst(o)peekFirst(o)無(wú)無(wú)
操作拋異常特定值阻塞超時(shí)
插入addLast(o)offerLast(o)putLast(o)offerLast(o, timeout, timeunit)
移除removeLast(o)pollLast(o)takeLast(o)pollLast(timeout, timeunit)
檢查getLast(o)peekLast(o)無(wú)無(wú)

四組不同的行為方式解釋:

  • 拋異常:如果試圖的操作無(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):

BlockingQueueBlockingDeque
add()addLast()
offer()offerLast()
put()putLast()
offer(e, time, unit)offerLast(e, time, unit)
remove()removeFirst()
poll()pollFirst()
take()takeFirst()
poll(time, unit)pollLast(time, unit)
element()getFirst()
peek()peekFirst()

BlockingDeque 的實(shí)現(xiàn)

既然 BlockingDeque 是一個(gè)接口,那么你想要使用它的話就得使用它的眾多的實(shí)現(xiàn)類的其中一個(gè)。java.util.concurrent 包提供了以下 BlockingDeque 接口的實(shí)現(xiàn)類:LinkedBlockingDeque

BlockingDeque 代碼示例

以下是如何使用 BlockingDeque 方法的一個(gè)簡(jiǎn)短代碼示例:

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” 的縮寫(xiě)。因此,雙端隊(duì)列是一個(gè)你可以從任意一端插入或者抽取元素的隊(duì)列。

LinkedBlockingDeque 是一個(gè)雙端隊(duì)列,在它為空的時(shí)候,一個(gè)試圖從中抽取數(shù)據(jù)的線程將會(huì)阻塞,無(wú)論該線程是試圖從哪一端抽取數(shù)據(jù)。

以下是 LinkedBlockingDeque 實(shí)例化以及使用的示例:

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 鎖住。此外,在你向其中寫(xiě)入對(duì)象的時(shí)候,ConcurrentHashMap 也不會(huì)鎖住整個(gè) Map。它的內(nèi)部只是把 Map 中正在被寫(xiě)入的部分進(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)類:

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() 方法的使用:

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() 方法的使用:

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。示例如下:

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)。

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 示例:

CyclicBarrier barrier = new CyclicBarrier(2);

等待一個(gè) CyclicBarrier

以下演示了如何讓一個(gè)線程等待一個(gè) CyclicBarrier:

barrier.await();

當(dāng)然,你也可以為等待線程設(shè)定一個(gè)超時(shí)時(shí)間。等待超過(guò)了超時(shí)時(shí)間之后,即便還沒(méi)有達(dá)成 N 個(gè)線程等待 CyclicBarrier 的條件,該線程也會(huì)被釋放出來(lái)。以下是定義超時(shí)時(shí)間示例:

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)傳給它:

Runnable barrierAction = ... ; CyclicBarrier barrier = new CyclicBarrier(2, barrierAction);

CyclicBarrier 示例

以下代碼演示了如何使用 CyclicBarrier:

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 類:

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è)線程寫(xiě)入控制臺(tái)的時(shí)序可能會(huì)跟你實(shí)際執(zhí)行不一樣。比如有時(shí) Thread-0 先打印,有時(shí) Thread-1 先打印。

Thread-0 waiting at barrier 1 Thread-1 waiting at barrier 1 BarrierAction 1 executed Thread-1 waiting at barrier 2 Thread-0 waiting at barrier 2 BarrierAction 2 executed Thread-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è)示例:

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 代碼: ?```java 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 B Thread-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ù)一個(gè)重要(代碼)部分防止一次超過(guò) N 個(gè)線程進(jìn)入
  • 在兩個(gè)線程之間發(fā)送信號(hào)

保護(hù)重要部分

如果你將信號(hào)量用于保護(hù)一個(gè)重要部分,試圖進(jìn)入這一部分的代碼通常會(huì)首先嘗試獲得一個(gè)許可,然后才能進(jìn)入重要部分(代碼塊),執(zhí)行完之后,再把許可釋放掉。比如這樣:

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 的示例:

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 例子:

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í)例的例子:

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 的示例:

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() 示例:

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 示例:

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());

以上代碼輸出:

Asynchronous Callable future.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 將被取消。

以下是示例代碼:

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è)代碼示例:

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ù),并忽略掉那些已提交但尚未開(kāi)始處理的任務(wù)。無(wú)法擔(dān)保執(zhí)行任務(wù)的正確執(zhí)行。可能它們被停止了,也可能已經(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)造子。比如:

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 示例:

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í)例。比如:

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è)示例:

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é)果:

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í)行才開(kāi)始。計(jì)劃任務(wù)在同一時(shí)間不會(huì)有多個(gè)線程同時(shí)執(zhí)行。

scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)

除了 period 有不同的解釋之外這個(gè)方法和 scheduleAtFixedRate() 非常像。

scheduleAtFixedRate() 方法中,period 被解釋為前一個(gè)執(zhí)行的開(kāi)始和下一個(gè)執(zhí)行的開(kāi)始之間的間隔時(shí)間。

而在本方法中,period 則被解釋為前一個(gè)執(zhí)行的結(jié)束和下一個(gè)執(zhí)行的結(jié)束之間的間隔。因此這個(gè)延遲是執(zhí)行結(jié)束之間的間隔,而不是執(zhí)行開(kāi)始之間的間隔。

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)閉。參見(jià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ù),只要它還能分割。可能聽(tīng)起來(lái)有些抽象,因此本節(jié)中我們將會(huì)解釋 ForkJoinPool 是如何工作的,還有任務(wù)分割是如何進(jìn)行的。

分叉和合并解釋

在我們開(kāi)始看 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ù)有一定開(kāi)銷,因此對(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 示例:

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ù)。它只是做一些工作,比如寫(xiě)數(shù)據(jù)到磁盤,然后就退出了。
一個(gè) RecursiveAction 可以把自己的工作分割成更小的幾塊,這樣它們可以由獨(dú)立的線程或者 CPU 執(zhí)行。
你可以通過(guò)繼承來(lái)實(shí)現(xiàn)一個(gè) RecursiveAction。示例如下:

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í)行:

MyRecursiveAction myRecursiveAction = new MyRecursiveAction(24); forkJoinPool.invoke(myRecursiveAction);

RecursiveTask

RecursiveTask 是一種會(huì)返回結(jié)果的任務(wù)。它可以將自己的工作分割為若干更小任務(wù),并將這些子任務(wù)的執(zhí)行結(jié)果合并到一個(gè)集體結(jié)果。可以有幾個(gè)水平的分割和合并。以下是一個(gè) RecursiveTask 示例:

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,這也就意味著它將返回一個(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:

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)單示例:

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()

lock() 將 Lock 實(shí)例鎖定。如果該 Lock 實(shí)例已被鎖定,調(diào)用 lock() 方法的線程將會(huì)阻塞,直到 Lock 實(shí)例解鎖。

  • lockInterruptibly()

lockInterruptibly() 方法將會(huì)被調(diào)用線程鎖定,除非該線程被打斷。此外,如果一個(gè)線程在通過(guò)這個(gè)方法來(lái)鎖定 Lock 對(duì)象時(shí)進(jìn)入阻塞等待,而它被打斷了的話,該線程將會(huì)退出這個(gè)方法調(diào)用。

  • tryLock()

tryLock() 方法試圖立即鎖定 Lock 實(shí)例。如果鎖定成功,它將返回 true,如果 Lock 實(shí)例已被鎖定該方法返回 false。這一方法永不阻塞。

  • tryLock(long timeout, TimeUnit timeUnit)

tryLock(long timeout, TimeUnit timeUnit) 的工作類似于 tryLock() 方法,除了它在放棄鎖定 Lock 之前等待一個(gè)給定的超時(shí)時(shí)間之外。

  • unlock()

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. 讀寫(xiě)鎖 ReadWriteLock

java.util.concurrent.locks.ReadWriteLock 讀寫(xiě)鎖是一種先進(jìn)的線程鎖機(jī)制。它能夠允許多個(gè)線程在同一時(shí)間對(duì)某特定資源進(jìn)行讀取,但同一時(shí)間內(nèi)只能有一個(gè)線程對(duì)其進(jìn)行寫(xiě)入。

讀寫(xiě)鎖的理念在于多個(gè)線程能夠?qū)σ粋€(gè)共享資源進(jìn)行讀取,而不會(huì)導(dǎo)致并發(fā)問(wèn)題。并發(fā)問(wèn)題的發(fā)生場(chǎng)景在于對(duì)一個(gè)共享資源的讀和寫(xiě)操作的同時(shí)進(jìn)行,或者多個(gè)寫(xiě)操作并發(fā)進(jìn)行。

本節(jié)只討論 Java 內(nèi)置 ReadWriteLock。如果你想了解 ReadWriteLock 背后的實(shí)現(xiàn)原理,請(qǐng)參考我的《Java 并發(fā)指南》主題中的《讀寫(xiě)鎖》小節(jié)。

ReadWriteLock 鎖規(guī)則

一個(gè)線程在對(duì)受保護(hù)資源在讀或者寫(xiě)之前對(duì) ReadWriteLock 鎖定的規(guī)則如下:
讀鎖:如果沒(méi)有任何寫(xiě)操作線程鎖定 ReadWriteLock,并且沒(méi)有任何寫(xiě)操作線程要求一個(gè)寫(xiě)鎖(但還沒(méi)有獲得該鎖)。因此,可以有多個(gè)讀操作線程對(duì)該鎖進(jìn)行鎖定。

寫(xiě)鎖:如果沒(méi)有任何讀操作或者寫(xiě)操作。因此,在寫(xiě)操作的時(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)行讀、寫(xiě)鎖定的簡(jiǎn)單示例代碼:

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ì)寫(xiě)訪問(wèn)進(jìn)行保護(hù)。

22. 原子性布爾 AtomicBoolean

AtomicBoolean 類為我們提供了一個(gè)可以用原子方式進(jìn)行讀和寫(xiě)的布爾值,它還擁有一些先進(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:

AtomicBoolean atomicBoolean = new AtomicBoolean();

以上示例新建了一個(gè)默認(rèn)值為 false 的 AtomicBoolean。

如果你想要為 AtomicBoolean 實(shí)例設(shè)置一個(gè)顯式的初始值,那么你可以將初始值傳給 AtomicBoolean 的構(gòu)造子:

AtomicBoolean atomicBoolean = new AtomicBoolean(true);

獲取 AtomicBoolean 的值

你可以通過(guò)使用 get() 方法來(lái)獲取一個(gè) AtomicBoolean 的值。示例如下:

AtomicBoolean atomicBoolean = new AtomicBoolean(true); boolean value = atomicBoolean.get();

以上代碼執(zhí)行后 value 變量的值將為 true。

設(shè)置 AtomicBoolean 的值

你可以通過(guò)使用 set() 方法來(lái)設(shè)置一個(gè) AtomicBoolean 的值。示例如下:

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è)新值。示例如下:

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() 示例:

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)行原子性讀和寫(xiě)操作的 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 示例如下:

AtomicInteger atomicInteger = new AtomicInteger();

本示例將創(chuàng)建一個(gè)初始值為 0 的 AtomicInteger。
如果你想要?jiǎng)?chuàng)建一個(gè)給定初始值的 AtomicInteger,你可以這樣:

AtomicInteger atomicInteger = new AtomicInteger(123);

本示例將 123 作為參數(shù)傳給 AtomicInteger 的構(gòu)造子,它將設(shè)置 AtomicInteger 實(shí)例的初始值為 123。

獲取 AtomicInteger 的值

你可以使用 get() 方法獲取 AtomicInteger 實(shí)例的值。示例如下:

AtomicInteger atomicInteger = new AtomicInteger(123); int theValue = atomicInteger.get();

設(shè)置 AtomicInteger 的值

你可以通過(guò) set() 方法對(duì) AtomicInteger 的值進(jìn)行重新設(shè)置。以下是 AtomicInteger.set() 示例:

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() 代碼示例:

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)景。以下是這兩種方法的示例:

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)行原子性讀和寫(xiě)操作的 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 如下:

AtomicLong atomicLong = new AtomicLong();

將創(chuàng)建一個(gè)初始值為 0 的 AtomicLong。
如果你想創(chuàng)建一個(gè)指定初始值的 AtomicLong,可以:

AtomicLong atomicLong = new AtomicLong(123);

本示例將 123 作為參數(shù)傳遞給 AtomicLong 的構(gòu)造子,后者將 AtomicLong 實(shí)例的初始值設(shè)置為 123。
獲取 AtomicLong 的值
你可以通過(guò) get() 方法獲取 AtomicLong 的值。AtomicLong.get() 示例:

AtomicLong atomicLong = new AtomicLong(123); long theValue = atomicLong.get();

設(shè)置 AtomicLong 的值
你可以通過(guò) set() 方法設(shè)置 AtomicLong 實(shí)例的值。一個(gè) AtomicLong.set() 的示例:

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() 使用示例:

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)景。示例如下:

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è)可以被原子性讀和寫(xiě)的對(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 如下:

AtomicReference atomicReference = new AtomicReference();

如果你需要使用一個(gè)指定引用創(chuàng)建 AtomicReference,可以:

String initialReference = "the initially referenced string"; AtomicReference atomicReference = new AtomicReference(initialReference);

創(chuàng)建泛型 AtomicReference

你可以使用 Java 泛型來(lái)創(chuàng)建一個(gè)泛型 AtomicReference。示例:

AtomicReference<String> atomicStringReference = new AtomicReference<String>();

你也可以為泛型 AtomicReference 設(shè)置一個(gè)初始值。示例:

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() 示例:

AtomicReference atomicReference = new AtomicReference("first value referenced"); String reference = (String) atomicReference.get();

注意如何對(duì) get() 方法返回的引用強(qiáng)制轉(zhuǎn)換為 String。
泛型化的 AtomicReference 示例:

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() 示例:

AtomicReference atomicReference = 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() 示例:

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。

原文鏈接:http://tutorials.jenkov.com/java-util-concurrent/index.html
譯文鏈接:http://blog.csdn.net/defonds/article/details/44021605#t8

總結(jié)

以上是生活随笔為你收集整理的Java并发编程-并发工具包java.util.concurrent使用指南的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問(wèn)題。

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