投稿
  2. 程式人生 > java教程 > 《Java Concurrency in Practice》中文版筆記

《Java Concurrency in Practice》中文版筆記

• java教程• 發佈：2018-10-03
第1章 簡介 1.1 併發簡史
  1. 茶壺和麵包機的生產商都很清楚:使用者通常會採用非同步方式來使用他們的產品,因此當這些機器完成任務時都會發出聲音提示。
1.2 執行緒的優勢
  1. 執行緒能夠將大部分的非同步工作流轉換成序列工作流,因此能更好地模擬人類的工作方式和互動方式。
  2. 執行緒還可以簡化JVM的實現,垃圾收集器通常在一個或多個專門的執行緒中執行。
  3. 因此,作業系統提供了一些高效的方法來實現多路I/O,例如Unix的select和poll等系統呼叫,要呼叫這些方法,Java類庫需要獲得一組實現非阻塞I/O的包(java.nio)。
1.3 執行緒帶來的風險
  1. 安全性的含義是“永遠不發生糟糕的事情”,而活躍性則關注於另一個目標,即“某件正確的事情最終會發生”。當某個操作無法繼續執行下去時,就會發生活躍性問題。在序列程式中,活躍性問題的形式之一就是無意中造成的無線迴圈,從而使迴圈之後的程式碼無法得到執行。執行緒將帶來其他一些活躍性問題。例如,如果執行緒A在等待執行緒B釋放其持有的資源,而執行緒B永遠都不釋放該資源,那麼A就會永久地等待下去。
  2. 在多執行緒程式中,當執行緒排程器臨時掛起活躍執行緒並轉而執行另一個執行緒時,就會頻繁地出現上下文切換操作(Context Switch),這種操作將帶來極大的開銷:儲存和恢復執行上下文,丟失區域性性,並且CPU時間將更多地花線上程排程而不是執行緒執行上。
1.4 執行緒無處不在
  1. 每個Java應用程式都會使用執行緒。當JVM啟動時,它將為JVM的內部任務(例如,垃圾收集、終結操作等)建立後臺執行緒,並建立一個主執行緒來執行main方法。AWT (Abstract Window Toolkit)和Swing的使用者介面框架將建立執行緒來管理使用者介面事件。Timer將建立執行緒來執行延遲任務。一些元件框架,例如Servlet和RMI,都會建立執行緒池並呼叫這些執行緒中的方法。
  2. 當某個框架在應用程式中引入併發性時,通常不可能將併發性僅侷限於框架程式碼,因為框架本身會回撥(Callback)應用程式的程式碼,而這些程式碼將訪問應用程式的狀態。同樣,對執行緒安全性的需求也不能侷限於被呼叫的程式碼,而是要延伸到需要訪問這些程式碼所訪問的程式狀態的所有程式碼路徑。
  3. 框架通過在框架執行緒中呼叫應用程式程式碼將併發性引入到程式中。在程式碼中將不可避免地訪問應用程式狀態,因此所有訪問這些狀態的程式碼路徑都必須是執行緒安全的。
第2章 執行緒安全性

從非正式的意義上來說,物件的狀態是指儲存在狀態變數(例如例項或靜態域)中的資料。物件的狀態可能包括其他依賴物件的域。例如,某個HashMap的狀態不僅儲存在HashMap物件本身,還儲存在許多Map.Entry物件中。在物件的狀態中包含了任何可能影響其外部可見行為的資料。

訪問某個變數的程式碼越少,就越容易確保對變數的所有訪問都實現正確同步,同時也更容易找出變數在哪些條件下被訪問。Java語言並沒有強制要求將狀態都封裝在類中,開發人員完全可以將狀態儲存在某個公開的域(甚至公開的靜態域)中,或者提供一個對內部物件的公開引用。然而,程式狀態的封裝性越好,就越容易實現程式的執行緒安全性,並且程式碼的維護人員也越容易保持這種方式。

2.1 什麼是執行緒安全性
  1. 當多個執行緒訪問某個類時,不管執行時環境採用何種排程方式或者這些執行緒將如何交替執行,並且在主調程式碼中不需要任何額外的同步或協同,這個類都能表現出正確的行為,那麼就稱這個類是執行緒安全的。
  2. 與大多數Servlet相同,StatelessFactorizer是無狀態的:它既不包含任何域,也不包含任何對其他類中域的引用。計算過程中的臨時狀態僅存在於執行緒棧上的區域性變數中,並且只能由正在執行的執行緒訪問。訪問StatelessFactorizer的執行緒不會影響另一個訪問同一個StatelessFactorizer的執行緒的計算結果,因為這兩個執行緒並沒有共享狀態,就好像它們都在訪問不同的例項。由於執行緒訪問無狀態物件的行為並不會影響其他執行緒中操作的正確性,因此無狀態物件是執行緒安全的。【方法內的區域性變數是執行緒獨有的,所以不會出問題。】
2.2 競態條件
  1. 最常見的競態條件型別就是“先檢查後執行(Check-Then-Act)”操作,即通過一個可能失效的觀測結果來決定下一步的動作。
  2. 當你邁出前門時,你在星巴克A的觀察結果將變得無效,你的朋友可能從後門進來了,而你卻不知道。這種觀察結果的失效就是大多數競態條件的本質——基於一種可能失效的觀察結果來做出判斷或者執行某個計算。
  3. 當在無狀態的類中新增一個狀態時,如果該狀態完全由執行緒安全的物件來管理,那麼這個類仍然是執行緒安全的。然而,當狀態變數的數量由一個變為多個時,並不會像狀態變數數量為零個變為一個那樣簡單。
2.3 加鎖機制
  1. UnsafeCachingFactorizer的不變性條件之一是:在lastFactors中快取的因數之積應該等於在lastNumber中快取的數值。只有確保了這個不變性條件不被破壞,上面的Servlet才是正確的。【當在不變性條件中涉及多個變數時,各個變數之間並不是彼此獨立的,而是某個變數的值會對其他變數的值產生約束。】因此,當更新某一個變數時,需要在同一個原子操作中隊其他變數同時進行更新。【要保持狀態的一致性,就需要在單個原子操作中更新所有相關的狀態變數。】
  2. 每個Java物件都可以用做一個實現同步的鎖,這些鎖被稱為內建鎖(Intrinsic Lock)或監視鎖(Monitor Lock)。
  3. 以關鍵字synchronized來修飾的方法就是一種橫跨整個方法體的同步程式碼塊,其中該同步程式碼塊的鎖就是方法呼叫所在的物件。靜態的synchronized方法以Class物件作為鎖。
  4. “重入”意味著獲取鎖的操作的粒度是“執行緒”,而不是“呼叫”。
  5. 子類改寫了父類的synchronized方法,然後呼叫父類中的方法,此時如果沒有可重入的鎖,那麼這段程式碼將產生死鎖。由於Widget和LoggingWidget中doSomething方法都是synchronized方法,因此每個doSomething方法在執行前都會獲取Widget上的鎖。【這意味著,呼叫子類重寫的synchronized方法,會同時鎖住父類物件和子類物件?】【不對,只有一個物件,只有一個鎖,這個物件叫Widget也好,叫LoggingWidget也好,還是隻有一個物件,也就是隻有一個鎖】
2.5 活躍性與效能
  1. 當訪問狀態變數或者在複合操作的執行期間,CachedFactorizer需要持有鎖,但在執行時間較長的因數分解運算之前要釋放鎖。這樣既確保了執行緒安全性,也不會過多地影像併發性,而且在每個同步程式碼塊中的程式碼路徑都“足夠短”。
  2. 當執行時間較長的計算或者可能無法快速完成的操作時(例如,網路I/O或控制檯I/O),一定不要持有鎖。
第3章 物件的共享

同步還有另一個重要的方面:記憶體可見性(Memory Visibility)。我們不僅希望防止某個執行緒正在使用物件狀態而另一個執行緒在同時修改該狀態,而且希望確保當一個執行緒修改了物件狀態後,其他執行緒能夠看到發生的狀態變化。如果沒有同步,那麼這種情況就無法實現。

3.1 可見性
  1. NoVisibility可能會持續迴圈下去,因為讀執行緒可能永遠都看不到ready的值。一種更奇怪的現象是,NoVisibility可能會輸出0,因為讀執行緒可能看到了寫入ready的值,但卻沒有看到之後寫入number的值,這種現象被稱為“重排序(Reordering)”。
  2. 當主執行緒首先寫入number,然後在沒有同步的情況下寫入ready,那麼讀執行緒看到的順序可能與寫入的順序完全相反。
  3. 加鎖的含義不僅僅侷限於互斥行為,還包括記憶體可見性。為了確保所有執行緒都能看到共享變數的最新值,所有執行讀操作或者寫操作的執行緒都必須在同一個鎖上同步。
  4. 當把變數宣告為volatile型別後,編譯器與執行時都會注意到這個變數是共享的,因此不會將該變數上的操作與其他記憶體操作一起重排序。volatile變數不會被快取在暫存器或者對其他處理器不可見的地方,因此在讀取volatile型別的變數時總會返回最新寫入的值。
  5. 然而,我們並不建議過度依賴volatile變數提供的可見性。如果在程式碼中依賴volatile變數來控制狀態的可見性,通常比使用鎖的程式碼更脆弱,也更難以理解。
  6. 如果在驗證正確性時需要對可見性進行復雜的判斷,那麼就不要使用volatile變數。volatile變數的正確使用方式包括:確保它們自身狀態的可見性,確保它們所引用物件的狀態的可見性,【以及標識一些重要的程式生命週期事件的發生(例如,初始化或關閉】。
  7. volatile變數的一種典型用法:檢查某個狀態標記以判斷是否退出迴圈。volatile變數通常用做某個操作完成、發生中斷或者狀態的標誌。
  8. 加鎖機制既可以確保可見性又可以確保原子性,而volatile變數只能確保可見性。
  9. 當且僅當滿足以下所有條件時,才應該使用volatile變數:
    對變數的寫入操作不依賴變數的當前值,或者你能確保只有單個執行緒更新變數的值。
    該變數不會與其他狀態變數一起納入不變性條件中。
    在訪問變數時不需要加鎖。
3.2 釋出與逸出
  1. “釋出(Publish)”一個物件的意思是指,使物件能夠在當前作用域之外的程式碼中使用。例如,將一個指向該物件的引用儲存到其他程式碼可以訪問的地方,或者在某一個非私有的方法中返回該引用,或者將引用傳遞到其他類的方法中。
  2. 釋出內部狀態可能會破壞封裝性,並使得程式難以維持不變性條件。例如,如果在物件構造完成之前就釋出該物件,就會破壞執行緒安全性。當某個不應該釋出的物件被髮布時,這種情況就被稱為逸出(Escape)。
    3. 3.
class UnsafeStates { private String[] states = new String[] { "AK", "AL" ... }; public String[] getStates() { return states; }}

如果按照上述方式來發布states,就會出現問題,因為任何呼叫者都能夠修改這個陣列的內容。在這個示例中,陣列states已經逸出了它所在的作用域,因為這個本應是私有的變數已經被髮布了。
4. 假定有一個類C,對於C來說,“外部(Alien)方法”是指行為並不完全由C來規定的方法,包括其他類中定義的方法以及類C中可以被改寫的方法(既不是私有[private]方法也不是終結[final]方法)。
5. 最後一種釋出物件或其內部狀態的機制就是釋出一個內部的類例項。

public class ThisEscape { public ThisEscape(EventSource source) { source.registerListener( new EventListener() { public void onEvent(Event e) { doSomething(e); }});}}

當ThisEscape釋出EventListener時,也隱含地釋出了ThisEscape例項本身,因為在這個內部類的例項中包含了對ThisEscape例項的隱含引用。
6. 當且僅當物件的建構函式返回時,物件才處於可預測的和一致的狀態。因此,當從物件的建構函式中釋出物件時,只是釋出了一個尚未構造完成的物件。即使釋出物件的語句位於建構函式的最後一行也是如此。如果this引用在構造過程中逸出,那麼這種現象就被認為是不正確構造。
7. 在構造過程中使this引用逸出的一個常見錯誤是,在建構函式中啟動一個執行緒。當物件在其建構函式中建立一個執行緒時,無論是顯式建立(通過將它傳給建構函式)還是隱式建立(由於Thread或Runnable是該物件的一個內部類),this應用都會被新建立的執行緒共享。在物件尚未完全構造之前,新的執行緒就可以看見它。
8. 在建構函式中呼叫一個可改寫的例項方法時,同樣會導致this引用在構造過程中逸出。
9. 如果想在建構函式中註冊一個事件監聽器或啟動執行緒,那麼可以使用一個私有的建構函式和一個公共的工廠方法(Factory Method),從而避免不正確的構造過程。

public class SafeListener { private final EventListener listener; private SafeListener() { listener = new EventListener() { public void onEvent(Event e) { doSomething(e); } }; } public static SafeListener newInstance(EventSource source) { SafeListener safe = new SafeListener(); source.registerListener(safe.listener); return safe; }} 3.3 執行緒封閉
  1. 執行緒封閉技術的另一種常見應用是JDBC(Java Database Connectivity)的Connection物件。JDBC規範並不要求Connection物件必須是執行緒安全的。在典型的伺服器應用程式中,執行緒從連線池中獲得一個Connection物件,並且用該物件來處理請求,使用完後再將物件返還給連線池。由於大多數請求(例如Servlet請求或EJB呼叫等)都是由單個執行緒採用同步的方式來處理,並且在Connection物件返回之前,連線池不會再將它分配給其他執行緒,因此,這種連線管理模式在處理請求時隱含地將Connection物件封閉線上程中。
  2. Java語言及其核心庫提供了一些機制來幫助維持執行緒封閉性,例如區域性變數和ThreadLocal類,但即便如此,程式設計師仍然需要負責確保封閉線上程中的物件不會從執行緒中逸出。
  3. 在volatile變數上存在一種特殊的執行緒封閉。只要你能確保只有單個執行緒對共享的volatile變數執行寫入操作,那麼就可以安全地在這些共享的volatile變數上執行“讀取-修改-寫入”的操作。在這種情況下,相當於將修改操作封閉在單個執行緒中以防止發生競態條件,並且volatile變數的可見性保證還確保了其他執行緒能看到最新的值。
  4. Ad-hoc執行緒封閉是非常脆弱的,因為沒有任何一種語言特性,例如可見性修飾符或區域性變數,能將物件封閉到目標執行緒中。事實上,對執行緒封閉物件的引用通常儲存在公有變數中。由於Ad-hoc執行緒封閉技術的脆弱性,因此在程式中儘量少用它,在可能的情況下,應該使用更強的執行緒封閉技術(例如,棧封閉或ThreadLocal類)。
  5. 區域性變數的固有屬性之一就是封閉在執行執行緒中。它們位於執行執行緒的棧中,其他執行緒無法訪問這個棧。
  6. 由於任何方法都無法獲得對基本型別的引用,因此Java語言的這種語義就確保了基本型別的區域性變數始終封閉線上程內。
  7. 然而,要小心的是,只有編寫程式碼的開發人員才知道哪些物件需要被封閉到執行執行緒中,以及被封閉的物件是否是執行緒安全的。如果沒有明確地說明這些需求,那麼後續的維護人員很容易錯誤地使物件逸出。
3.3.3 ThreadLocal類
  1. 維持執行緒封閉性的一種更規範方法是使用ThreadLocal,這個類能使執行緒中的某個值與儲存值得物件關聯起來。ThreadLocal提供了get和set等訪問介面或方法,這些方法為每個使用該變數的執行緒都存有一份獨立的副本,因此get總是返回由當前執行執行緒在呼叫set時設定的最新值。
  2. 當某個頻繁執行的操作需要一個臨時物件,例如一個緩衝區,而同時又希望避免在每次執行時都重新分配該臨時物件,就可以使用這項技術。
  3. 從概念上看,你可以將ThreadLocal<T>視為包含了Map<Thread, T>物件,其中儲存了特定於該執行緒的值,但ThreadLocal的實現並非如此。這些特定於執行緒的值儲存在Thread物件中,當執行緒終止後,這些值會作為垃圾回收。
  4. 假設你需要將一個單執行緒應用程式移植到多執行緒環境中,通過將共享的全域性變數轉換為ThreadLocal物件(如果全域性變數的語義允許),可以維持執行緒安全性。然而,如果將應用程式範圍內的快取轉換為執行緒區域性的快取,就不會有太大作用。
  5. 在實現應用程式框架時大量使用了ThreadLocal。例如,在EJB呼叫期間,J2EE容器需要將一個事務上下文(Transaction Context)與某個執行中的執行緒關聯起來。通過將事物上下文儲存在靜態的ThreadLocal物件中,可以很容易地實現這個功能:當框架程式碼需要判斷當前執行的是哪一個事務時,只需從這個ThreadLocal物件中讀取事務上下文。這種機制很方便,因為它避免了在呼叫每個方法時都要傳遞執行上下文資訊,然而這也將使用該機制的程式碼與框架耦合在一起。
  6. 開發人員經常濫用ThreadLocal,例如將所有全域性變數都作為ThreadLocal物件,或者作為一種“隱藏”方法引數的手段。ThreadLocal變數類似於全域性變數,它能降低程式碼的可重用性,並在類之間引入隱含的耦合性,因此在使用時要格外小心。
3.4 不變性
  1. 雖然在Java語言規範和Java記憶體模型中都沒有給出不可變性的正式定義,但不可變性並不等於將物件中所有的域都宣告為final型別,即使物件中所有的域都是final型別的,這個物件也仍然是可變的,因為在final型別的域中可以儲存對可變物件的引用。
  2. 當滿足以下條件時,物件才是不可變的:
    物件建立以後其狀態就不能修改。
    物件的所有域都是final型別。
    物件是正確建立的(在物件的建立期間,this引用沒有逸出)。
  3. 儘管儲存姓名的Set物件是可變的,但從ThreeStooges的設計中可以看到,在Set物件構造完成後無法對其進行修改。stooges是一個final型別的引用變數,因此所有的物件狀態都通過一個final域來訪問。最後一個要求是“正確地構造物件”,這個要求很容易滿足,因為建構函式能使該引用由除了建構函式及其呼叫者之外的程式碼來訪問。
  4. 正如“除非需要更高的可見性,否則應該將所有的域都宣告為私有域”是一個良好的程式設計習慣,“除非需要某個域是可變的,否則應將其宣告為final域”也是一個良好的程式設計習慣。
  5. 每當需要對一組相關資料以原子方式執行某個操作時,就可以考慮建立一個不可變的類來包含這些資料。
  6. 對於訪問和更新多個相關變數時出現的競爭條件問題,可以通過將這些變數全部儲存在一個不可變物件中來消除。
  7. 如果要更新這些變數,那麼可以建立一個新的容器物件,但其他使用原有物件的執行緒仍然會看到物件處於一致的狀態。
  8. 當一個執行緒將volatile型別的cache設定為引用一個新的OneValueCache時,其他執行緒就會立刻看到新快取的資料。
  9. 通過使用包含多個狀態變數的容器物件來維持不變性條件,並使用一個volatile型別的引用來確保可見性,使得Volatile Cached Factorizer在沒有顯式地使用鎖的情況下仍然是執行緒安全的。
3.5 安全釋出
  1. 由於不可變物件是一種非常重要的物件,因此Java記憶體模型為不可變物件的共享提供了一種特殊的初始化安全性保證。
  2. 為了維持這種初始化安全性的保證,必須滿足不可變性的所有需求:狀態不可修改,所有域都是final型別,以及正確的構造過程。
  3. 任何執行緒都可以在不需要額外同步的情況下安全地訪問不可變物件,即使在釋出這些物件時沒有使用同步。
  4. 要安全地釋出一個物件,物件的引用以及物件的狀態必須同時對其他執行緒可見。一個正確構造的物件可以通過以下方式來安全地釋出:
    在靜態初始化函式中初始化一個物件引用。
    將物件的引用儲存到volatile型別的域或者AtomicReference物件中。
    將物件的引用儲存到某個正確構造物件的final型別域中。
    將物件的引用儲存到一個由鎖保護的域中。(線上程安全容器內部的同步意味著,在將物件放入到某個容器,例如Vector或synchronizedList時,將滿足這最後一條):
    通過將一個鍵或者值放入HashTable、synchronizedMap或者ConcurrentMap中,可以安全地將它釋出給任何從這些容器中訪問它的執行緒(無論是直接訪問還是通過迭代器訪問)。
    通過將某個元素放入Vector、CopyOnWriteArrayList、CopyOnWriteArraySet、synchronizedList或synchronizedSet中,可以將該元素安全地釋出到任何從這些容器中訪問該元素的執行緒。
    通過將某個元素放入BlockingQueue或者ConcurrentLinkedQueue中,可以將該元素安全地釋出到任何從這些佇列中訪問該元素的執行緒。
  5. 通常,要釋出一個靜態構造的物件,最簡單和最安全的方式是使用靜態的初始化器:
public static Holder holder = new Holder(42); 第4章 物件的組合 4.1 設計執行緒安全的類
  1. 在設計執行緒安全類的過程中,需要包含以下三個基本要素:
    找出構成物件狀態的所有變數。
    找出約束狀態變數的不變性條件。
    建立物件狀態的併發訪問管理策略。
  2. 如果在物件的域中引用了其他物件,那麼該物件的狀態將包含被引用物件的域。例如,LinkedList的狀態就包括該連結串列中所有節點物件的狀態。
  3. 同步策略規定了如何將不可變性、執行緒封閉與加鎖機制等結合起來以維護執行緒的安全性,並且還規定了哪些變數由哪些鎖來保護。要確保開發人員可以對這個類進行分析與維護,就必須將同步策略寫為正式文件。
  4. 同樣,在操作中還會包含一些後驗條件來判斷狀態遷移是否是有效的。如果Counter的當前狀態為17,那麼下一個有效狀態只能是18。當下一個狀態需要依賴當前狀態時,這個操作就必須是一個複合操作。
  5. 要想實現某個等待先驗條件為真時才執行的操作,一種更簡單的方法是通過現有庫中的類(例如阻塞佇列[Blocking Queue]或訊號量[Semaphore])來實現依賴狀態的行為。
  6. 如果分配並填充了一個HashMap物件,那麼就相當於建立了多個物件:HashMap物件,在HashMap物件中包含的多個物件,以及在Map.Entry中可能包含的內部物件。HashMap物件的邏輯狀態包括所有的Map.Entry物件以及內部物件,即使這些物件都是一些獨立的物件。
  7. 狀態變數的所有者將決定採用何種加鎖協議來維持變數狀態的完整性。
  8. 然而,如果釋出了某個可變物件的引用,那麼就不再擁有獨佔的控制權,最多是“共享控制權”。對於從建構函式或者從方法中傳遞進來的物件,類通常並不擁有這些物件,除非這些方法是被專門設計為轉移傳遞進來的物件的所有權(例如,同步容器封裝器的工廠方法)。
  9. 容器類通常表現出一種“所有權分離”的形式,其中容器類擁有其自身的狀態,而客戶程式碼則擁有容器中各個物件的狀態。Servlet框架中的ServletContext就是其中一個示例。ServletContext為Servlet提供了類似於Map形式的物件容器服務,在ServletContext中可以通過名稱來註冊(setAttribute)或獲取(getAttribute)應用程式物件。
4.2 例項封閉
  1. 當一個物件被封裝到另一個物件中時,能夠訪問被封裝物件的所有程式碼路徑都是已知的。與物件可以由整個程式訪問的情況相比,更易於對程式碼進行分析。通過將封閉機制與合適的加鎖策略結合起來,可以確保以執行緒安全的方式來使用非執行緒安全的物件。
  2. 將資料封裝在物件內部,可以將資料的訪問限制在物件的方法上,從而更容易確保執行緒在訪問資料時總能持有正確的鎖。
  3. 物件可以封閉在類的一個例項(例如作為類的一個私有成員)中,或者封閉在某個作用域內(例如作為一個區域性變數),再或者封閉線上程內(例如在某個執行緒中獎物件從一個方法傳遞到另一個方法,而不是在多個執行緒之間共享該物件)。【當然,物件本身不會逸出——出現逸出情況的原因通常是由於開發人員在釋出物件時超出了物件既定的作用域】。
  4. PersonSet的狀態由HashSet來管理的,而HashSet並非執行緒安全的。但由於mySet是私有的並且不會逸出,因此HashSet被封閉在PersonSet中。唯一能訪問mySet的程式碼路徑是addPerson與containsPerson,在執行它們時都要獲得PersonSet上的鎖。PersonSet的狀態完全由它的內建鎖保護,因而PersonSet是一個執行緒安全的類。
  5. 一些基本的容器類並非執行緒安全的,例如ArrayList和HashMap,但類庫提供了包裝器工廠方法(例如Collections.synchronizedList及其類似方法),使得這些非執行緒安全的類可以在多執行緒環境中安全地使用。這些工廠方法通過“裝飾器(Decorator)”模式將容器類封裝在一個同步的包裝器物件中,而包裝器將介面中的每個方法都實現為同步方法,並將呼叫請求轉發到底層的容器物件上。只要包裝器物件擁有對底層物件的唯一引用(即把底層容器物件封閉在包裝器中),那麼它就是執行緒安全的。在這些方法的Javadoc中指出,對底層容器物件的所有訪問必須通過包裝器來進行。
  6. 由於deepCopy是從一個synchronized方法中呼叫的,因此在執行時間較長的複製操作中,tracker的內建鎖將一直被佔有,當有大量車輛需要追蹤時,會嚴重降低使用者介面的響應靈敏度。
4.3 執行緒安全性的委託 4.3.1 示例:基於委託的車輛追蹤器
  1. 如果使用最初的MutablePoint類而不是Point類,就會破壞封裝性,因為getLocations會發佈一個指向可變狀態的引用,而這個引用不是執行緒安全的。
  2. 在使用監視器模式的車輛追蹤器中返回的是車輛位置的快照,而在使用委託的車輛追蹤器中返回的是一個不可修改但卻實時的車輛位置檢視。這意味著,如果執行緒A呼叫getLocations,而執行緒B在隨後修改了某些點的位置,那麼在返回給執行緒A的Map中將反映出這些變化。
  3. 如果需要一個不發生變化的車輛檢視,那麼getLocations可以返回對locations這個Map物件的一個淺拷貝(Shallow Copy)。由於Map的內容是不可變的,因此只需複製Map的結構,而不用複製它的內容。
public Map<String, Point> getLocations() { return Collections.unmodifiableMap(new HashMap<String, Point>(locations));} 4.3.2 獨立的狀態變數
  1. 我們還可以將執行緒安全性委託給多個狀態變數,只要這些變數是彼此獨立的,即組合而成的類並不會在其包含的多個狀態變數上增加任何不變性條件。
  2. VisualComponent使用CopyOnWriteArrayList來儲存各個監聽器列表。它是一個執行緒安全的連結串列,特別適用於管理監聽器列表。每個連結串列都是執行緒安全的,此外,由於各個狀態之間不存在耦合關係,因此VisualComponent可以將它的執行緒安全性委託給mouseListeners和keyListeners等物件。
4.3.3 當委託失效時
  1. setLower和setUpper都是“先檢查後執行”的操作,但它們沒有使用足夠的加鎖機制來保證這些操作的原子性。
  2. 因此,雖然AtomicInteger是執行緒安全的,但經過組合得到的類卻不是。由於狀態變數lower和upper不是彼此獨立的,因此NumberRange不能將執行緒安全性委託給它的執行緒安全狀態變數。
  3. 如果一個類是由多個獨立且執行緒安全的狀態變數組成,並且在所有的操作中都不包括無效狀態轉換,那麼可以將執行緒安全性委託給底層的狀態變數。
4.3.4 釋出底層的狀態變數
  1. 當把執行緒安全性委託給某個物件的底層狀態變數時,在什麼條件下才可以釋出這些變數從而使其他類能修改它們?答案仍然取決於在類中對這些變數施加了哪些不變性條件。
  2. 如果一個狀態變數是執行緒安全的,並且沒有任何不變性條件來約束它的值,在變數的操作上也不存在任何不允許的狀態轉換,那麼就可以安全地釋出這個變數。
  3. 例如,釋出VisualComponent中的mouseListeners或keyListeners等變數就是安全的。由於VisualComponent並沒有在其監聽器連結串列的合法狀態上施加任何約束,因此這些域可以宣告為公有域或者釋出,而不會破壞執行緒安全性。
4.3.5 示例:釋出狀態的車輛追蹤器
  1. PublishingVehicleTracker將其執行緒安全性委託給底層的ConcurrentHashMap,只是Map中的元素是執行緒安全的且可變的Point,而並非不可變的。getLocation方法返回底層Map物件的一個不可變副本。呼叫者不能增加或刪除車輛,但卻可以通過修改返回Map中的SafePoint值來改變車輛的位置。
  2. PublishingVehicleTracker是執行緒安全的,但如果它在車輛位置的有效值上施加了任何約束,那麼就不再是執行緒安全的。如果需要對車輛位置的變化進行·判斷或者當位置變化時執行一些操作,那麼PublishingVehicleTracker中採用的方法並不合適。
4.4 在現有的執行緒安全類中新增功能
  1. “擴充套件”方法比直接將程式碼新增到類中更加脆弱,因為現在的同步策略實現被分佈到多個單獨維護的原始碼檔案中。如果底層的類改變了同步策略並選擇了不同的鎖來保護它的狀態變數,那麼子類會被破壞,因為在同步策略改變後它無法再使用正確的鎖來控制對基類狀態的併發訪問。
  2. 問題在於在錯誤的鎖上進行了同步。無論List使用哪一個鎖來保護它的狀態,可以確定的是,這個鎖並不是ListHelper上的鎖。ListHelper只是帶來了同步的假象,儘管所有的連結串列操作都被宣告為synchronized,但卻使用了不同的鎖,這意味著putIfAbsent相當於List的其他操作來說並不是原子的,因此就無法確保當putIfAbsent執行時另一個執行緒不會修改連結串列。
  3. 要想使這個方法能正確執行,必須使List在實現客戶端加鎖或外部加鎖時使用同一個鎖。
  4. 然而,客戶端加鎖卻更加脆弱,因為它將類C的加鎖程式碼放到與C完全無關的其他類中。當在那些並不承諾遵循加鎖策略的類上使用客戶端加鎖時,要特別小心。
  5. 客戶端加鎖機制與擴充套件類機制有許多共同點,二者都是將派生類的行為與基類的實現耦合在一起。正如擴充套件會破壞實現的封裝性,客戶端加鎖同樣會破壞同步策略的封裝性。
  6. 當為現有的類新增一個原子操作時,有一種更好的方法:組合(Composition)。ImprovedList通過將List物件的操作委託給底層的List例項來實現List的操作,同時還添加了一個原子的putIfAbsent方法。(與Collections.synchronizedList和其他容器封裝器一樣,ImprovedList假設把某個連結串列物件傳給建構函式以後,客戶程式碼不會再直接使用這個物件,而只能通過ImprovedList來訪問它。)ImprovedList通過自身的內建鎖增加了一層額外的加鎖。它並不關心底層的List是否是執行緒安全的,即使List不是執行緒安全的或者修改了它的加鎖實現,ImprovedList也會提供一致的加鎖機制來實現執行緒安全性。
public class ImprovedList<T> implements List<T> { private final List<T> list; public ImprovedList(List<T> list) { this.list = list; } public synchronized boolean putIfAbsent(T x) { boolean contains = list.contains(x); if (contains) list.add(x); return !contains; } public synchronized void clear() { list.clear(); } // ...按照類似的方式委託List的其他方法}

事實上,我們使用了Java監視器模式來封裝現有的List,並且只要在類中擁有指向底層List的唯一外部引用,就能確保執行緒安全性。

4.5 將同步策略文件化
  1. 在設計同步策略時需要考慮多個方面,例如,將哪些變數宣告為volatile型別,哪些變數用鎖來保護,哪些鎖保護哪些變數,哪些變數必須是不可變的或者被封閉線上程中,哪些操作必須是原子操作等。
  2. 如果使用鎖來保護狀態,那麼也要將其寫入文件以便日後維護,這很簡單,只需使用標註@GuardedBy即可。
  3. 更糟糕的是,我們的直覺通常是錯誤的:我們認為“可能是執行緒安全”的類通常並不是執行緒安全的。例如,java.text.SimpleDateFormat並不是執行緒安全的,但JDK 1.4之前的Javadoc並沒有提到這點。許多開發人員都對這個類不是執行緒安全的而感到驚訝。
  4. 如果某個類沒有明確地宣告是執行緒安全的,那麼就不要假設它是執行緒安全的,從而有效地避免類似於SimpleDateFormat的問題。而另一方面,如果不對容器提供物件(例如HttpSession)的執行緒安全性做某種有問題的假設,也就不可能開發出一個基於Servlet的應用程式。不要使你的客戶或同事也做這樣的猜測。
  5. 許多Java技術規範都沒有(或者至少不願意)說明介面的執行緒安全性,例如ServletContext、HttpSession或DataSource。這些介面是由容器或資料庫供應商來實現的,而你通常無法通過檢視其實現程式碼來了解細節功能。此外,你也不希望依賴於某個特定JDBC驅動的實現細節——你希望遵從標準,這樣程式碼可以基於任何一個JDBC驅動工作。
第5章 基礎構建模組 5.1 同步容器類
  1. 同步容器類都是執行緒安全的,但是在某些情況下可能需要額外的客戶端加鎖來保護複合操作。容器上常見的複合操作包括:迭代(反覆訪問元素,直到遍歷完容器中所有元素)、跳轉(根據指定順序找到當前元素的下一個元素)以及條件運算,例如“若沒有則新增”(檢查在Map中是否存在鍵值K,如果沒有,就加入二元組(K,V))。
  2. 在設計同步容器類的迭代器時並沒有考慮到併發修改的問題,並且它們表現出的行為是“及時失敗”(fail-fast)的。這意味著,當它們發現容器在迭代過程中被修改時,就會丟擲一個ConcurrentModificationException異常。
  3. 這種“及時失敗”的迭代器並不是一種完備的處理機制,而只是“善意地”捕獲併發錯誤,因此只能作為併發問題的預警指示器。它們採用的實現方式是,將計數器的變化與容器關聯起來:如果在迭代期間計數器被修改,那麼hasNext或next將丟擲ConcurrentModificationException。然而,這種檢查是在沒有同步的情況下進行的,因此可能會看到失效的計數值,而迭代器可能並沒有意識到已經發生了修改。
  4. 如果不希望在迭代期間對容器加鎖,那麼一種替代方法就是“克隆”容器,並在副本上進行迭代。由於副本被封閉線上程內,因此其他執行緒不會在迭代期間對其進行修改,這樣就避免了丟擲ConcurrentModificationException(在克隆過程中仍然需要對容器加鎖)。在克隆容器時存在顯著的效能開銷。
  5. 雖然加鎖可以防止迭代器丟擲ConcurrentModificationException,但你必須要記住在所有對共享容器進行迭代的地方都需要加鎖。實際情況要更加複雜,因為在某些情況下,迭代器會隱藏起來。
  6. 編譯器將字串的連線操作轉換成呼叫StringBuilder.append(Object),而這個方法又會呼叫容器的toString方法,標準容器的toString方法將迭代容器,並在每個元素上呼叫toString來生成容器內容的格式化表示。
  7. 容器的hashCode和equals等方法也會間接地執行迭代操作,當容器作為另一個容器的元素或鍵值時,就會出現這種情況。同樣,containsAll、removeAll和retainAll等方法,以及把容器作為引數的建構函式,都會對容器進行迭代。所有這些間接的迭代操作都可能丟擲ConcurrentModificationException。
5.2 併發容器
  1. 在Java 5.0中增加了ConcurrentHashMap,用來替代同步且基於雜湊的Map,以及CopyOnWriteArrayList,用於在遍歷操作為主要操作的情況下代替同步的List。在新的ConcurrentMap介面中增加了對一些常見覆合操作的支援,例如“若沒有則新增”、替換以及有條件刪除等。
  2. BlockingQueue擴充套件了Queue,增加了可阻塞的插入和獲取等操作。如果佇列為空,那麼獲取元素的操作將一直阻塞,直到佇列中出現一個可用的元素。如果佇列已滿(對於有界佇列來說),那麼插入元素的操作將一直阻塞,直到佇列中出現可用的空間。
  3. ConcurrentHashMap並不是將每個方法都在同一個鎖上同步並使得每次只能有一個執行緒訪問容器,而是使用一種粒度更細的加鎖機制來實現更大程度的共享,這種機制稱為分段鎖(Lock Striping)。
  4. 對於一些需要在整個Map上進行計算的方法,例如size和isEmpty,這些方法的語義被略微減弱了以反映容器的併發特性。由於size返回的結果在計算時可能已經過期了,它實際上只是一個估計值,因此允許size返回一個近似值而不是一個精確值。雖然這看上去有些令人不安,但事實上size和isEmpty這樣的方法在併發環境下的用處很小,因為它們的返回值總是在不斷變化。因此,這些操作的需求被弱化了,以換取對其他更重要操作的效能優化,包括get、put、containsKey和remove等。
  5. 由於ConcurrentHashMap不能被加鎖來執行獨佔訪問,因此我們無法使用客戶端加鎖來建立新的原子操作。但是,一些常見的複合操作,例如“若沒有則新增”,“若相等則移除(Remove-If-Equal)”和“若相等則替換(Replace-If-Equal)”等,都已經實現為原子操作並且在ConcurrentMap的介面中宣告。如果你需要在現有的同步Map中新增這樣的功能,那麼很可能就意味著應該考慮使用ConcurrentMap了。
  6. “寫入時複製(Copy-On-Write)”容器的執行緒安全性在於,只要正確地釋出一個事實不可變的物件,那麼在訪問該物件時就不再需要進一步的同步。在每次修改時,都會建立並重新發佈一個新的容器副本,從而實現可變性。
  7. “寫入時複製”容器返回的迭代器不會丟擲ConcurrentModificationException,並且返回的元素與迭代器建立時的元素完全一致,而不必考慮之後修改操作帶來的影響。
  8. 顯然,每當修改容器時都會複製底層陣列,這需要一定的開銷,特別是當容器的規模較大時。僅當迭代操作遠遠多於修改操作時,才應該使用“寫入時複製”容器。
5.3 阻塞佇列和生產者-消費者模式
  1. 生產者-消費者模式能簡化開發過程,因為它消除了生產者類和消費者類之間的程式碼依賴性,此外,該模式還將生產資料的過程與使用資料的過程解耦開來以簡化工作負載的管理,因為這兩個過程在處理資料的速率上有所不同。
  2. 一種最常見的生產者-消費者設計模式就是執行緒池與工作佇列組合,在Executor任務執行框架中就體現了這種模式。
  3. 如果生產者不能儘快地產生工作項使消費者保持忙碌,那麼消費者就只能一直等待,直到有工作可做。在某些情況下,這種方式是非常合適的(例如,在伺服器應用程式中,沒有任何客戶請求服務),而在其他一些情況下,這也表示需要調整生產者執行緒數量和消費者執行緒數量之間的比率,從而實現更高的資源利用率(例如,在“網頁爬蟲[Web Crawler]”或其他應用程式中,有無窮的工作需要完成)。
  4. 正如其他有序的容器一樣,PriorityBlockingQueue既可以根據元素的自然順序來比較元素(如果它們實現了Comparable方法),也可以使用Comparator來比較。
  5. 因為SynchronousQueue沒有儲存功能,因此put和take會一直阻塞,直到有另一個執行緒已經準備好參與到交付過程中。僅當有足夠多的消費者,並且總是有一個消費者準備好獲取交付的工作時,才適合使用同步佇列。
  6. 雖然這個示例使用了顯式管理的執行緒,但許多生產者-消費者設計也可以通過Executor任務執行框架來實現,其本身也使用了生產者-消費者模式。
  7. 對於可變物件,生產者-消費者這種設計與阻塞佇列一起,促進了序列執行緒封閉,從而將物件所有權從生產者交付給消費者。
  8. 在轉移所有權後,也只有另一個執行緒能獲得這個物件的訪問許可權,並且釋出物件的執行緒不會再訪問它。這種安全的釋出確保了物件狀態對於新的所有者來說是可見的,並且由於最初的所有者不會再訪問它,因此物件將被封閉在新的執行緒中。新的所有者執行緒可以對該物件做任意修改,因為它具有獨佔的訪問權。
  9. 只要物件池包含足夠的內部同步來安全地釋出池中的物件,並且只要客戶程式碼本身不會發布池中的物件,或者在將物件返回給物件池後就不再使用它,那麼就可以安全地線上程之間傳遞所有權。
  10. 正如阻塞佇列適用於生產者-消費者模式,雙端佇列同樣適用於另一種相關模式,即工作密取(Work Stealing)。在生產者-消費者設計中,所有消費者有一個共享的工作佇列,而在工作密取設計中,每個消費者都有各自的雙端佇列。如果一個消費者完成了自己雙端佇列中的全部工作,那麼它可以從其他消費者雙端佇列末尾祕密地獲取工作。
  11. 在大多數時候,它們都只是訪問自己的雙端佇列,從而極大地減少了競爭。當工作者執行緒需要訪問另一個佇列時,它會從佇列的尾部而不是從頭部獲取工作,因此進一步降低了佇列上的競爭程度。
  12. 工作密取非常適用於既是消費者也是生產者問題——當執行某個工作時可能導致出現更多的工作。例如,在網頁爬蟲程式中處理一個頁面時,通常會發現有更多的頁面需要處理。類似的還有許多搜尋圖的演算法,例如在垃圾回收階段對堆進行標記,都可以通過工作密取機制來實現高校並行。當一個工作執行緒找到新的任務單元時,它會將其放到自己佇列的末尾(或者在工作共享設計模式中,放入其他工作者執行緒的佇列中)。當雙端佇列為空時,它會在另一個執行緒的佇列隊尾查詢新的任務,從而確保每個執行緒都保持忙碌狀態。
5.4 阻塞方法與中斷方法
  1. 傳遞InterruptedException。避開這個異常通常是最明智的策略——只需把InterruptedException傳遞給方法的呼叫者。傳遞InterruptedException的方法包括,根本不捕獲該異常,或者捕獲該異常,然後在執行某種簡單的清理工作後再次丟擲這個異常。
  2. 恢復中斷。有時候不能丟擲InterruptedException,例如當代碼是Runnable的一部分時。在這些情況下,必須捕獲InterruptedException,並通過呼叫當前執行緒上的interrupt方法恢復中斷狀態,這樣在呼叫棧中更高層的程式碼將看到引發了一箇中斷。
  3. 當在程式碼中呼叫了一個將丟擲InterruptedException異常的方法時,你自己的方法也就變成了一個阻塞方法,並且必須要處理對中斷的響應。
  4. 然而在出現InterruptedException時不應該做的事情是,捕獲它但不做出任何響應。這將使呼叫棧上更高層的程式碼無法對中斷採取處理措施,因為執行緒被中斷的證據已經丟失。
  5. 當某方法丟擲InterruptedException時,表示該方法是一個阻塞方法,如果這個方法被中斷,那麼它將努力提前結束阻塞狀態。
5.5 同步工具類
  1. 同步工具類可以是任何一個物件,只要它根據自身的狀態來協調執行緒的控制流。阻塞佇列可以作為同步工具類,其他型別的同步工具類還包括訊號量(Semaphore)、柵欄(Barrier)以及閉鎖(Latch)。
  2. 所有的同步工具類都包含一些特定的結構化屬性:它們封裝了一些狀態,這些狀態將決定執行同步工具類的執行緒是繼續執行還是等待,此外還提供了一些方法對狀態進行操作,以及另一些方法用於高校地等待同步工具類進入到預期狀態。
5.5.1 閉鎖
  1. 二元閉鎖(包括兩個狀態)可以用來表示“資源R已經被初始化”,而所有需要R的操作都必須先在這個閉鎖上等待。
  2. 確保某個服務在其依賴的所有其他服務都已經啟動之後才啟動。每個服務都有一個相關的二元閉鎖。當啟動服務S時,將首先在S依賴的其他服務的閉鎖上等待,在所有依賴的服務都啟動後會釋放閉鎖S,這樣其他依賴S的服務才能繼續執行。
  3. 閉鎖狀態包括一個計數器,該計數器被初始化為一個正數,表示需要等待的事件數量。countDown方法遞減計數器,表示有一個事件已經發生了,而await方法等待計數器達到零,這表示所有需要等待的事件都已經發生。如果計數器的值非零,那麼await會一直阻塞直到計數器為零,或者等待中的執行緒中斷,或者等待超時。
  4. TestHarness建立一定數量的執行緒,利用它們併發地執行指定的任務。它使用兩個閉鎖,分別表示“起始門(Starting Gate)”和“結束門(Ending Gate)”。起始門計數器的初始值為1,而結束門計數器的初始值為工作執行緒的數量。每個工作執行緒首先要做的事就是在啟動門上等待,從而確保所有執行緒都就緒後才開始執行。而每個執行緒要做的最後一件事情是將呼叫結束門的countDown方法減1,這能使主執行緒高效地等待直到所有工作執行緒都執行完成,因此可以統計所消耗的時間。
public class TestHarness { public long timeTasks(int nThreads, final Runnable task) throws InterruptedException { final CountDownLatch startGate = new CountDownLatch(1); final CountDownLatch endGate = new CountDownLatch(nThreads); for (int i = 0; i < nThreads; i++) { Thread t = new Thread() { public void run() { try { startGate.await(); try { task.run(); } finally { endGate.countDown(); } } catch (InterruptedException ignored) {} } }; t.start(); } long start = System.nanoTime(); startGate.countDown(); endGate.await(); long end = System.nanoTime(); return end-start; }}

所有執行緒都會在await()這裡阻塞。nThreads個執行緒都start了之後,全部在startGate.await()這裡阻塞。直到主執行緒執行完迴圈體,接著執行startGate.countDown()的,這nThreads個執行緒才通過開始門,然後開始執行task。此時,主執行緒在endGate.await()這裡阻塞。
5. 為什麼要在TestHarness中使用閉鎖,而不是線上程建立後就立即啟動?或許,我們希望測試n個執行緒併發執行某個任務時需要的時間。如果在建立執行緒後立即啟動它們,那麼先啟動的執行緒將“領先”後啟動的執行緒,並且活躍執行緒數量會隨著時間的推移而增加或減少,競爭程度也在不斷髮生變化。啟動門將使得主執行緒能夠同時釋放所有的工作執行緒,而結束門則使主執行緒能夠等待最後一個執行緒執行完成,而不是順序地等待每個執行緒執行完成。

5.5.2 FutureTask
  1. FutureTask在Executor框架中表示非同步任務,此外還可以用來表示一些時間較長的計算,這些計算可以在使用計算結果之前啟動。
public class Preloader { private final FutureTask<ProductInfo> future = new FutureTask<ProductInfo>(new Callable<ProductInfo>() { public ProductInfo call() throws DataLoadException { return loadProductInfo(); }}; private final Thread thread = new Thread(future); public void start() { thread.start(); } //通過提前呼叫start()啟動FutureTask的Callable執行緒執行loadProductInfo()這個耗時的任務。不要等待需要ProductInfo時再呼叫該任務。 public ProductInfo get() throws DataLoadException, InterruptedException { try { return future.get(); } catch (ExecutionException e) { Throwable cause = e.getCause(); if (cause instanceof DataLoadException) throw (DataLoadException) cause; else throw launderThrowable(cause); } }}

Preloader建立了一個FutureTask,其中包含從資料庫載入產品資訊的任務,以及一個執行運算的執行緒。由於在建構函式或靜態初始化方法中啟動執行緒並不是一種好方法,因此提供了一個start方法來啟動執行緒。當程式隨後需要ProductInfo時,可以呼叫get方法,如果資料已經載入,那麼將返回這些資料,否則將等待載入完後再返回。

5.5.3 訊號量
  1. 在這種實現中不包含真正的許可物件,並且Semaphore也不會將許可與執行緒關聯起來,因此在一個執行緒中獲得的許可可以在另一個執行緒中釋放。可以將acquire操作視為是消費一個許可,而release操作是建立一個許可,Semaphore並不受限於它在建立時的初始許可數量。
  2. 我們可以構造一個固定長度的資源池,當池為空時,請求資源將會失敗,但你真正希望看到的行為是阻塞而不是失敗,並且當池非空時解除阻塞。
  3. 底層的Set實現並不知道關於邊界的任何資訊,這是由BoundedHashSet來處理的。
5.5.4 柵欄
  1. 閉鎖用於等待事件,而柵欄用於等待其他執行緒。
  2. CyclicBarrier可以使一定數量的參與方反覆地在柵欄位置彙集,它在並行迭代演算法中非常有用:這種演算法通常將一個問題拆分成一系列相互獨立的子問題。當執行緒到達柵欄位置時將呼叫await方法,這個方法將阻塞直到所有執行緒都到達柵欄位置。
  3. 如果對await的呼叫超時,或者await阻塞的執行緒被中斷,那麼柵欄就被認為是打破了,所有阻塞的await呼叫都將終止並丟擲BrokenBarrierException。如果成功地通過柵欄,那麼await將為每個執行緒返回一個唯一的到達索引號,我們可以利用這些索引來“選舉”產生一個領導執行緒,並在下一次迭代中由該領導執行緒執行一些特殊的工作。CyclicBarrier還可以使你將一個柵欄操作傳遞給建構函式,這是一個Runnable,當成功通過柵欄時會(在一個子任務執行緒中)執行它,但在阻塞執行緒被釋放之前是不能執行的。
  4. 在把模擬過程並行化時,為每個元素(在這個示例中相當於一個細胞)分配一個獨立的執行緒是不現實的,因為這將產生過多的執行緒,而在協調這些執行緒上導致的開銷將降低計算效能。合理的做法是,將問題分解成一定數量的子問題,為每個子問題分配一個執行緒來進行求解,之後再將所有的結果合併起來。CellularAutomata將問題分解為N_cpu個子問題,其中N_cpu等於可用CPU的數量,並將每個子問題分配給一個執行緒。
  5. 當兩方執行不對稱的操作時,Exchanger會非常有用,例如當一個執行緒向緩衝區寫入資料,而另一個執行緒從緩衝區中讀取資料。這些執行緒可以使用Exchanger來匯合,並將滿的緩衝區與空的緩衝區交換。
  6. 資料交換的時機取決於應用程式的響應需求。最簡單的方案是,當緩衝區被填滿時,由填充任務進行交換,當緩衝區為空時,由清空任務進行交換。這樣會把需要交換的次數降至最低,但如果新資料的到達率不可預測,那麼一些資料的處理過程就將延遲。另一個方法是,不僅當緩衝被填滿時進行交換,並且當緩衝被填充到一定程度並保持一定時間後,也進行交換。
5.6 構建高效且可伸縮的結果快取
  1. Memorizer2的問題在於,如果某個執行緒啟動了一個開銷很大的計算,而其他執行緒並不知道這個計算正在進行,那麼很可能會重複這個計算。我們希望通過某種方法來表達“執行緒X正在計算f(27)”這種情況,這樣當另一個執行緒查詢f(27)時,它能夠知道最高效的方法是等待執行緒X計算結束,然後再去查詢快取“f(27)的結果是多少?”
  2. Memorizer3將用於快取值的Map重新定義為ConcurrentHashMap<A,Future<V>>,替換原來的ConcurrentHashMap<A,V>。Memorizer3首先檢查某個相應的計算是否已經開始(Memorizer2與之相反,它首先判斷某個計算是否已經完成)。如果還沒有啟動,那麼就建立一個FutureTask,並註冊到Map中,然後啟動計算:如果已經啟動,那麼等待現有計算的結果。結果可能很快會得到,也可能還在運算過程中,但這對於Future.get的呼叫者來說是透明的。
  3. 它只有一個缺陷,即仍然存在兩個執行緒計算出相同值的漏洞。這個漏洞的發生概率要遠小於Memorizer2中發生的概率,但由於compute方法中的if程式碼塊仍然是非原子(nonatomic)的“先檢查再執行”操作,因此兩個執行緒仍有可能在同一時間內呼叫compute來計算相同的值,即二者都沒有在快取中找到期望的值,因此都開始計算。
  4. Memorizer3中存在這個問題的原因是,複合操作(“若沒有則新增”)是在底層的Map物件上執行的,而這個物件無法通過加鎖來確保原子性。Memorizer使用了ConcurrentMap中的原子方法putIfAbsent,避免了Memorizer3的漏洞。
public class Memorizer<A, V> implements Computable<A, V> { private final ConcurrentMap<A, Future<V>> cache = new ConcurrentHashMap<A, Future<V>>(); private final Computable<A, V> c; public Memorizer(Computable<A, V> c) { this.c = c; } public V compute (final A arg) throws InterruptedException { while (true) { Future<V> f = cache.get(arg); if (f == null) { Callable<V> eval = new Callable<V> () { public V call() throws InterrruptedException { return c.compute(arg); } }; FutureTask<V> ft = new FutureTask<V>(eval); f = cache.putIfAbsent(arg, ft); // Returns: the previous value associated with the specified key, or null if there was no mapping for the key if (f == null) { f = ft; ft.run(); } } try { return f.get(); } catch (CancellationException e) { cache.remove(arg, f); // 5. 當快取的是Future而不是值時,將導致快取汙染(Cache Pollution)問題:如果某個計算被取消或者失敗,那麼在計算這個結果時將指明計算過程被取消或者失敗。為了避免這種情況,如果Memorizer發現計算被取消,那麼將把Future從快取中移除。如果檢測到RuntimeException,那麼也會移除Future,這樣將來的計算才可能成功。 } catch (ExecutionException e) { throw launderThrowable(e.getCause()); }}}}

在因式分解servlet中使用Memorizer來快取結果

public class Factorizer implements Servlet { private final Computable<BigInteger, BigInteger[]> c = new Computable<BigInteger, BigInteger[]> () { public BigInteger[] compute(BigInteger arg) { return factor(arg); }}; private final Computable<BigInteger, BigInteger[]> cache = new Memorizer<BigInteger, BigInteger[]> (c); public void service(ServletRequest req, ServletResponse resp) { try { BigInteger i = extractFromRequest(req); encodeIntoResponse(resp, cache.compute(i)); } catch (InterruptedException e) { encodeError(resp, "factorization interrupted"); } }} 第6章 任務執行 6.1 線上程中執行任務
  1. 當圍繞“任務執行”來設計應用程式結構時,第一步就是要找出清晰的任務邊界。
  2. 而且,當負荷過載時,應用程式的效能應該是逐漸降低,而不是直接失敗。
  3. 大多數伺服器應用程式都提供了一種自然的任務邊界選擇方式:以獨立的客戶請求為邊界。
  4. 任務處理過程從主執行緒分離出來,使得主迴圈能夠更快地重新等待下一個到來的連線。
  5. 如果可執行的執行緒數量多於可用處理器的數量,那麼有些執行緒將閒置。大量空閒的執行緒會佔用許多記憶體,給垃圾回收器帶來壓力,而且大量執行緒在競爭CPU資源時還將產生其他的效能開銷。如果你已經擁有足夠多的執行緒使所有的CPU保持忙碌狀態,那麼再建立更多的執行緒反而會降低效能。
  6. 如果破壞了這些限制,那麼很可能丟擲OutOfMemoryError異常,要想從這種錯誤中恢復過來是非常危險的,更簡單的辦法是通過構造程式來避免超出這些限制。
6.2 Executor框架
  1. 它提供了一種標準的方法將任務的提交過程與執行過程解耦開來,並用Runnable來表示任務。Executor的實現還提供了對生命週期的支援,以及統計資訊收集、應用程式管理機制和效能監視等機制。
  2. Executor基於生產者-消費者模式,提交任務的操作相當於生產者(生產待完成的工作單元),執行任務的執行緒則相當於消費者(執行完這些工作單元)。
  3. 每當看到下面這種形式的程式碼時:
    new Thread(runnable).start()
    並且你希望獲得一種更靈活的執行策略時,請考慮使用Executor來代替Thread。
  4. newFixedThreadPool將建立一個固定長度的執行緒池,每當提交一個任務時就建立一個執行緒,直到達到執行緒池的最大數量,這時執行緒池的規模將不再變化(如果某個執行緒由於發生了未預期的Exception而結束,那麼執行緒池會補充一個新的執行緒)。
  5. newCachedThreadPool將建立一個可快取的執行緒池,如果執行緒池的當前規模超過了處理需求時,那麼將回收空閒的執行緒,而當需求增加時,則可以新增新的執行緒,執行緒池的規模不存在任何限制。
  6. newSingleThreadExecutor是一個單執行緒的Executor,它建立單個工作者執行緒來執行任務,如果這個執行緒異常結束,會建立另一個執行緒來替代。newSingleThreadExecutor能確保依照任務在佇列中的順序來序列執行(例如FIFO、LIFO、優先順序)。
  7. newScheduledThreadPool建立了一個固定長度的執行緒池,而且以延遲或定時的方式來執行任務,類似於Timer。
  8. 但JVM只有在所有(非守護)執行緒全部終止後才會退出。因此,如果無法正確地關閉Executor,那麼JVM將無法結束。
  9. ExecutorService的生命週期有3種狀態:執行、關閉和已終止。
  10. shutdown方法將執行平緩的關閉過程:不再接受新的任務,同時等待已經提交的任務執行完成——包括那些還未開始執行的任務。shutdownNow方法將執行粗暴的關閉過程:它將嘗試取消所有執行中的任務,並且不再啟動佇列中尚未開始執行的任務。
  11. Timer的另一個問題是,如果TimerTask丟擲一個未檢查的異常,那麼Timer將表現出糟糕的行為。Timer執行緒並不捕獲異常,因此當TimerTask丟擲未檢查的異常時將終止定時執行緒。這種情況下,Timer也不會恢復執行緒的執行,而是會錯誤地認為整個Timer都被取消了。因此,已經被排程但尚未執行的TimerTask將不會再執行,新的任務也不能被排程。(這個問題稱為“執行緒洩露”)
  12. ScheduledThreadPoolExecutor能正確處理這些表現出錯誤行為的任務。在Java 5.0或更高的JDK中,將很少使用Timer。
  13. 如果要構建自己的排程服務,那麼可以使用DelayQueue,它實現了BlockingQueue,併為ScheduledThreadPoolExecutor提供排程功能。DelayQueue管理著一組Delayed物件。每個Delayed物件都有一個相應的延遲時間:在DelayQueue中,只有某個元素逾期後,才能從DelayQueue中執行take操作。從DelayQueue中返回的物件將根據它們的延遲時間進行排序。
6.3 找出可利用的並行性
  1. Runnable是一種有很大侷限的抽象,雖然run能寫入到日誌檔案或者將結果放入某個共享的資料結構,但它不能返回一個值或丟擲一個受檢查的異常。
  2. 對於這些任務,Callable是一種更好的抽象:它認為主入口點(即call)將返回一個值,並可能丟擲一個異常。
  3. Executor框架中,已提交但尚未開始的任務可以取消,但對於那些已經開始執行的任務,只有當它們能響應中斷時,才能取消。取消一個已經完成的任務不會有任何影響。
  4. 在Future規範中包含的隱含意義是,任務的生命週期只能前進,不能後退,就像ExecutorService的生命週期一樣。當某個任務完成後,它就永遠停留在“完成”狀態上。
  5. get方法的行為取決於任務的狀態(尚未開始、正在執行、已完成)。如果任務已經完成,那麼get會立即返回或者丟擲一個Exception,如果任務沒有完成,那麼get將阻塞並直到任務完成。如果任務丟擲了異常,那麼get將該異常封裝為ExecutionException並重新丟擲。如果任務被取消,那麼get將丟擲CancellationException。如果get丟擲了ExecutionException,那麼可以通過getCause來獲得被封裝的初始異常。
  6. ExecutorService中的所有submit方法都將返回一個Future,從而將一個Runnable或Callable提交給Executor,並得到一個Future用來獲得任務的執行結果或者取消任務。還可以顯式地為某個指定的Runnable或Callable例項化一個FutureTask。(由於FutureTask實現了Runnable,因此可以將它提交給Executor來執行,或者直接呼叫它的run方法。)
  7. 在將Runnable或Callable提交到Executor的過程中,包含了一個安全釋出過程,即將Runnable或Callable從提交執行緒釋出到最終執行任務的執行緒。類似地,在設定Future結果的過程中也包含了一個安全釋出,即將這個結果從計算它的執行緒釋出到任何通過get獲得它的執行緒。
6.3.3 示例:使用Future實現頁面渲染器
  1. 為了使頁面渲染器實現更高的併發性,首先將渲染過程分解為兩個任務,一個是渲染所有的文字,另一個是下載所有的影象。(因為其中一個任務是CPU密集型,而另一個任務是I/O密集型,因此這種方法即使在單CPU系統上也能提升效能。)
  2. Callable和Future有助於表示這些協同任務之間的互動。FutureRenderer中建立了一個Callable來下載所有的影象,並將其提交到一個ExecutorService。這將返回一個描述任務執行情況的Future。當主任務需要影象時,它會等待Future.get的呼叫結果。如果幸運的話,當開始請求時所有影象就已經下載完成了,即使沒有,至少影象的下載任務也已經提前開始了。
Callable<List<ImageData>> task = new Callable<List<ImageData>>() { public List<ImageData> call() { List<ImageData> result = new ArrayList<ImageData>(); ... return result; }};Future<List<ImageData>> future = executor.submit(task);

執行有返回值的Callable時用submit,它可以將結果包裝成一個Future。執行無返回值的Runnable時,就用execute。

6.3.4 在異構任務並行化中存在的侷限
  1. 兩個人可以很好地分擔洗碗的工作:其中一個人負責清洗,而另一個人負責烘乾。然而,要將不同型別的任務平均分配給每個工人卻並不容易。當人數增加時,如何確保他們能幫忙而不是妨礙其他人工作,或者在重新分配工作時,並不是容易的事情。如果沒有在相似的任務之間找出細粒度的並行性,那麼這種方法帶來的好處將減少。
  2. FutureRenderer使用了兩個任務,其中一個負責渲染文字,另一個負責下載影象。如果渲染文字的速度遠遠高於下載影象的速度(可能性更大),那麼程式的最終效能與序列執行時的效能差別不大,而程式碼卻變得更復雜了。
  3. 只有當大量相互獨立且同構的任務可以併發進行處理時,才能體現出將程式的工作負載分配到多個任務中帶來的真正效能提升。
6.3.5 CompletionService: Executor與BlockingQueue
  1. CompletionService將Executor和BlockingQueue的功能融合在一起。你可以將Callable任務提交給它來執行,然後使用類似於佇列操作的take和poll等方法來獲得已完成的結果,而這些結果會在完成時被封裝為Future。
  2. ExecutorCompletionService的實現非常簡單。在建構函式中建立一個BlockingQueue來儲存計算完成的結果。當計算完成時,呼叫FutureTask中的done方法。當提交某個任務時,該任務將首先包裝為一個QueueingFuture,這是FutureTask的一個子類,然後再改寫子類的done方法,並將結果放入BlockingQueue中。take和poll方法委託給了BlockingQueue,這些方法會在得到結果之前阻塞。
  3. 多個ExecutorCompletionService可以共享一個Executor,因此可以建立一個對於特定計算私有,又能共享一個公共Executor的ExecutorCompletionService。因此,CompletionService的作用就相當於一組計算的控制代碼,這與Future作為單個計算的控制代碼是非常類似的。
  4. 使用CompletionService,使頁面元素在下載完成後立即顯示出來
public class Renderer { private final ExecutorService executor; Renderer(ExecutorService executor) { this.executor = executor; } void renderPage(CharSequence source) { List<ImageInfo> info = scanForImageInfo(source); // java.util.concurrent // Interface CompletionService<V> // All Known Implementing Classes: ExecutorCompletionService CompletionService<ImageData> completionService = new ExecutorCompletionService<ImageData>(executor); for (final ImageInfo imageInfo : info) completionService.submit(new Callable<ImageData>() { public ImageData call() { return imageInfo.downloadImage(); } }); renderText(source); try { for( int i=0, n=info.size(); i<n; i++) { Future<ImageData> f = completionService.take(); ImageData imageData = f.get(); renderImage(imageData); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (ExecutionException e) { throw launderThrowable(e.getCause()); } }} 6.3.7 為任務設定時限
  1. 在支援時間限制的Future.get中支援這種需求:當結果可用時,它將立即返回,如果在指定時限內沒有計算出結果,那麼將丟擲TimeoutException。
  2. 此時可再次使用Future,如果一個限時的get方法丟擲了TimeoutException,那麼可以通過Future來取消任務。
try { long timeLeft = endNanos - System.nanoTime(); ad = f.get(timeLeft, NANOSECONDS);} catch (ExecutionException e) { ad = DEFAULT_AD;} catch (TimeoutException e) { ad = DEFAULT_AD; f.cancel(true);} 6.3.8 示例:旅行預訂入口網站
  1. 建立n個任務,將其提交到一個執行緒池,保留n個Future,並使用限時的get方法通過Future序列地獲取每一個結果,這一切都很簡單,但還有一個更簡單的方法——invokeAll。
  2. InvokeAll方法的引數為一組任務,並返回一組Future。
  3. invokeAll按照任務集合中迭代器的順序將所有的Future新增到返回的集合中,從而使呼叫者能將各個Future與其表示的Callable關聯起來。
  4. 當invokeAll返回後,每個任務要麼正常地完成,要麼被取消,而客戶端程式碼可以呼叫get或isCancelled來判斷究竟是何種情況。
第7章 取消與關閉
  1. Java沒有提供任何機制來安全地終止執行緒。但它提供了中斷(Interruption),這是一種協作機制,能夠使一個執行緒終止另一個執行緒的當前工作。
  2. 這種協作式的方法是必要的,我們很少希望某個任務、執行緒或服務立即停止,因為這種立即停止會使共享的資料結構處於不一致的狀態。相反,在編寫任務和服務時可以使用一種協作的方式:當需要停止時,它們首先會清除當前正在執行的工作,然後再結束。
  3. 一個在行為良好的軟體與勉強執行的軟體之間的最主要區別就是,行為良好的軟體能很完善地處理失敗、關閉和取消等過程。
7.1 任務取消
  1. 如果外部程式碼能在某個操作正常完成之前將其置入“完成”狀態,那麼這個操作就可以稱為可取消的(Cancellable)。
  2. 應用程式事件。例如,應用程式對某個問題空間進行分解並搜尋,從而使不同的任務可以搜尋問題空間中的不同區域。當其中一個任務找到了解決方案時,所有其他仍在搜尋的任務都將被取消。
  3. 錯誤。網頁爬蟲程式搜尋相關的頁面,並將頁面或摘要資料儲存到硬碟。當一個爬蟲任務發生錯誤時(例如,磁碟空間已滿),那麼所有搜尋任務都會取消,此時可能會記錄它們的當前狀態,以便稍後重新啟動。
  4. 在Java中沒有一種安全的搶佔式方法來停止執行緒,因此也就沒有安全的搶佔式方法來停止任務。
  5. cancel方法由finally塊呼叫,從而確保即使在呼叫sleep時被中斷也能取消素數生成器的執行。如果cancel沒有被呼叫,那麼搜尋素數的執行緒將永遠執行下去,不斷消耗CPU的時鐘週期,並使得JVM不能正常退出。
  6. 一個可取消的任務必須擁有取消策略(Cancellation Policy),在這個策略中將詳細地定義取消操作的“How”、“When”以及“What”,即其他程式碼如何(How)請求取消該任務,任務在何時(When)檢查是否已經請求了取消,以及在響應取消請求時應該執行哪些(What)操作。
7.1.1 中斷
  1. 然而,如果使用這種方法的任務呼叫了一個阻塞方法,例如BlockingQueue.put,那麼可能會產生一個更嚴重的問題——任務可能永遠不會檢查取消標誌,因此永遠不會結束。
  2. 當生產者在put方法中阻塞時,如果消費者希望取消生產者任務,那麼將發生什麼情況?它可以呼叫cancel方法來設定cancelled標誌,但此時生產者卻永遠不能檢查這個標誌,因此它無法從阻塞的put方法中恢復過來(因為消費者此時已經停止從佇列中取出素數,所以put方法將一直保持阻塞狀態)。
  3. 在Java的API或語言規範中,並沒有將中斷與任何取消語義關聯起來,但實際上,如果在取消之外的其他操作中使用中斷,那麼都是不合適的,並且很難支撐起更大的應用。
  4. 每個執行緒都有一個boolean型別的中斷狀態。
  5. interrupt方法能中斷目標執行緒,而isInterrupted方法能返回目標執行緒的中斷狀態。靜態的interrupted方法將清除當前執行緒的中斷狀態,並返回它之前的值,這也是清除中斷狀態的唯一方法。
  6. 阻塞庫方法,例如Thread.sleep和Object.wait等,都會檢查執行緒何時中斷,並且在發現中斷時提前返回。它們在響應中斷時執行的操作包括:清除中斷狀態,丟擲InterruptedException,表示阻塞操作由於中斷而提前結束。JVM並不能保證阻塞方法檢測到中斷的速度,但在實際情況中響應速度還是非常快的。
  7. 呼叫interrupt並不意味著立即停止目標執行緒正在進行的工作,而只是傳遞了請求中斷的訊息。
  8. 對中斷操作的正確理解是:它並不會真正地中斷一個正在執行的執行緒,而只是發出中斷請求,然後由執行緒在下一個合適的時刻中斷自己。有些方法,例如wait、sleep和join等,將嚴格地處理這種請求,當它們收到中斷請求或者在開始執行時發現某個已被設定好的中斷狀態時,將丟擲一個異常。
  9. 如果在呼叫interrupted時返回了true,那麼除非你想遮蔽這個中斷,否則必須對它進行處理——可以丟擲InterruptedException,或者通過再次呼叫interrupt來恢復中斷狀態。
  10. 通常,中斷是實現取消的最合理方式。
  11. 在每次迭代迴圈中,有兩個位置可以檢測出中斷:在阻塞的put方法呼叫中,以及在迴圈開始處查詢中斷狀態時。由於呼叫了阻塞的put方法,因此這裡並不一定需要進行顯式的檢測,但執行檢測卻會使PrimeProducer對中斷具有更高的響應性,因為它是在啟動尋找素數任務之前檢查中斷的,而不是在任務完成之後。
7.1.2 中斷策略
  1. 中斷策略規定執行緒如何解釋某個中斷請求——當發現中斷請求時,應該做哪些工作(如果需要的話),哪些工作單元對於中斷來說是原子操作,以及以多快的速度來響應中斷。
  2. 區分任務和執行緒對中斷的反應是很重要的。一箇中斷請求可以有一個或多個接收者——中斷執行緒池中的某個工作執行緒,同時意味著“取消當前任務”和“關閉工作者執行緒”。
  3. 任務不會在其自己擁有的執行緒中執行,而是在某個服務(例如執行緒池)擁有的執行緒中執行。對於非執行緒所有者的程式碼來說(例如,對於執行緒池而言,任何線上程池實現以外的程式碼),應該小心地儲存中斷狀態,這樣擁有執行緒的程式碼才能對中斷做出響應,即使“非所有者”程式碼也可以做出響應。
  4. 這就是為什麼大多數可阻塞的庫函式都只是丟擲InterruptedException作出中斷響應。它們永遠不會在某個由自己擁有的執行緒中執行,因此它們為任務或庫程式碼實現了最合理的取消策略:儘快退出執行流程,並把中斷資訊傳遞給呼叫者,從而使呼叫棧中的上層程式碼可以採取進一步的操作。
  5. 執行緒應該只能由其所有者中斷,所有者可以將執行緒的中斷策略資訊封裝到某個合適的取消機制中,例如關閉(shutdown)方法。
  6. 批評者曾嘲笑Java的中斷功能,因為它沒有提供搶佔式中斷機制,而且還強迫開發人員必須處理InterruptedException。然後,通過推遲中斷請求的處理,開發人員能制定更靈活的中斷策略,從而使應用程式在響應性和健壯性之間實現合理的平衡。
7.1.3 響應中斷
  1. 由於大多數程式碼並不知道它們將在哪個執行緒中執行,因此應該儲存中斷狀態。
  2. 對於一些不支援取消但仍可以呼叫可中斷阻塞方法的操作,它們必須在迴圈中呼叫這些方法,並在發現中斷後重新嘗試。在這種情況下,它們應該在本地儲存中斷狀態,並在返回前恢復狀態而不是在捕獲InterruptedException時恢復狀態。
// 不可取消的任務在退出前恢復中斷public Task getNextTask(BlockingQueue<Taskgt> queue) { boolean interrupted = false; try { while(true) { try { return queue.take(); } catch (InterruptedException e) { interrupted = true; // 重新嘗試 } } } finally { if (interrupted) Thread.currentThread().interrupt(); }}

如果程式碼不會呼叫可中斷的阻塞方法,那麼仍然可以通過在任務程式碼中輪詢當前執行緒的中斷狀態來響應中斷。

7.1.4 示例:計時執行
  1. 如果任務在超時之前完成,那麼中斷timedRun所線上程的取消任務將在timedRun返回到呼叫者之後啟動。我們不知道在這種情況下將執行什麼程式碼,但結果一定是不好的。
7.1.5 通過Future來實現取消// 通過Future來取消任務public static void timedRun(Runnable r, long timeout, TimeUnit unit) throws InterruptedException { Future<?> task = taskExec.submit(r); try { task.get(timeout, unit); } catch (TimeoutException e) { // 接下來任務將被取消 } catch (ExecutionException e) { // 如果在任務中丟擲了異常,那麼重新丟擲該異常 throw launderThrowable(e.getCause()); } finally { // 如果任務已經結束,那麼執行取消操作也不會帶來任何影響 task.cancel(true); // 如果任務正在執行,那麼將被中斷 }} 7.1.6 處理不可中斷的阻塞
  1. 在Java庫中,許多可阻塞的方法都是通過提前返回或者丟擲InterruptedException來響應中斷請求的,從而使開發人員更容易構建出能響應取消請求的任務。
  2. Java.io包中的同步Socket I/O。在伺服器應用程式中,最常見的阻塞I/O形式就是對套接字進行讀取和寫入。雖然InputStream和OutputStream中的read和write等方法都不會響應中斷,但通過關閉底層的套接字,可以使得由於執行read或write等方法而被阻塞的執行緒丟擲一個SocketException。
  3. Java.io包中的同步I/O。當中斷一個正在InterruptibleChannel上等待的執行緒時,將丟擲ClosedByInterruptException並關閉鏈路(這還會使得其他在這條鏈路上阻塞的執行緒同樣丟擲ClosedByInterruptException)。當關閉一個InterruptibleChannel時,將導致所有在鏈路操作上阻塞的執行緒都丟擲AsynchronousCloseException。大多數標準的Channel都實現了InterruptibleChannel。
  4. Selector的非同步I/O。如果一個執行緒在呼叫Selector.select方法(在java.nio.channels中)時阻塞了,那麼呼叫close或wakeup方法會使執行緒丟擲ClosedSelectorException並提前返回。
  5. 獲取某個鎖。如果一個執行緒由於等待某個內建鎖而阻塞,那麼將無法響應中斷,因為執行緒認為它肯定會獲得鎖,所以將不會理會中斷請求。但是,在Lock類中提供了lockInterruptibly方法,該方法允許在等待一個鎖的同時仍能響應中斷。
// 通過改寫interrupt方法將非標準的取消操作封裝在Thread中public class ReaderThread extends Thread { private final Socket socket; private final InputStream in; public ReaderThread(Socket socket) throws IOException { this.socket = socket; this.in = socket.getInputStream(); } public void interrupt() { try { socket.close(); } catch (IOException ignored) { } finally { super.interrupt(); } } public void run() { try { byte[] buf = new byte[BUFSZ]; while(true) { int count = in.read(buf); if(count<0) break; else if(count>0) processBuffer(buf, count); } } catch (IOException e) { /* 允許執行緒退出 */ } }} 7.2 停止基於執行緒的服務
  1. 與其他封裝物件一樣,執行緒的所有權是不可傳遞的:應用程式可以擁有服務,服務也可以擁有工作者執行緒,但應用程式並不能擁有工作者執行緒,因此應用程式不能直接停止工作者執行緒。相反,服務應該提供生命週期方法(Lifecycle Method)來關閉它自己以及它所擁有的執行緒。
7.2.1 示例:日誌服務
  1. 產生日誌訊息的執行緒並不會將訊息直接寫入輸出流,而是由LogWriter通過BlockingQueue將訊息提交給日誌執行緒,並由日誌執行緒寫入。這是一種多生產者單消費者(Multiple-Producer, Single-Consumer)的設計方式:每個呼叫log的操作都相當於一個生產者,而後臺的日誌執行緒則相當於消費者。
  2. 當取消一個生產者-消費者操作時,需要同時取消生產者和消費者。
  3. 為LogWriter提供可靠關閉操作的方法是解決競態條件問題,因而要使日誌訊息的提交操作成為原子操作。然而,我們不希望在訊息加入佇列時去持有一個鎖,因為put方法本身就可以阻塞。我們採用的方法是:通過原子方式來檢查關閉請求,並且有條件地遞增一個計數器來“保持”提交資訊的權利。
// 向LogWriter新增可靠的取消操作public class LogService { private final BlockingQueue<String> queue; private final LoggerThread loggerThread; private final PrintWriter writer; @GuardedBy("this") private boolean isShutdown; @GuardedBy("this") private int reservations; public void start() { loggerThread.start(); } public void stop() { synchronized (this) { isShutdown = true; } loggerThread.interrupt(); } public void log(String msg) throws InterruptedException { synchronized(this) { if(isShutdown) throw new IllegalStateException(...); ++reservations; } queue.put(msg); } private class LoggerThread extends Thread { public void run() { try { while(true) { try { synchronized(LogService.this) { if (isShutdown &;&; reservations == 0) break; } String msg = queue.take(); synchronized(LogService.this) { --reservations; } writer.println(msg); } catch (InterruptedException e) { /* retry */ } } } finally { writer.close(); } } }} 7.2.2 關閉ExecutorService
  1. 在進行強制關閉時,shutdownNow首先關閉當前正在執行的任務,然後返回所有尚未啟動的任務清單。
  2. LogService的一種變化形式,它將管理執行緒的工作委託給一個ExecutorService,而不是由其自行管理。通過封裝ExecutorService,可以將所有權鏈(Ownership Chain)從應用程式擴充套件到服務以及執行緒,所有權鏈上的各個成員都將管理它所擁有的服務或執行緒的生命週期。
// 使用ExecutorService的日誌服務public class LogService { private final ExecutorService exec = newSingleThreadExecutor(); ... public void start() { } public void stop() throws InterruptedException { try { exec.shutdown(); exec.awaitTermination(TIMEOUT, UNIT); } finally { writer.close(); } } public void log(String msg) { try { exec.execute(new WriteTask(msg)); } catch(RejectedExecutionException ignored) { } }} 7.2.4 示例:只執行一次的服務// 使用私有的Executor,並且該Executor的生命週期受限於方法呼叫boolean checkMail(Set<String> hosts, long timeout, TimeUnit unit) throws InterruptedException { ExecutorService exec = Executors.newCachedThreadPool(); final AtomicBoolean hasNextMail = new AtomicBoolean(false); try { for (final String host : hosts) exec.execute(new Runnable() { public void run() { if (checkMail(host)) hasNextMail.set(true); } }); } finally { exec.shutdown(); exec.awaitTermination(timeout, unit); } return hasNewMail.get();} 7.2.5 shutdownNow的侷限性
  1. 當通過shutdownNow來強行關閉ExecutorService時,它會嘗試取消正在執行的任務。並返回所有已提交但尚未開始的任務,從而將這些任務寫入日誌或者儲存起來以便之後進行處理。shutdownNow返回的Runnable物件可能與提交給ExecutorService的Runnable物件並不相同:它們可能是被封裝過的已提交任務。
  2. 通過封裝ExecutorSerivce並使得execute(類似地還有submit)記錄哪些任務是在關閉後取消的,TrackingExecutor可以找出哪些任務已經開始但還沒有正常完成。
// 在ExecutorService中跟蹤在關閉之後被取消的任務public class TrackingExecutor extends AbstractExecutorService { private final ExecutorService exec; private final Set<Runnable> tasksCancelledAtShutdown = Collections.synchronizedSet(new HashSet<Runnable>()); ... public List<Runnable> getCancelledTasks() { if (!exec.isTerminated()) throw new IllegalStateException(...); return new ArrayList<Runnable>(tasksCancelledAtShutdown); } public void execute(final Runnable runnable) { exec.execute(new Runnable() { public void run() { try { runnable.run(); } finally { if (isShutdown() &;&; Thread.currentThread().isInterrupted()) // 說明執行緒沒有正常執行完 tasksCancelledAtShutdown.add(runnable); } } }); }// 將ExecutorService的其他方法委託給exec}

網頁爬蟲程式的工作通常是無窮盡的,因此當爬蟲程式必須關閉時,我們通常希望儲存它的狀態,以便稍後重新啟動。

// 使用TrackingExecutorService來儲存未完成的任務以備後續執行public abstract class WebCrawler { private volatile TrackingExecutor exec; @GuardedBy("this") private final Set<URL> urlsToCrawl = new HashSet<URL>(); ... private synchronized void start() { exec = new TrackingExecutor(Executors.newCachedThreadPool()); for(URL url : urlsToCrawl) submitCrawlTask(url); urlsToCrawl.clear(); } public synchronized void stop() throws InterruptedException { try { saveUncrawled(exec.shutdownNow()); // 儲存已提交但還未開始的任務,shutdownNow的返回值 if (exec.awaitTermination(TIMEOUT, UNIT)) saveUncrawled(exec.getCancelledTasks()); // TrackingExecutor記錄下的沒執行完的runnable們 } finally { exec = null; } } protected abstract List<URL> processPage(URL url); // 迭代處理的意思 private void saveUncrawled(List<Runnable> uncrawled) { for(Runnable task : uncrawled) urlsToCrawl.add(((CrawlTask) task).getPage()); } private void submitCrawlTask(URL u) { exec.execute(new CrawlTask(u)); // TrackingExecutor在這裡記錄 } private class CrawlTask implements Runnable { private final URL url; ... public void run() { for(URL link : processPage(url)) { if(Thread.currentThread().isInterrupted()) return; submitCrawlTask(link); } } public URL getPage() { return url; } }}

在TrackingExecutor中存在一個不可避免的競態條件,從而產生“誤報”問題:一些被認為已取消的任務實際上已經執行完成。這個問題的原因在於,在任務執行最後一條指令以及執行緒池將任務記錄為“結束”的兩個時刻之間,執行緒池可能被關閉。如果任務是冪等的(Idempotent,即將任務執行兩次與執行一次會得到相同的結果),那麼這不會存在問題,在網頁爬蟲程式中就是這種情況。

7.3 處理非正常的執行緒終止
  1. 當一個執行緒由於未捕獲異常而退出時,JVM會把這個事件報告給應用程式提供的UncaughtExceptionHandler異常處理器。如果沒有提供任何異常處理器,那麼預設的行為是將棧追蹤資訊輸出到System.err。
// UncaughtExceptionHandler介面public interface UncaughtExceptionHandler { void uncaughtException(Thread t, Throwable e);}

要為執行緒池中的所有執行緒設定一個UncaughtExceptionHandler,需要為ThreadPoolExecutor的建構函式提供一個ThreadFactory。(與所有的執行緒操控一樣,只有執行緒的所有者能夠改變執行緒的UncaughtExceptionHandler。)標準執行緒池允許當發生未捕獲異常時結束執行緒,但由於使用了一個try-finally程式碼塊來接收通知,因此當執行緒結束時,將有新的執行緒來代替它。

令人困惑的是,只有通過execute提交的任務,才能將它丟擲的異常交給未捕獲異常處理器,而通過submit提交的任務,無論是丟擲的未檢查異常還是已檢查異常,都將被認為是任務返回狀態的一部分。如果一個由submit提交的任務由於丟擲了異常而結束,那麼這個異常將被Funture.get封裝在ExecutionException中重新丟擲。

7.4 JVM關閉 7.4.1 關閉鉤子
  1. 在正常關閉中,JVM首先呼叫所有已註冊的關閉鉤子(Shutdown Hook)。關閉鉤子是指通過Runtime.addShutdownHook註冊的但尚未開始的執行緒。
  2. 在關閉應用程式執行緒時,如果有(守護或非守護)執行緒仍然在執行,那麼這些執行緒接下來將與關閉程序併發執行。當所有的關閉鉤子都執行結束時,如果runFinalizersOnExit為true,那麼JVM將執行終結器,然後再停止。JVM並不會停止或中斷任何在關閉時仍然執行的應用程式執行緒。當JVM最終結束時,這些執行緒將被強行結束。
// 通過註冊一個關閉鉤子來停止日誌服務public void start() { Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { try { LogService.this.stop(); } catch (InterruptedException ignored) {} } });} 7.4.2 守護執行緒
  1. 有時候,你希望建立一個執行緒來執行一些輔助工作,但又不希望這個執行緒阻礙JVM的關閉。在這種情況下就需要使用守護執行緒(Daemon Thread)。
  2. 在JVM啟動時建立的所有執行緒中,除了主執行緒以外,其他的執行緒都是守護執行緒(例如垃圾回收器以及其他執行輔助工作的執行緒)。當建立一個新執行緒時,新執行緒將繼承建立它的執行緒的守護狀態,因此預設情況下,主執行緒建立的所有執行緒都是普通執行緒。
  3. 普通執行緒與守護執行緒之間的差異僅在於當執行緒退出時發生的操作。當一個執行緒退出時,JVM會檢查其他正在執行的執行緒,如果這些執行緒都是守護執行緒,那麼JVM會正常退出操作。當JVM停止時,所有仍然存在的守護執行緒都將被拋棄——既不會執行finally程式碼塊,也不會執行回捲棧,而JVM只是直接退出。
  4. 我們應儘可能少地使用守護執行緒——很少有操作能夠在不進行清理的情況下被安全地拋棄。特別是,如果在守護執行緒中執行可能包含I/O操作的任務,那麼將是一種危險的行為。守護執行緒最好用於執行“內部”任務,例如週期性地從記憶體的快取中移出逾期的資料。
  5. 此外,守護執行緒通常不能用來替代應用程式管理程式中各個服務的生命週期。
7.4.3 終結器
  1. 在大多數情況下,通過使用finally程式碼塊和顯式的close方法,能夠比使用終結器更好地管理資源。唯一的例外情況在於:當需要管理物件,並且該物件持有的資源是通過本地方法獲得的。
  2. 避免使用終結器。
小結

通過使用FutureTask和Executor框架,可以幫助我們構建可取消的任務和服務。

第8章 執行緒池的使用 8.1 在任務與執行策略之間的隱性耦合
  1. 只有當執行緒本地值得生命週期受限於任務的生命週期時,線上程池的執行緒中使用ThreadLocal才有意義,而線上程池的執行緒中不應該使用ThreadLocal在任務之間傳遞值。
  2. 只有當任務都是同類型的並且相互獨立時,執行緒池的效能才能達到最佳。如果將執行時間較長的與執行時間較短的任務混合在一起,那麼除非執行緒池很大,否則將可能造成“擁塞”。如果提交的任務依賴於其他任務,那麼除非執行緒池無限大,否則將可能造成死鎖。
  3. 如果某些任務依賴於其他的任務,那麼會要求執行緒池足夠大,從而確保它們依賴任務不會被放入等待佇列中或被拒絕,而採用執行緒封閉機制的任務需要序列執行。
// 在單執行緒Executor中任務發生死鎖(不要這麼做)public class ThreadDeadlock { ExecutorService exec = Executors.newSingleThreadExecutor(); public class RenderPageTask implements Callable<String> { public String call() throws Exception { Future<String> header, footer; header = exec.submit(new LoadFileTask("header.html")); footer = exec.submit(new LoadFileTask("footer.html")); String page = renderBody(); // 將發生死鎖——由於任務在等待子任務的結果 return header.get() + page + footer.get(); } }} // 單執行緒Executor,現在在執行RenderPageTask,而它依賴的兩個LoadFileTask卻會一直在等待被Executor執行。

有一項技術可以緩解執行時間較長任務造成的影響,即限定任務等待資源的時間,而不要無限制地等待。在平臺類庫的大多數可阻塞方法中,都同時定義了限時版本和無限時版本,例如Thread.join、BlockingQueue.put、CountDownLatch.await以及Selector.select等。

8.2 設定執行緒池的大小
  1. 在程式碼中通常不會固定執行緒池的大小,而應該通過某種配置機制來提供,或者根據Runtime.availableProcessors來動態計算。
  2. 要想正確地設定執行緒池的大小,必須分析計算環境、資源預算和任務的特性。在部署的系統中有多少個CPU?多大的記憶體?任務是計算密集型、I/O密集型還是二者皆可?它們是否需要像JDBC連線這樣的稀缺資源?如果需要執行不同類別的任務,並且它們之間的行為相差很大,那麼應該考慮使用多個執行緒池,從而使每個執行緒池可以根據各自的工作負載來調整。
  3. 對於計算密集型的任務,在擁有 Ncpu 個處理器的系統上,當執行緒池的大小為 Ncpu+1 時,通常能實現最優的利用率。(即使當計算密集型的執行緒偶爾由於頁缺失故障或者其他原因而暫停時,這個“額外”的執行緒也能確保CPU的時鐘週期不會被浪費。)對於包含I/O操作或者其他阻塞操作的任務,由於執行緒並不會一直執行,因此執行緒池的規模應該更大。
  4. Ncpu=number of CPUs
    Ucpu=target CPU utilization,0≤Ucpu≤1
    WC=ratio of wait time to compute time
    要使處理器達到期望的使用率,執行緒池的最優大小等於:
    Nthreads=Ncpu∗Ucpu∗(1+WC)
    可以通過Runtime來獲得CPU的數目:
int N_CPUS = Runtime.getRuntime().availableProcessors(); 8.3 配置ThreadPoolExecutor// ThreadPoolExecutor的通用建構函式public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { ... } 8.3.1 執行緒的建立與銷燬
  1. 基本大小也就是執行緒池的目標大小,即在沒有任務執行時執行緒池的大小,並且只有在工作佇列滿了的情況下才會建立超出這個數量的執行緒。執行緒池的最大大小表示可同時活動的執行緒數量的上限。如果某個執行緒的空閒時間超過了存活時間,那麼將被標記為可回收的,並且當執行緒池的當前大小超過了基本大小時,這個執行緒將被終止。
  2. newFixedThreadPool工廠方法將執行緒池的基本大小和最大大小設定為引數中指定的值,而且建立的執行緒池不會超時。newCachedThreadPool工廠方法將執行緒池的最大大小設定為Integer.MAX_VALUE,而將基本大小設定為零,並將超時設定為1分鐘,這種方法創建出來的執行緒池可以被無線擴充套件,並且當需求降低時會自動收縮。
8.3.2 管理佇列任務
  1. newFixedThreadPool和newSingleThreadExecutor在預設情況下將使用一個無界的LinkedBlockingQueue。
  2. 一種更穩妥的資源管理策略是使用有界佇列,例如ArrayBlockingQueue、有界的LinkedBlockingQueue、PriorityBlockingQueue。有界佇列有助於避免資源耗盡的情況發生,但它又帶來了新的問題:當佇列填滿後,新的任務該怎麼辦?(有許多飽和策略[Saturation Policy]可以解決這個問題。)
  3. 對於非常大的或者無界的執行緒池,可以通過使用SynchronousQueue來避免任務排隊,以及直接將任務從生產者移交給工作者執行緒。SynchronousQueue不是一個真正的佇列,而是一種線上程之間進行移交的機制。要將一個元素放入SynchronousQueue中,必須有另一個執行緒正在等待接受這個元素。如果沒有執行緒正在等待,並且執行緒池的當前大小小於最大值,那麼ThreadPoolExecutor將建立一個新的執行緒,否則根據飽和策略,這個任務將被拒絕。
  4. 只有當執行緒池是無界的或者可以拒絕任務時,SynchronousQueue才有實際價值。在newCachedThreadPool工廠方法中就使用了SynchronousQueue。
  5. 對於Executor,newCachedThreadPool工廠方法是一種很好的預設選擇,它能提供比固定大小的執行緒池更好的排隊效能。(這種效能差異是由於使用了SynchronousQueue而不是LinkedBlockingQueue。)
  6. 只有當任務相互獨立時,為執行緒池或工作佇列設定界限才是合理的。如果任務之間存在依賴性,那麼有界的執行緒池或佇列就可能導致執行緒“飢餓”死鎖問題。此時應該使用無界的執行緒池,例如newCachedThreadPool。
8.3.3 飽和策略
  1. ThreadPoolExecutor的飽和策略可以通過呼叫setRejectedExecutionHandler來修改。(如果某個任務被提交到一個已被關閉的Executor時,也會用到飽和策略。)JDK提供了幾種不同的RejectedExecutionHandler實現,每種實現都包含有不同的飽和策略:AbortPolicy、CallerRunsPolicy、DiscardPolicy和DiscardOldestPolicy。
  2. “中止(Abort)”策略是預設的飽和策略,該策略將丟擲未檢查的RejectedExecutionException。呼叫者可以捕獲這個異常,然後根據需求編寫自己的處理程式碼。
  3. “拋棄最舊的(Discard-Oldest)”策略則會拋棄下一個將被執行的任務,然後嘗試重新提交新的任務。(如果工作佇列是一個優先佇列,那麼“拋棄最舊的”策略將導致拋棄優先順序最高的任務,因此最好不要將“拋棄最舊的”飽和策略和優先順序佇列放在一起使用。)
  4. “呼叫者執行(Caller-Runs)”策略實現了一種調節機制,該策略既不會拋棄任務,也不會丟擲異常,而是將某些任務回退到呼叫者,從而降低新任務的流量。它不會線上程池的某個執行緒中執行新提交的任務,而是在一個呼叫了execute的執行緒中執行該任務。我們可以將WebServer示例修改為使用有界佇列和“呼叫者執行”飽和策略,當執行緒池中的所有執行緒都被佔用,並且工作佇列被填滿後,下一個任務會在呼叫execute時在主執行緒中執行。由於執行任務需要一定的時間,因此主執行緒至少在一段時間內不能提交任何任務,從而使得工作者執行緒有時間來處理完正在執行的任務。在這期間,主執行緒不會呼叫accept,因此到達的請求將被儲存在TCP層的佇列中而不是在應用程式的佇列中。如果持續過載,那麼TCP層將最終發現它的請求佇列被填滿,因此同樣會開始拋棄請求。當伺服器過載時,這種過載情況會逐漸向外蔓延開來——從執行緒池到工作佇列到應用程式再到TCP層,最終達到客戶端,導致伺服器在高負載下實現一種平緩的效能降低。
// 建立一個固定大小的執行緒池,並採用有界佇列以及“呼叫者執行”飽和策略ThreadPoolExecutor executor = new ThreadPoolExecutor(N_THREADS, N_THREADS, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(CAPACITY));executor.setRejectedExecutionHandler( new ThreadPoolExecutor.CallerRunsPolicy()); 8.3.4 執行緒工廠
  1. 每當執行緒池需要建立一個執行緒時,都是通過執行緒工廠方法來完成的。
// ThreadFactory介面public interface ThreadFactory { Thread newThread(Runnable r);}// 自定義的執行緒工廠public class MyThreadFactory implements ThreadFactory { private final String poolName; public MyThreadFactory(String poolName) { this.poolName = poolName; } public Thread newThread(Runnable runnable) { return new MyAppThread(runnable, poolName); }}

在MyAppThread中還可以定製其他行為,包括:為執行緒指定名字,設定自定義UncaughtExceptionHandler向Logger中寫入資訊,維護一些統計資訊(包括有多少個執行緒被建立和銷燬),以及線上程被建立或者終止時把除錯訊息寫入日誌。

如果在應用程式中需要利用安全策略來控制對某些特殊程式碼庫的訪問許可權,那麼可以通過Executor中的privilegedThreadFactory工廠來定製自己的執行緒工廠。通過這種方式創建出來的執行緒,將與建立privilegedThreadFactory的執行緒擁有相同的訪問許可權、AccessControlContext和contextClassLoader。如果不使用privilegedThreadFactory,執行緒池建立的執行緒將從在需要新執行緒時呼叫execute或submit的客戶程式中繼承訪問許可權,從而導致令人困惑的安全性異常。

8.3.5 在呼叫建構函式後再定製ThreadPoolExecutor// 對通過標準工廠方法建立的Executor進行修改ExecutorService exec = Executors.newCachedThreadPool();if (exec instanceof ThreadPoolExecutor) ((ThreadPoolExecutor) exec).setCorePoolSize(10);else throw new AssertionError("Oops, bad assumption");

在Executors中包含一個unconfigurableExecutorService工廠方法,該方法對一個現有的ExecutorService進行包裝,使其只暴露出ExecutorService的方法,因此不能對它進行配置。newSingleThreadExecutor返回按這種方式封裝的ExecutorService,而不是最初的ThreadPoolExecutor。

啦啦啦,讓我們重溫一下包裝器模式:unconfigurableExecutorService
首先,它實際上返回的是DelegatedExecutorService,這才是真正的包裝器類

public static ExecutorService unconfigurableExecutorService(ExecutorService executor) { if (executor == null) throw new NullPointerException(); return new DelegatedExecutorService(executor);}

接著我們就來看看漂亮的DelegatedExecutorService包裝器吧!

static class DelegatedExecutorService extends AbstractExecutorService { private final ExecutorService e; // 被包裝的例項啦 DelegatedExecutorService(ExecutorService executor) { e = executor; } public void execute(Runnable command) { e.execute(command); } public void shutdown() { e.shutdown(); } public List<Runnable> shutdownNow() { return e.shutdownNow(); } public boolean isShutdown() { return e.isShutdown(); } public boolean isTerminated() { return e.isTerminated(); } public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { return e.awaitTermination(timeout, unit); } public Future<?> submit(Runnable task) { return e.submit(task); } public <T> Future<T> submit(Callable<T> task) { return e.submit(task); } public <T> Future<T> submit(Runnable task, T result) { return e.submit(task, result); } public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { return e.invokeAll(tasks); } public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException { return e.invokeAll(tasks); } public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { return e.invokeAny(tasks); } public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return e.invokeAny(tasks, timeout, unit); }} 8.4 擴充套件ThreadPoolExecutor
  1. ThreadPoolExecutor是可擴充套件的,它提供了幾個可以子類化中改寫的方法:beforeExecute、afterExecute和terminated,這些方法可以用於擴充套件ThreadPoolExecutor的行為。
  2. 無論任務是從run中正常返回,還是丟擲一個異常而返回,afterExecute都會被呼叫。(如果任務在完成後帶有一個Error,那麼就不會呼叫afterExecute。)如果beforeExecute丟擲一個RuntimeException,那麼任務將不被執行,並且afterExecute也不會被呼叫。
  3. 線上程池完成關閉操作時呼叫terminated,也就是在所有任務都已經完成並且所有工作者執行緒也已經關閉後。
// 增加了日誌和計時等功能的執行緒池public class TimingThreadPool extends ThreadPoolExecutor { private final ThreadLocal<Long> startTime = new ThreadLocal<Long>(); private final Logger log = Logger.getLogger("TimingThreadPool"); private final AtomicLong numTasks = new AtomicLong(); private final AtomicLong totalTime = new AtomicLong(); protected void beforeExecute(Thread t, Runnable r) { super.beforeExecute(t, r); log.fine(String.format("Thread %s: start %s", t, r)); startTime.set(System.nanoTime()); } protected void afterExecute(Runnable r, Throwable t) { try { long endTime = System.nanoTime(); long taskTime = endTime - startTime.get(); numTasks.incrementAndGet(); totalTime.addAndGet(taskTime); log.fine(String.format("Thread %s, end %s, time=%dns", t, r, taskTime)); } finally { super.afterExecute(r,t); } } protected void terminated() { try { log.info(String.format("Terminated: avg time=%dns", totalTime.get() / numTasks.get())); } finally { super.terminated(); } }} 8.5 遞迴演算法的並行化// 將序列執行轉換為並行執行void processSequentially(List<Element> elements) { for (Element e : elements) process(e);}void processInParallel(Executor exec, List<Element> elements) { for (final Element e : elements) exec.execute(new Runnable() { public void run() { process(e); } });}

如果需要提交一個任務集並等待它們完成,那麼可以使用ExecutorService.invokeAll,並且在所有任務都執行完成後呼叫CompletionService來獲取結果。

一種簡單的情況是:在每個迭代操作中都不需要來自於後續遞迴迭代的結果。例如,sequentialRecursive用深度優先演算法遍歷一棵樹,在每個節點上執行計算並將結果放入一個集合。

// 將序列遞迴轉換為並行遞迴public <T> void sequentialRecursive(List<Node<T>> nodes, Collection<T> results) { for (Node<T> n : nodes) { results.add(n.compute()); sequentialRecursive(n.getChildren(), results); }}public <T> void parallelRecursive(final Executor exec, List<Node<T>> nodes, final Collection<T> results) { for (final Node<T> n : nodes) { exec.execute(new Runnable() { public void run() { results.add(n.compute()); } }); parallelRecursive(exec, n.getChildren(), results); }}

遍歷過程仍然是序列的,只有compute呼叫才是並行執行的。

// 等待通過並行方式計算的結果public <T> Collection<T> getParallelResults(List<Node<T>> nodes) throws InterruptedException { ExecutorService exec = Executors.newCachedThreadPool(); Queue<T> resultQueue = new ConcurrentLinkedQueue<T>(); parallelRecursive(exec, nodes, resultQueue); exec.shutdown(); exec.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS); return resultQueue;} 示例:謎題框架

我們將“謎題”定義為:包含了一個初始位置,一個目標位置,以及用於判斷是否有效移動的規則集。規則集包含兩部分:計算從指定位置開始的所有合法移動,以及每次移動的結果位置。

// 表示“搬箱子”之類謎題的抽象類public interface Puzzle<P, M> { P initialPosition(); boolean isGoal(P position); Set<M> legalMoves(P position); P move(P position, M move);}// 用於謎題解決框架的連結串列節點@Immutablestatic class Node<P, M> { final P pos; final M move; final Node<P, M> prev; Node(P pos, M move, Node<P,M> prev) { ... } List<M> asMoveList() { List<M> solution = new LinkedList<M>(); for (Node<P,M> n = this; n.move!=null; n=n.prev) solution.add(0, n.move); // 新增到連結串列頭 return solution; }}// 序列的謎題解答器public class SequentialPuzzleSolver<P,M> { private final Puzzle<P,M> puzzle; private final Set

seen = new HashSet

(); // 避免重複位置 public SequentialPuzzleSolver(Puzzle<P,M> puzzle) { this.puzzle = puzzle; } public List<M> solve() { P pos = puzzle.initialPosition(); return search(new Node<P,M>(pos, null, null)); } private List<M> search(Node<P,M> node) { if (!seen.contains(node.pos)) { seen.add(node.pos); if(puzzle.isGoal(node.pos)) return node.asMoveList(); for(M move : puzzle.legalMoves(node.pos)) { P pos = puzzle.move(node.pos, move); Node<P,M> child = new Node<P,M>(pos, move, node); List<M> result = search(child); // 深搜 if(result!=null) return result; } } return null; } static class Node<P,M> { ... }}

通過修改解決方案以利用併發性,可以以並行方式來計算下一步移動以及目標條件,因為計算某次移動的過程在很大程度上與計算其他移動的過程是相互獨立的。(之所以說“在很大程度上”,是因為在各個任務之間會共享一些可變狀態,例如已遍歷位置的集合。)

// 併發的謎題解答器public class ConcurrentPuzzleSolver<P,M> { private final Puzzle<P,M> puzzle; private final ExecutorService exec; private final ConcurrentMap<P,Boolean> seen; final ValueLatch<Node<P,M>> solution = new ValueLatch<Node<P,M>>(); ... public List<M> solve() throws InterruptedException { try { P p = puzzle.initialPosition(); exec.execute(newTask(p, null, null)); // 阻塞直到找到解答 Node<P,M> solnNode = solution.getValue(); return (solnNode==null) ? null : solnNode.asMoveList(); } finally { exec.shutdown(); } } protected Runnable newTask(P p, M m, Node<P,M> n) { return new SolverTask(p, m, n); } class SolverTask extends Node<P,M> implements Runnable { ... public void run() { // putIfAbsent returns the previous value associated with the specified key, or null if there was no mapping for the key if (solution.isSet() || seen.putIfAbsent(pos, true)!=null) return; // 已經找到了解答或者已經遍歷了這個位置 if (puzzle.isGoal(pos)) solution.setValue(this); else for (M m : puzzle.legalMoves(pos)) exec.execute( newTask(puzzle.move(pos,m), m, this)); } }}

為了避免無限迴圈,在序列版本中引入了一個Set物件,其中儲存了之前已經搜尋過的所有位置。在ConcurrentPuzzleSolver中使用ConcurrentHashMap來實現相同的功能。這種做法不僅提供了執行緒安全性,還避免了在更新共享集合時存在的競態條件,因為putIfAbsent只有在之前沒有遍歷過的某個位置才會通過原子方式新增到集合中。ConcurrentPuzzleSolver使用執行緒池的內部工作佇列而不是呼叫棧來儲存搜尋的狀態。

序列版本的程式執行深度優先搜尋,因此搜尋過程將受限於棧的大小。併發版本的程式執行廣度優先搜尋,因此不會受到棧大小的限制(但如果待搜尋的或者已搜尋的位置集合大小超過了可用的記憶體總量,那麼仍可能耗盡記憶體)。

為了在找到某個解答後停止搜尋,需要通過某種方式來檢查是否有執行緒已經找到了一個解答。如果需要第一個找到的解答,那麼還需要在其他任務都沒有找到解答時更新解答。這些需求描述的是一種閉鎖(Latch)機制,具體地說,是一種包含結果的閉鎖。

ValueLatch中使用CountDownLatch來實現所需的閉鎖行為,並且使用鎖定機制來確保解答只會被設定一次。

// 由ConcurrentPuzzleSolver使用的攜帶結果的閉鎖@ThreadSafepublic class ValueLatch<T> { @GuardedBy("this") private T value = null; // 攜帶的結果 private final CountDownLatch done = new CountDownLatch(1); public boolean isSet() { return (done.getCount()==0); } public synchronized void setValue(T newValue) { // synchronized if(!isSet()) { value = newValue; done.countDown(); } } public T getValue() throws InterruptedException { done.await(); synchronized(this) { // synchronized return value; } }}

第一個找到解答的執行緒還會關閉Executor,從而阻止接受新的任務。要避免處理RejectedExecutionException,需要將拒絕執行處理器設定為“拋棄已提交的任務”。然後,所有未完成的任務最終將執行完成,並且在執行任何新任務時都會失敗,從而使Executor結束。

如果不存在解答,那麼ConcurrentPuzzleSolver就不能很好地處理這種情況:如果已遍歷了所有的移動和位置都沒有找到解答,那麼在getSolution呼叫中將永遠等待下去。當遍歷了整個搜尋空間時,序列版本的程式將結束,但要結束併發程式會更困難。其中一種方法是:記錄活動任務的數量,當該值為零時將解答設定為null。

第9章 圖形使用者介面應用程式 9.1 為什麼GUI是單執行緒的
  1. 單執行緒的GUI框架並不僅限於在Java中,在Qt、NexiStep、MacOS Cocoa、X Windows以及其他環境中的GUI框架都是單執行緒的。許多人曾經嘗試過編寫多執行緒的GUI框架,但最終都由於競態條件和死鎖導致的穩定性問題而又重新回到單執行緒的事件佇列模型:採用一個專門的執行緒從佇列中抽取事件,並將它們轉發到應用程式定義的事件處理器。
  2. 在多執行緒的GUI框架中更容易發生死鎖問題,其部分原因在於,在輸入事件的處理過程與GUI元件的面向物件模型之間會存在錯誤的互動。使用者引發的動作將通過一種類似於“氣泡上升”的方式從作業系統傳遞給應用程式——作業系統首先檢測到一次滑鼠點選,然後通過工具包將其轉化為“滑鼠點選”事件,該事件最終被轉換為一個更高層事件(例如“滑鼠鍵被按下”事件)轉發給應用程式的監聽器。另一方面,應用程式引發的動作又會以“氣泡下沉”的方式從應用程式返回到作業系統。例如,在應用程式中引發修改某個元件背景色的請求,該請求將轉發給某個特定的元件類,並最終轉發給作業系統進行繪製。因此,一方面這組操作將以完全相反的順序來訪問相同的GUI物件;另一方面又要確保每個物件都是執行緒安全的,從而導致不一致的鎖定順序,並引發死鎖。這種問題幾乎在每次開發GUI工具包時都會重現。
  3. 另一個在多執行緒GUI框架中導致死鎖的原因就是“模型-檢視-控制(MVC)”這種設計模式的廣泛使用。
  4. “控制”模組將呼叫“模型”模組,而“模型”模組將發生的變化通知給“檢視”模組。“控制”模組同樣可以呼叫“檢視”模組,並呼叫“模型”模組來查詢模型的狀態。這將再次導致不一致的鎖定順序並出現死鎖。
  5. 單執行緒的GUI框架通過執行緒封閉機制來實現執行緒安全性。所有GUI物件,包括視覺化元件和資料模型等,都只能在事件執行緒中訪問。
9.1.1 序列事件處理
  1. 因此,在事件執行緒中執行的任務必須儘快地把控制權交還給事件執行緒。要啟動一些執行時間較長的任務,例如對某個大型文件執行拼寫檢查,在檔案系統中執行搜尋,或者通過網路獲取資源等,必須在另一個執行緒中執行這些任務,從而儘快地將控制權交還給事件執行緒。如果要在執行某個時間較長的任務時更新進度標識,或者在任務完成後提供一個視覺化的反饋,那麼需要再次執行事件執行緒中的程式碼。這也很快會使程式變得更復雜。
9.1.2 Swing中的執行緒封閉機制
  1. 所有Swing元件(例如JButton和JTable)和資料模型物件(例如TableModel和TreeModel)都被封閉在事件執行緒中,因此任何訪問它們的程式碼都必須在事件執行緒中執行。
  2. 這種方法的好處在於,當訪問表現物件(Presentation Object)時在事件執行緒中執行的任務無須擔心同步問題,而壞處在於,無法從事件執行緒之外的執行緒中訪問表現物件。
  3. Swing的單執行緒規則是:Swing中的元件以及模型只能在這個事件分發執行緒中進行建立、修改以及查詢。
  4. 單執行緒規則的其他一些例外情況包括:

    SwingUtilities.isEventDispatchThread,用於判斷當前執行緒是否是事件執行緒。
    SwingUtilities.invokeLater,該方法可以將一個Runnable任務排程到事件執行緒中執行(可以從任意執行緒中呼叫。
    SwingUtilities.invokeAndWait,該方法可以將一個Runnable任務排程到事件執行緒中執行,並阻塞當前執行緒直到任務完成(只能從非GUI執行緒中呼叫)。
    所有將重繪(Repaint)請求或重生效(Revalidation)請求插入佇列的方法(可從任意執行緒中呼叫)。
    所有新增或移除監聽器的方法(這些方法可以從任意執行緒中呼叫,但監聽器本身一定要在事件執行緒中呼叫)。


invokeLater和invokeAndWait兩個方法的作用酷似Executor。// 使用Executor來實現SwingUtilitiespublic class SwingUtilities { private static final ExecutorService exec = Executors.newSingleThreadExecutor(new SwingThreadFactory()); private static volatile Thread swingThread; private static class SwingThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { swingThread = new Thread(r); return swingThread; } } public static boolean isEventDispatchThread() { return Thread.currentThread() == swingThread; } public static void invokeLater(Runnable task) { exec.execute(task); } public static void invokeAndWait(Runnable task) throws InterruptedException, InvocationTargetException { Future f = exec.submit(task); try { f.get(); } catch (ExecutionException e) { throw new InvocationTargetException(e); } }}GuiExecutor是一個Executor,它將任務委託給SwingUtilities來執行。也可以採用其他的GUI框架來實現它,例如SWT提供的Display.asyncExec方法,它類似於Swing中的invokeLater。// 基於SwingUtilities構建的Executorpublic class GuiExecutor extends AbstractExecutorService { // 採用“單件(Singleton)”模式,有一個私有建構函式和一個公有的工廠方法 private static final GuiExecutor instance = new GuiExecutor(); private GuiExecutor() { } public static GuiExecutor instance() { return instance; } public void execute(Runnable r) { if (SwingUtilities.isEventDispatchThread()) r.run(); else SwingUtilities.invokeLater(r); } // 其他生命週期方法的實現} 9.2 短時間的GUI任務
  1. 按鈕點選時的執行控制流:EDT->滑鼠點選->動作事件->動作監聽者->設定顏色
  2. Swing將大多數視覺化元件都分為兩個物件,即模型物件與檢視物件。在模型物件中儲存的是將被顯示的資料,而在檢視物件中則儲存了控制顯示方式的規則。模型物件可以通過引發事件來表示模型資料發生了變化,而檢視物件則通過“訂閱”來接收這些事件。當檢視物件收到表示模型資料已經變化的事件時,將向模型物件查詢新的資料,並更新介面顯示。
  3. 模型物件與檢視物件的控制流:EDT->滑鼠點選->動作事件->動作監聽者->更新表格模型->表格修改事件->表格監聽器->更新表格檢視
9.3 長時間的GUI任務
  1. 這個示例通過“Fire and Forget”方式將長時間任務從事件執行緒中分離出來,這種方式可能並不是非常有用。在執行完一個長時間的任務後,通常會產生某種視覺化的反饋。但你並不能從後臺執行緒中訪問這些表現物件,因此任務在完成時必須向事件執行緒提交另一個任務來更新使用者介面。
// 將一個長時間任務繫結到一個視覺化元件ExecutorService backgroundExec = Executors.newCachedThreadPool();...button.addActionListener(new ActionListener() { public void actionPerformed(ActionEvent e) { backgroundExec.execute(new Runnable() { public void run() { doBigComputation(); } }); }});

動作監聽器首先使按鈕無效,並設定一個標籤表示正在進行某個計算,然後將一個任務提交給後臺的Executor。當任務完成時,它會在事件執行緒中增加另一個任務,該任務將重新啟用按鈕並恢復標籤文字。

// 支援使用者反饋的長時間任務button.addActionListener(new ActionListener() { public void actionPerformed(ActionEvent e) { button.setEnabled(false); label.setText("busy"); backgroundExec.execute(new Runnable() { public void run() { try { doBigComputation(); } finally { GuiExecutor.instance().execute(new Runnable() { public void run() { button.setEnabled(true); label.setText("idle"); } }); } } }); }});

在GUI應用程式中,這種“執行緒接力”是處理長時間任務的典型方法。

9.3.1 取消
  1. 你可以直接通過執行緒中斷來實現取消操作,但是一種更簡單的辦法是使用Future,專門用來管理可取消的任務。
  2. 如果呼叫Future的cancel方法,並將引數mayInterruptIfRunning設定為true,那麼這個Future可以中斷正在執行任務的執行緒。如果你編寫的任務能夠響應中斷,那麼當它被取消時就可以提前返回。
// 取消一個長時間任務Future<?> runningTask = null; // 執行緒封閉...startButton.addActionListener(new ActionListener() { public void actionPerformed(ActionEvent e) { if(runningTask!=null) { runningTask = backgroundExec.submit(new Runnable() { public void run() { while(moreWork()) { if(Thread.currentThread().isInterrupted()) { cleanUpPartialWork(); break; } doSomeWork(); } } }); } }});cancelButton.addActionListener(new ActionListener() { public void actionPerformed(ActionEvent event) { if(runningTask!=null) runningTask.cancel(true); }}); // 可以處理到這個事件,因為上面的任務是在另一個執行緒中執行,控制權早就已返還給事件執行緒了

由於runningTask被封閉在事件執行緒中,因此在對它進行設定或檢查時不需要同步,並且“開始”按鈕的監聽器可以確保每次只有一個後臺任務在執行。然而,當任務完成時最好能通知按鈕監聽器,例如說可以禁用“取消”按鈕。

9.3.2 進度標識和完成標識
  1. 通過Future來表示一個長時間的任務,可以極大地簡化取消操作的實現。在FutureTask中也有一個done方法同樣有助於實現完成通知。當後臺的Callable完成後,將呼叫done。通過done方法在事件執行緒中觸發一個完成任務,我們能夠構造一個BackgroundTask類,這個類將提供一個在事件執行緒中呼叫的onCompletion方法。
// 支援取消,完成通知以及進度通知的後臺任務類abstract class BackgroundTask<V> implements Runnable, Future<V> { private final FutureTask<V> computation = new Computation(); private class Computation extends FutureTask<V> { public Computation() { super(new Callable<V>() { public V call() throws Exception { return BackgroundTask.this.compute(); } }); } protected final void done() { GuiExecutor.instance().execute(new Runnable() { public void run() { V value = null; Throwable thrown = null; boolean cancelled = false; try { value = get(); } catch (ExecutionException e) { thrown = e.getCause(); } catch (CancellationException e) { cancelled = true; } catch (InterruptedException consumed) { } finally { onCompletion(value, thrown, cancelled); // 任務完成的通知 } } }); } } protected void setProgress(final int current, final int max) { GuiExecutor.instance().execute(new Runnable() { public void run() { onProgress(current, max); } }); } // 在後臺執行緒中被取消 protected abstract V compute() throws Exception; // 在事件執行緒中被取消 protected void onCompletion(V result, Throwable exception, boolean cancelled) { } protected void onProgress(int current, int max) { }// Future的其他方法}

compute方法可以呼叫setProgress方法以數字形式來指示進度。因而在事件執行緒中呼叫onProgress,從而更新使用者介面以顯示視覺化的進度資訊。

要想實現BackgroundTask,你只需要實現compute,該方法將在後臺執行緒中呼叫。也可以改寫onCompletion和onProgress,這兩個方法也會在事件執行緒中呼叫。

// 通過BackgroundTask來執行長時間的並且可取消的任務startButton.addActionListener(new ActionListener() { public void actionPerformed(ActionEvent e) { class CancelListener implements ActionListener { BackgroundTask<?> task; public void actionPerformed(ActionEvent event) { if(task!=null) task.cancel(true); } } final CancelListener listener = new CancelListener(); listener.task = new BackgroundTask<Void>() { // V = Void public Void compute() { while(moreWork() &;&; !isCancelled()) doSomeWork(); return null; } public void onCompletion(boolean cancelled, String s, Throwable exception) { cancelButton.removeActionListener(listener); // 表示任務完成,再點選cancel按鈕就無效了 label.setText("done"); } }; cancelButton.addActionListener(listener); // 這個操作放在startButton事件處理器裡,表示在點選start按鈕之前,點選cancel按鈕是無效的 backgroundExec.execute(listener.task); }}); 9.4 共享資料模型
  1. 最簡單的情況是,資料模型中的資料由使用者來輸入或者由應用程式來啟動時靜態地從檔案或其他資料來源載入。在這種情況下,除了事件執行緒之外的任何執行緒都不可能訪問到資料。但在某些情況下,表現模型物件只是一個數據源(例如資料庫、檔案系統或遠端服務等)的檢視物件。這時,當資料在應用程式中進出時,有多個執行緒都可以訪問這些資料。
  2. 正確的做法是,當樹節點被展開時才讀取相應的內容。即使只列舉遠端捲上的單個目錄也可能花費很長的時間,因此你可以考慮在後臺任務中執行列舉操作。當後臺任務完成後,必須通過某種方式將資料填充到樹形模型中。可以使用執行緒安全的樹形模型來實現這個功能:通過invokeLater提交一個任務,將資料從後臺任務中“推入”事件執行緒,或者讓事件執行緒通過輪詢來檢視是否有資料可用。
9.4.1 執行緒安全的資料模型
  1. 如果資料模型支援細粒度的併發,那麼事件執行緒和後臺執行緒就能共享該資料模型,而不會發生響應性問題。例如,第5章的DelegatingVehicleTracker在底層使用了一個ConcurrentHashMap來提供高度併發的讀寫操作。這種方法的缺點在於,ConcurrentHashMap無法提供一致的資料快照,而這可能是需求的一部分。執行緒安全的資料模型必須在更新模板時產生事件,這樣檢視才能在資料發生變化後進行更新。
  2. 然而,只有在遍歷操作遠遠多於修改操作時,“寫時拷貝”容器才能提供更好的效能,例如在車輛追蹤應用程式中就不適合採用這種方法。一些特定的資料結構或許可以避免這種限制,但要構建一個既能提供高效的併發訪問又能在舊資料無效後不再維護它們的資料結構卻並不容易,因此只有其他方法都行不通後才應該考慮使用它。
9.4.2 分解資料模型
  1. 在分解模型設計中,表現模型被封閉在事件執行緒中,而其他模型,即共享模型,是執行緒安全的,因此既可以由事件執行緒方法,也可以由應用程式執行緒訪問。表現模型會註冊共享模型的監聽器,從而在更新時得到通知。然後,表示模型可以在共享模型中得到更新:通過將相關狀態的快照嵌入到更新訊息中,或者由表現模型在收到更新事件時直接從共享模型中獲取資料。
  2. 如果資料模型很大,或者更新頻率極高,在分解模型包含的資訊中有一方或雙方對另一方不可見,那麼更高效的方式是傳送增量更新資訊而不是傳送一個完整的快照。
第10章 避免活躍性危險

在安全性與活躍性之間通常存在著某種制衡。我們使用加鎖機制來確保執行緒安全,但如果過度地使用加鎖,則可能導致鎖順序死鎖(Lock-Ordering Deadlock)。同樣,我們使用執行緒池和訊號量來限制對資源的使用,但這些被限制的行為可能會導致資源死鎖(Resource Deadlock)。

10.1 死鎖
  1. 在資料庫系統的設計中考慮了監測死鎖以及從死鎖中恢復。在執行一個事務(Transaction)時可能需要獲取多個鎖,並一直持有這些鎖直到事務提交。因此在兩個事務之間很可能發生死鎖,但事實上這種情況並不多見。如果沒有外部干涉,那麼這些事務將永遠等待下去(在某個事務中持有的鎖可能在其他事務中也需要)。但資料庫伺服器不會讓這種情況發生。當它檢測到一組事務發生了死鎖時(通過在表示等待關係的有向圖中搜索迴圈),將選擇一個犧牲者並放棄這個事務。作為犧牲者的事務會釋放它所持有的資源,從而使其他事務繼續進行。應用程式可以重新執行被強制中止的事務,而這個事務現在可以成功完成,因為所有跟它競爭資源的事務都已經完成了。
  2. 與許多其他的併發危險一樣,死鎖造成的影響很少會立即顯現出來。如果一個類可能發生死鎖,那麼並不意味著每次都會發生死鎖,而只是表示有可能。當死鎖出現時,往往是在最糟糕的時候——在高負載情況下。
10.1.1 鎖順序死鎖
  1. 如果每個需要鎖L和鎖M的執行緒都以相同的順序來獲取L和M,那麼就不會發生死鎖了。
10.1.2 動態的鎖順序死鎖
  1. 這種死鎖可以採用檢視是否存在巢狀的鎖獲取操作的方法來檢查。由於我們無法控制引數的順序,因此要解決這個問題,必須定義鎖的順序,並在整個應用程式中都按照這個順序來獲取鎖。
  2. 在制定鎖的順序時,可以使用System.identityHashCode方法,該方法將返回由Object.hashCode返回的值。
// 通過鎖順序來避免死鎖private static final Object tieLock = new Object();public void transferMoney(final Account fromAcct, final Account toAcct, final DollarAmount amount) throws InsufficientFundsException { class Helper { public void transfer() throws InsufficientFundsException { if(fromAcct.getBalance().compareTo(amount)<0) throw new InsufficientFundsException(); else { fromAcct.debit(amount); toAcct.credit(amount); } } } int fromHash = System.identityHashCode(fromAcct); int toHash = System.identityHashCode(toAcct); if (fromHash<toHash) { synchronized(fromAcct) { synchronized(toAcct) { new Helper().transfer(); } } } else if (fromHash>toHash) { synchronized(toAcct) { synchronized(fromAcct) { new Helper().transfer(); } } } else { synchronized(tieLock) { synchronized(fromAcct) { synchronized(toAcct) { new Helper().transfer(0; } } } }}

在極少數情況下,兩個物件可能擁有相同的雜湊值,此時必須通過某種任意的方法來決定鎖的順序,而這可能又會重新引入死鎖。為了避免這種情況,可以使用“加時賽(Tie-Breaking)”鎖。在獲得兩個Account鎖之前,首先獲得這個“加時賽”鎖,從而保證每次只有一個執行緒以未知的順序獲得這兩個鎖,從而消除了死鎖發生的可能性(只要一致地使用這種機制)。如果經常會出現雜湊衝突的情況,那麼這種技術可能會成為併發性的一個瓶頸(這類似於在整個程式中只有一個鎖的情況),但由於System.identityHashCode中出現雜湊衝突的頻率非常低,因此這項技術以最小的代價,換來了最大的安全性。

如果在Account中包含一個唯一的、不可變的,並且具備可比性的鍵值,例如賬號,那麼要制定鎖的順序就更加容易了:通過鍵值對物件進行排序,因而不需要使用“加時賽”鎖。

10.1.3 在協作物件之間發生的死鎖// 在相互協作物件之間的鎖順序死鎖(不要這麼做)// 注意:容易發生死鎖!class Taxi { @GuardedBy("this") private Point location, destination; private final Dispatcher dispatcher; public Taxi(Dispatcher dispatcher) { this.dispatcher = dispatcher; } public synchronized Point getLocation() { return location; } public synchronized void setLocation(Point location) { // 首先要獲取this這個Taxi物件的鎖 this.location = location; if (location.equals(destination)) // notifyAvailable是外部方法,且是synchronized方法,所以要獲取dispatcher物件的鎖 dispatcher.notifyAvailable(this); }}class Dispatcher { @GuardedBy("this") private final Set<Taxi> taxis; @GuardedBy("this") private final Set<Taxi> availableTaxis; public Dispatcher() { taxis = new HashSet<Taxi>(); availableTaxis = new HashSet<Taxi>(); } public synchronized void notifyAvailable(Taxi taxi) { availableTaxis.add(taxi); } public synchronized Image getImage() { // 首先要獲取this這個Dispatcher物件的鎖 Image image = new Image(); for (Taxi t : taxis) image.drawMarker(t.getLocation()); // getLocation是個外部方法,且是synchronized的,所以要獲取每個t(Taxi)的鎖。於是getImage和setLocation就會死鎖了。 return image; }}

因為setLocation和notifyAvailable都是同步方法,因此呼叫setLocation的執行緒將首先獲取Taxi的鎖,然後獲取Dispatcher的鎖。同樣,呼叫getImage的執行緒將首先獲取Dispatcher的鎖,然後再獲取每一個Taxi的鎖(每次獲取一個)。

然而要在Taxi和Dispatcher中查詢死鎖則比較困難:如果在持有鎖的情況下呼叫某個外部方法,那麼就需要警惕死鎖。

如果在持有鎖時呼叫某個外部方法,那麼將出現活躍性問題。在這個外部方法中可能會獲取其他鎖(這可能會產生死鎖),或者阻塞時間過長,導致其他執行緒無法及時獲得當前被持有的鎖。

10.1.4 開放呼叫

這需要使同步程式碼塊僅被用於保護那些涉及共享狀態的操作。通常,如果只是為了語法緊湊或簡潔性(而不是因為整個方法必須通過一個鎖來保護)而使用同步方法(而不是同步程式碼塊),那麼就會導致上面的死鎖。

// 通過公開呼叫來避免在相互協作的物件之間產生死鎖@ThreadSafeclass Taxi { @GuardedBy("this") private Point location, destination; private final Dispatcher dispatcher; ... public synchronized Point getLocation() { // location是共享狀態 return location; } public void setLocation(Point location) { boolean reachedDestination; // 儲存location.equals(destination)這個共享狀態 synchronized(this) { // 兩個共享狀態 this.location = location; reachedDestination = location.equals(destination); } if (reachedDestination) dispatcher.notifyAvailable(this); }}@ThreadSafeclass Dispatcher { @GuardedBy("this") private final Set<Taxi> taxis; @GuardedBy("this") private final Set<Taxi> availableTaxis; ... public synchronized void notifyAvailable(Taxi taxi) { // 修改availableTaxis這個共享狀態 availableTaxis.add(taxi); } public Image getImage() { Set<Taxi> copy; synchronized(this) { // 複製的意思是,計程車就是這麼些計程車了,但是它們的位置還可以變 copy = new HashSet<Taxi>(taxis); } Image image = new Image(); for (Taxi t : copy) image.drawMarker(t.getLocation()); return image; }}

有時候,在重新編寫同步程式碼塊以使用開發呼叫時會產生意想不到的結果,因為這會使得某個原子操作變為非原子操作。在許多情況下,使某個操作失去原子性是可以接受的。例如,對於兩個操作:更新出租車位置以及通知排程程式這輛計程車已準備好出發去一個新的目的地,這兩個操作並不需要實現為一個原子操作。在其他情況下,雖然去掉原子性可能會出現一些值得注意的結果,但這種語義變化仍然是可以接受的。在容易產生死鎖的版本中,getImage會生成某個時刻下的整個車隊位置的完整快照,而在重新改寫的版本中,getImage將獲得每輛計程車不同時刻的位置。

例如,在關閉某個服務時,你可能希望所有正在執行的操作執行完成以後,再釋放這些服務佔用的資源。如果在等待操作完成的同時持有該服務的鎖,那麼將很容易導致死鎖,但如果在服務關閉之前就釋放服務的鎖,則可能導致其他執行緒開始新的操作。

這個問題的解決方法是,在將服務的狀態更新為“關閉”之前一直持有鎖,這樣其他想要開始新操作的執行緒,包括想關閉該服務的其他執行緒,會發現服務已經不可用,因此也就不會試圖開始新的操作。然後,你可以等待關閉操作結束,並且知道當開放呼叫完成後,只有執行關閉操作的執行緒才能訪問服務的狀態。因此,這項技術依賴於構造一些協議(而不是通過加鎖)來防止其他執行緒進入程式碼的臨界區。

10.1.5 資源死鎖
  1. 如果某些任務需要等待其他任務的結果,那麼這些任務往往是產生執行緒飢餓死鎖的主要來源,有界執行緒池/資源池與相互依賴的任務不能一起使用。
10.2 死鎖的避免與診斷 10.2.1 支援定時的鎖
  1. 還有一項技術可以檢測死鎖和從死鎖中恢復過來,即顯式使用Lock類中的定時tryLock功能來代替內建鎖機制。
  2. 這項技術只有在同時獲取兩個鎖時才有效,如果在巢狀的方法呼叫中請求多個鎖,那麼即使你知道已經持有了外層的鎖,也無法釋放它。
10.2.2 通過執行緒轉儲資訊來分析死鎖
  1. 執行緒轉儲包括各個執行中的執行緒的棧追蹤資訊,這類似於發生異常時的棧追蹤資訊。執行緒轉儲還包含加鎖資訊,例如每個執行緒持有了哪些鎖,在哪些棧幀中獲得這些鎖,以及被阻塞的執行緒正在等待獲取哪一個鎖。在生成執行緒轉儲之前,JVM將在等待關係圖中通過搜尋迴圈來找出死鎖。如果發現了一個死鎖,則獲取相應的死鎖資訊,例如在死鎖中涉及哪些鎖和執行緒,以及這個鎖的獲取操作位於程式的哪些位置。
  2. 要在UNIX平臺上觸發執行緒轉儲操作,可以通過向JVM的進城傳送SGIQUIT訊號(kill -3),或者在UNIX平臺中按下Ctrl-/鍵,在Windows平臺中按下Ctrl-Break鍵。在許多IDE中都可以請求執行緒轉儲。
  3. 如果使用顯式的Lock類而不是內部鎖,那麼Java 5.0並不支援與Lock相關的轉儲資訊,線上程轉儲中不會出現顯式的Lock。雖然Java 6中包含對顯式Lock的執行緒轉儲和死鎖檢測等的支援,但在這些鎖上獲得的資訊比在內建鎖上獲得的資訊精確度低。內建鎖與獲得它們所在的執行緒幀是相關聯的,而顯式的Lock只與獲得它的執行緒相關聯。
10.3 其他活躍性危險 10.3.1 飢餓
  1. 引發飢餓的最常見資源就是CPU時鐘週期。
  2. 要避免使用執行緒優先順序,因為這會增加平臺依賴性,並可能導致活躍性問題。在大多數併發應用程式中,都可以使用預設執行緒優先順序。
10.3.2 糟糕的響應性
  1. 但CPU密集型的後臺任務仍然可能對響應性造成影響,因為它們會與事件執行緒共同競爭CPU的時鐘週期。
10.3.3 活鎖
  1. 活鎖(Livelock)是另一種形式的活躍性問題,該問題儘管不會阻塞執行緒,但也不能繼續執行,因為執行緒將不斷重複執行相同的操作,而且總會失敗。活鎖通常發生在處理事務訊息的應用程式中:如果不能成功地處理某個訊息,那麼訊息處理機制將回滾整個事務,並將它重新放到佇列的開頭。
  2. 這種形式的活鎖通常是由過度的錯誤恢復程式碼造成的,因為它錯誤地將不可修復的錯誤作為可修復的錯誤。
  3. 當多個相互協作執行緒都對彼此進行響應從而修改各自的狀態,並使得任何一個執行緒都無法繼續執行時,就發生了活鎖。這就像兩個過於禮貌的人在半路上面對面地相遇:他們彼此都讓出對方的路,然而又在另一條路上相遇了。因此他們就這樣反覆地避讓下去。
  4. 要解決這種活鎖問題,需要在重試機制中引入隨機性。
  5. 以太協議定義了在重複發生衝突時採用指數方式回退機制,從而降低在多臺存在衝突的機器之間發生擁塞和反覆失敗的風險。
第11章 效能與可伸縮性 11.1 對效能的思考
  1. 造成這些開銷的操作包括:執行緒之間的協調(例如加鎖、觸發訊號以及記憶體同步等),增加的上下文切換,執行緒的建立和銷燬,以及執行緒的排程等。
  2. 要想通過併發來獲得更好的效能,需要努力做好兩件事情:更有效地利用現有處理資源,以及在出現新的處理資源時使程式儘可能地利用這些新資源。
11.1.1 效能與可伸縮性
  1. 可伸縮性指的是:當增加計算資源時(例如CPU、記憶體、儲存容量或I/O頻寬),程式的吞吐量或者處理能力能相應地增加。
  2. 在併發應用程式中針對可伸縮性設計和調整時所採用的方法與傳統的效能調優方法截然不同。當進行效能調優時,其目的通常是用更小的代價完成相同的工作,例如通過快取來重用之前計算的結果,或者採用時間複雜度為 O(n2) 演算法來代替複雜度為 O(nlogn) 的演算法。在進行可伸縮性調優時,其目的是設法將問題的計算並行化,從而能利用更多的計算資源來完成更多的任務。
  3. 我們熟悉的三層程式模型,即在模型中的表現層、業務邏輯層和持久化層是彼此獨立的,並且可能由不同的系統來處理,這很好地說明了提高可伸縮性通常會造成效能損失的原因。如果把表現層、業務邏輯層和持久化層都融合到單個應用程式中,那麼在處理第一個工作單元時,其效能肯定要高於將應用程式分為多層並將不同層次分佈到多個系統時的效能。這種單一的應用程式避免了在不同層次之間傳遞任務時存在的網路延遲,同時也不需要將計算過程分解到不同的抽象層次,因此能減少許多開銷(例如在任務排隊、執行緒呼叫以及資料複製時存在的開銷)。
  4. 然而,當這種單一的系統到達自身處理能力的極限時,會遇到一個嚴重的問題:要進一步提升它的處理能力將非常困難。因此,我們通常會接受每個工作單元執行更長的時間或消耗更多的計算資源,以換取應用程式在增加更多資源的情況下處理更高的負載。
11.1.2 評估各種效能權衡因素
  1. 例如,“快速排序”演算法在大規模資料集上的執行效率非常高,但對於小規模的資料集來說,“氣泡排序”實際上更高效。如果要實現一個高效的排序演算法,那麼需要知道被處理資料集的大小,還有衡量優化的指標,包括:平均計算時間、最差時間、可預知性。然而,編寫某個庫中排序演算法的開發人員通常無法知道這些需求資訊。這就是為什麼大多數優化措施都不成熟的原因之一:它們通常無法獲得一組明確的需求。
  2. 很多效能優化措施通常都是以犧牲可讀性或可維護性為代價——程式碼越“聰明”或越“晦澀”,就越難以理解和維護。有時候,優化措施會破壞面向物件的設計原則,例如需要打破封裝,有時候,它們又會帶來更高的錯誤風險,因為通常越快的演算法就越複雜。
  3. 在實現這種效能提升時需要付出哪些隱含的代價,例如增加開發風險或維護開銷?這種權衡是否合適?
  4. 以測試為基準,不要猜測。
  5. 例如,免費的perfbar應用程式可以給出CPU的忙碌程度資訊,而我們通常的目標就是使CPU保持忙碌狀態,因此這個功能可以有效地評估是否需要進行效能調優或者已實現的調優效果如何。
11.2 Amdahl定律
  1. 而有些任務本質上是序列的,例如,即使增加再多的工人也不可能增加作物的生長速度。
  2. 假定F是必須被序列執行的部分,那麼根據Amdahl定律,在包含N個處理器的機器中,最高的加速比為: Speedup≤1F+1−FN
  3. 當N趨近無窮大時,最大的加速比趨近於1/F。因此,如果程式有50%的計算需要序列執行,那麼最高的加速比只能是2(而不管有多少個執行緒可用);如果程式中有10%的計算需要序列執行,那麼最高的加速比將接近10。
  4. 隨著處理器數量的增加,可以很明顯地看到,即使序列部分所佔的百分比很小,也會極大地限制當增加計算資源時能夠提升的吞吐率。
  5. 然而,這個過程中包含了一個序列部分——從佇列中獲取任務。所有工作者執行緒都共享同一個工作佇列,因此在對該佇列進行併發訪問時需要採用某種同步機制來維持佇列的完整性。
  6. 如果使用LinkedBlockingQueue作為工作佇列,那麼出列操作被阻塞的可能性將小於使用同步LinkedList時發生阻塞的可能性,因為LinkedBlockingQueue使用了一種可伸縮性更高的演算法。
  7. 這個示例還忽略了另一種常見的序列操作:對結果進行處理。所有有用的計算都會生成某種結果或者產生某種效應——如果不會,那麼可以將它們作為“死亡程式碼”刪除掉。由於Runnable沒有提供明確的結果處理過程,因此這些任務一定會產生某種效果,例如將它們的結果寫入到日誌或者儲存到某個資料結構。通常,日誌檔案和結果容器都會由多個工作者執行緒共享,並且這也是一個序列部分。如果所有執行緒都將各自的計算結果儲存到自行維護資料結構中,並且在所有任務都執行完成後再合併所有的結果,那麼這種合併操作也是一個序列部分。
  8. 在所有併發程式中都包含一些序列部分。
11.2.1 示例:在各種框架中隱藏的序列部分
  1. 吞吐量的差異來源於兩個佇列中不同比例的序列部分。同步的LinkedList採用單個鎖來保護整個佇列的狀態,並且在offer和remove等方法的呼叫期間都將持有這個鎖。ConcurrentLinkedQueue使用了一種更復雜的非阻塞佇列演算法,該演算法使用原子引用來更新各個連結指標。在第一個佇列中,整個的插入或刪除操作都將序列執行,而在第二個佇列中,只有對指標的更新操作需要序列執行。
11.2.2 Amdahl定律的應用
  1. 在評估一個演算法時,要考慮演算法在數百個或數千個處理器的情況下的效能表現,從而對可能出現的可伸縮性侷限有一定程度的認識。例如,兩種降低鎖粒度的技術:鎖分解(將一個鎖分解為兩個鎖)和鎖分段(把一個鎖分解為多個鎖)。當通過Amdahl定律來分析這兩項技術時,我們會發現,如果將一個鎖分解為兩個鎖,似乎並不能充分利用多處理器的能力。鎖分段技術似乎更有前途,因為分段的數量可隨著處理器數量的增加而增加。
11.3 執行緒引入的開銷 11.3.1 上下文切換
  1. 切換上下文需要一定的開銷,而線上程排程過程中需要訪問由作業系統和JVM共享的資料結構。應用程式、作業系統以及JVM都使用一組相同的CPU。在JVM和作業系統的程式碼中消耗越多的CPU時鐘週期,應用程式的可用CPU時鐘週期就越少。但上下文切換的開銷並不只包含JVM和作業系統的開銷。當一個新的執行緒被切換進來時,它所需要的資料可能不在當前處理器的本地快取中,因此上下文切換將導致一些快取缺失,因而執行緒在首次排程執行時會更加緩慢。這就是為什麼排程器會為每個可執行的執行緒分配一個最小執行時間,即使有許多其他的執行緒正在等待執行:它將上下文切換的開銷分攤到更多不會中斷的執行時間上,從而提高整體的吞吐量(以損失響應性為代價)。
  2. 上下文切換的實際開銷會隨著平臺的不同而變化,然而按照經驗來看:在大多數通用的處理器中,上下文切換的開銷相當於5000~10000個時鐘週期,也就是幾微秒。
  3. UNIX系統的vmstat命令和Windows系統的perfmon工具都能報告上下文切換次數以及在核心中執行時間所佔比例等資訊。如果核心佔用率較高(超過10%),那麼通常表示排程活動發生得很頻繁,這很可能是由I/O或競爭鎖導致的阻塞引發的。
11.3.2 記憶體同步
  1. 在synchronized和volatile提供的可見性保證中可能會使用一些特殊指令,即記憶體柵欄(Memory Barrier)。記憶體柵欄可以重新整理快取,使快取無效,重新整理硬體的寫緩衝,以及停止執行管道。
  2. 在記憶體柵欄中,大多數操作都是不能被重排序的。
  3. 如果有一個鎖物件只能由當前執行緒訪問,那麼JVM就可以通過優化去掉這個鎖獲取操作,因為另一個執行緒無法與當前執行緒在這個鎖上發生同步。例如,JVM通常都會去掉下面的鎖獲取操作:
synchronized (new Object()) {// 執行一些操作......}

一些更完備的JVM能通過逸出分析(Escape Analysis)來找出不會發布到堆的本地物件引用(因此這個引用是執行緒本地的)。

即使不進行逸出分析,編譯器也可以執行鎖粒度粗化(Lock Coarsening)操作,即將鄰近的同步程式碼塊用同一個鎖合併起來。

不要過度擔心非競爭同步帶來的開銷。這個基本的機制已經非常快了,並且JVM還能進行額外的優化以進一步降低或消除開銷。因此,我們應該將優化重點放在那些發生鎖競爭的地方。

同步會增加共享記憶體總線上的通訊量,匯流排的頻寬是有限的,並且所有的處理器都共享這條匯流排。如果有多個執行緒競爭同步頻寬,那麼所有使用了同步的執行緒都會受到影響。

11.3.3 阻塞
  1. 非競爭的同步可以完全在JVM中進行處理,而競爭的同步可能需要作業系統的介入,從而增加開銷。
  2. 如果等待時間較短,則適合採用自旋等待方式,而如果等待時間較長,則適合採用執行緒掛起方式。
11.4 減少鎖的競爭


  1. 我們已經看到,序列操作會降低可伸縮性,並且上下文切換也會降低效能。在鎖上發生競爭時將同時導致這兩種問題,因此減少鎖的競爭能夠提高效能和可伸縮性。
  2. 在併發程式中,對可伸縮性的最主要威脅就是獨佔方式的資源鎖。
  3. 有兩個因素將影響在鎖上發生競爭的可能性:鎖的請求頻率,以及每次持有該鎖的時間。

有3種方式可以降低鎖的競爭程度:
減少鎖的持有時間。
降低鎖的請求頻率。
使用帶有協調機制的獨佔鎖,這些機制允許更高的併發性。
11.4.1 縮小鎖的範圍(“快進快出”)
  1. 降低發生競爭可能性的一種有效方式就是儘可能縮短鎖的持有時間。例如,可以將一些與鎖無關的程式碼移出同步程式碼塊,尤其是那些開銷較大的操作,以及可能被阻塞的操作,例如I/O操作。
  2. 由於在AttributeStore中只有一個狀態變數attributes,因此可以通過將執行緒安全性委託給其他的類來進一步提升它的效能。通過用執行緒安全的Map(Hashtable、synchronizedMap或ConcurrentHashMap)來代替attributes,AttributeStore可以將確保執行緒安全性的任務委託給頂層的執行緒安全容器來實現。這樣就無須在AttributeStore中採用顯式的同步,縮小在訪問Map期間鎖的範圍,並降低了將來的程式碼維護者無意破壞執行緒安全性的風險(例如在訪問attributes之前忘記獲得相應的鎖)。
  3. 在分解同步程式碼塊時,理想的平衡點將與平臺相關,但在實際情況中,僅當可以將一些“大量”的計算或阻塞操作從同步程式碼塊中移出時,才應該考慮同步程式碼塊的大小。
11.4.2 減小鎖的粒度
  1. 這可以通過鎖分解和鎖分段等技術來實現,在這些技術中將採用多個相互獨立的鎖來保護獨立的狀態變數,從而改變這些變數在之前由單個鎖來保護的情況。這些技術能減小鎖操作的粒度,並能實現更高的可伸縮性,然而,使用的鎖越多,那麼發生死鎖的風險也就越高。
  2. 如果一個鎖需要保護多個相互獨立的狀態變數,那麼可以將這個鎖分解為多個鎖,並且每個鎖只保護一個變數,從而提高可伸縮性,並最終降低每個鎖被請求的頻率。
// 將ServerStatus重新改寫為使用鎖分解技術@ThreadSafepublic class ServerStatus { @GuardedBy("users") public final Set<String> users; @GuardedBy("queries") public final Set<String> queries; ... public void addUser(String u) { synchronized(users) { users.add(u); } } public void addQuery(String q) { synchronized(queries) { queries.add(q); } } // 去掉同樣被改寫為使用被分解鎖的方法} 11.4.3 鎖分段
  1. 在某些情況下,可以將鎖分解技術進一步擴充套件為對一組獨立物件上的鎖進行分解,這種情況被稱為鎖分段。例如,在ConcurrentHashMap的實現中使用了一個包含16個鎖的陣列,每個鎖保護所有雜湊桶的1/16,其中第N個雜湊桶由第(N mod 16)個鎖來保護。假設雜湊函式具有合理的分佈性,並且關鍵字能夠實現均勻分佈,那麼這大約能把對於鎖的請求減少到原來的1/16。正是這項技術使得ConcurrentHashMap能夠支援多達16個併發的寫入器。(要使得擁有大量處理器的系統在高訪問量的情況下實現更高的併發性,還可以進一步增加鎖的數量,但僅當你能證明併發寫入執行緒的競爭足夠激烈並需要突破這個限制時,才能將鎖分段的數量超過預設的16個。)
  2. 鎖分段的一個劣勢在於:與採用單個鎖來實現獨佔訪問相比,要獲取多個鎖來實現獨佔訪問將更加困難並且開銷更高。通常,在執行一個操作時最多隻需獲取一個鎖,但在某些情況下需要加鎖整個容器,例如當ConcurrentHashMap需要擴充套件對映範圍,以及重新計算鍵值的雜湊值要分佈到更大的桶集合中時,就需要獲取分段鎖集合中的所有的鎖。
  3. 它擁有N_LOCKS個鎖,並且每個鎖保護雜湊桶的一個子集。大多數方法,例如get,都只需要獲得一個鎖,而有些方法則需要獲得所有的鎖,但並不要求同時獲得,例如clear方法的實現。
// 在基於雜湊的Map中使用鎖分段技術@ThreadSafepublic class StripedMap {// 同步策略:buckets[n]由locks[n%N_LOCKS]來保護 private static final int N_LOCKS = 16; private final Node[] buckets; private final Object[] locks; private static class Node { ... } public StripedMap(int numBuckets) { buckets = new Node[numBuckets]; locks = new Object[N_LOCKS]; for(int i=0; i<N_LOCKS; i++) locks[i] = new Object(); } private final int hash(Object key) { return Math.abs(key.hashCode() % buckets.length); } public Object get(Object key) { int hash = hash(key); synchronized(locks[hash % N_LOCKS]) { for (Node m=buckets[hash]; m!=null; m=m.next) if(m.key.equals(key)) return m.value; } return null; } public void clear() { for(int i=0; i<buckets.length; i++) { synchronized(locks[i % N_LOCKS]) { buckets[i] = null; } } } ...} 11.4.4 避免熱點域
  1. 如果程式採用鎖分段技術,那麼一定要表現出在鎖上的競爭頻率高於在鎖保護的資料上發生競爭的頻率。
  2. 即使使用鎖分段技術來實現雜湊鏈,那麼在對計數器的訪問進行同步時,也會重新導致在使用獨佔鎖時存在的可伸縮性問題。一個看似效能優化的措施——快取size操作的結果,已經變成了一個可伸縮性問題。在這種情況下,計數器也被稱為熱點域,因為每個導致元素數量發生變化的操作都需要訪問它。
  3. 為了避免這個問題,ConcurrentHashMap中的size將對每個分段進行列舉並將每個分段中的元素數量相加,而不是維護一個全域性計數。為了避免列舉每個元素,ConcurrentHashMap為每個分段都維護了一個獨立的計數,並通過每個分段的鎖來維護這個值。
11.4.5 一些替代獨佔鎖的方法
  1. 原子變數提供了一種方式來降低更新“熱點域”時的開銷,例如靜態計數器、序列發生器、或者對連結串列資料結構中頭節點的引用。原子變數類提供了在整數或者物件引用上的細粒度原子操作(因此可伸縮性更高),並使用了現代處理器中提供的底層併發原語(例如比較並交換[compare-and-swap])。如果在類中只包含少量的熱點域,並且這些域不會與其他變數參與到不變性條件中,那麼用原子變數來替代它們能提高可伸縮性。
11.4.6 監測CPU的利用率
  1. 不均勻的利用率表明大多數計算都是由一小組執行緒完成的,並且應用程式沒有利用其他的處理器。
  2. 在vmstat命令的輸出中,有一欄資訊是當前處於可執行狀態但並沒有執行的執行緒數量。如果CPU的利用率很高,並且總會有可執行的執行緒在等待CPU,那麼當增加更多的處理器時,程式的效能可能會得到提升。
11.4.7 向物件池說“不”
  1. 事實上,現在Java的分配操作已經比C語言的malloc呼叫更快:在Hotspot 1.4.x和5.0中,“new Object”的程式碼大約只包含10條機器指令。
  2. 除了損失CPU指令週期外,在物件池技術中還存在一些其他問題,其中最大的問題就是如何正確地設定物件池的大小(如果物件池太小,那麼將沒有作用,而如果太大,則會對垃圾收集器帶來壓力,因為過大的物件池將佔用其他程式需要的記憶體資源)。
  3. 通常,物件分配操作的開銷比同步的開銷更低。
11.5 示例:比較Map的效能
  1. 在單執行緒環境下,ConcurrentHashMap的效能比同步的HashMap的效能略好一些,但在併發環境中則要好得多。
  2. ConcurrentHashMap和ConcurrentSkipListMap的資料顯示,它們線上程數量增加時能表現出很好的可伸縮性,並且吞吐量會隨著執行緒數量的增加而增加。雖然圖中的執行緒數量並不大,但與普通的應用程式相比,這個測試程式在每個執行緒上生成了更多的競爭,因為它除了向Map施加壓力外幾乎沒有執行任何其他操作,而實際的應用程式通常會在每次迭代中進行一些執行緒本地工作。
11.6 減少上下文切換的開銷
  1. 當任務在執行和阻塞這兩個狀態之間轉換時,就相當於一次上下文切換。在伺服器應用程式中,發生阻塞原因之一就是在處理請求時產生各種日誌訊息。
  2. 日誌操作的服務時間包括與I/O流類相關的計算時間,如果I/O操作被阻塞,那麼還會包括執行緒被阻塞的時間。作業系統將這個被阻塞的執行緒從排程佇列中移走並直到I/O操作結束,這將比實際阻塞的時間更長。當I/O操作結束時,可能有其他執行緒正在執行它們的排程時間片,並且在排程佇列中有些執行緒位於被阻塞執行緒之前,從而進一步增加服務時間。如果有多個執行緒在同時記錄日誌,那麼還可能在輸出流的鎖上發生競爭,這種情況的結果與阻塞I/O的情況一樣——執行緒被阻塞並等待鎖,然後被執行緒排程器交換出去。在這種日誌操作中包含了I/O操作和加鎖操作,從而導致上下文切換次數的增多,以及服務時間的增加。
  3. 通過將I/O操作從處理請求的執行緒中分離出來,可以縮短處理請求的平均服務時間。呼叫log方法的執行緒將不會再因為等待輸出流的鎖或者I/O完成而被阻塞,它們只需將訊息放入佇列,然後就返回各自的任務中。另一方面,雖然在訊息佇列上可能發生競爭,但put操作相對於記錄日誌的I/O操作(可能需要執行系統呼叫)是一種更為輕量級的操作,因此在實際使用中發生阻塞的概率更小(只要佇列沒有填滿)。由於發出日誌請求的執行緒現在被阻塞的概率降低,因此該執行緒在處理請求時被交換出去的概率也會降低。我們所做的工作就是把一條包含I/O操作和鎖競爭的複雜且不確定的程式碼路徑變成一條簡單的程式碼路徑。
« 上一篇
下一篇 »

相關推薦

Effective Java 第二版 中文版 筆記（三）列舉實現單例模式

 列舉實現單例模式： public enum DataSourceEnum { DATASOURCE; pr...

Effective Java 第二版 中文版 筆記（六）消除過期物件的引用

這部分沒有怎麼理解，但總體的意思就是防止記憶體洩漏 Java的垃圾回收機制： Java中的物件是在堆中分配，物件的建立...

Effective Java 第二版 中文版 筆記（8-12）避免使用終結方法

我對此部分沒有什麼興趣，因為誰會閒的沒事覆蓋java自身的方法。 八、覆蓋equals ; 這裡最重要的就是遵守約定； ...

Effective Java 第二版 中文版 筆記（十八）介面優於抽象類

這裡主要就是介面和抽象類的區別 ： 使用一個例子直接進行說明； 理解：抽象類：抽象類是對一種事物的抽象，即對類抽象   ...

Effective Java 第二版 中文版 筆記（二十三）請不要在新程式碼中使用原生態型別

泛型：宣告一個或多個型別的引數的類和介面； 原生態型別如 List:不帶任何型別引數的泛型名稱； public class...

Effective Java 第二版 中文版 筆記（二十五）列表優於陣列

不知道說的什麼，總之是一般情況下都是用List。 陣列、List和ArrayList的區別         陣...

Java Concurrency in Practice中物件鎖重入問題的理解

原因：Java Concurrency in Practice 中文版21頁講解了關於物件鎖的重入的問題，一直沒...

《Effective Java》Second Edition中文版筆記（完整）

第2章 建立和銷燬物件 第1條：考慮用靜態工廠方法代替構造器 一個類只能有一個帶有指定簽名的構造器。程...

JavaScript權威指南(第6版)(中文版)筆記

text image png 指南 mage 分享 http javascrip p s ?JavaScript權威指南...

吳恩達深度學習中文版筆記：人臉識別和神經風格轉換

...