《Java并發編程實戰》- 個人閱讀總結（一）

2019-11-06 06:25:40

字體：大中小

來源：轉載

供稿：網友

背景

前段時間想更深入了解下java多線程相關的知識，對Java多線程有一個全面的認識，所以想找一本Java多線程相關的書籍來閱讀，最后我選擇了《Java并發編程實戰》這本個人認為還算相當不錯，至于為什么選擇它，下面有介紹。

書的介紹

中文書名：《Java并發編程實戰》英文書名：《Java Concurrency in PRactice》作者：Brian Goetz / Tim Peierls / Joshua Bloch / Joseph Bowbeer / David Holmes / Doug Lea 譯者：童云蘭出版社: 機械工業出版社華章公司購買可以在各大電商網站搜索書名購買，請支持正版圖書封面

選擇該書原因

1、亞馬遜排名考前，評論多評分也不錯； 2、很多java程序員必看數據整理里面的書單之一； 3、豆瓣評分9.0； 4、作者都很牛B。

圖書簡介

《Java并發編程實戰》深入淺出地介紹了Java線程和并發,是一本完美的Java并發參考手冊。書中從并發性和線程安全性的基本概念出發,介紹了如何使用類庫提供的基本并發構建塊,用于避免并發危險、構造線程安全的類及驗證線程安全的規則,如何將小的線程安全類組合成更大的線程安全類,如何利用線程來提高并發應用程序的吞吐量,如何識別可并行執行的任務,如何提高單線程子系統的響應性,如何確保并發程序執行預期任務,如何提高并發代碼的性能和可伸縮性等內容,最后介紹了一些高級主題,如顯式鎖、原子變量、非阻塞算法以及如何開發自定義的同步工具類。《Java并發編程實戰》適合Java程序開發人員閱讀。

作者簡介

本書作者都是Java Community Process JSR 166專家組（并發工具）的主要成員，并在其他很多JCP專家組里任職。Brian Goetz有20多年的軟件咨詢行業經驗，并著有至少75篇關于Java開發的文章。 Tim Peierls是“現代多處理器”的典范，他在BoxPop.biz、唱片藝術和戲劇表演方面也頗有研究。Joseph Bowbeer是一個Java ME專家，他對并發編程的興趣始于Apollo計算機時代。David Holmes是《The Java Programming Language》一書的合著者，任職于Sun公司。Joshua Bloch是Google公司的首席Java架構師，《Effective Java》一書的作者，并參與著作了《Java Puzzlers》。Doug Lea 是《Concurrent Programming》一書的作者，紐約州立大學 Oswego分校的計算機科學教授。備注：縮寫的解釋如下 - Java Community Process（JCP） wiki 英文介紹 wiki 中文介紹 - Java Specification Requests（JSR）

一些需要提前掌握的基礎

進程與線程

進程與線程的解釋

線程 & 進程 & 服務器硬件之間的關系

總核數 = 物理CPU個數 X 每顆物理CPU的核數總邏輯CPU數 = 物理CPU個數 X 每顆物理CPU的核數 X 超線程數同一進程中的多條線程將共享該進程中的全部系統資源，如虛擬地址空間，文件描述符和信號處理等等。但同一進程中的多個線程有各自的調用棧（call stack），自己的寄存器環境（register context），自己的線程本地存儲（thread-local storage）。大量用戶線程復用少量的輕權進程（內核線程）進程才是程序（那些指令和數據）的真正運行實例。若干進程有可能與同一個程序相關系，且每個進程皆可以同步（循序）或異步（平行）的方式獨立運行。

進程之間切換的原理以及代價

進程狀態轉換圖進程狀態的轉換參考下面兩篇博文進程的狀態轉換 linux進程狀態解析

操作系統基本原理

了解操作系統的基本原理對多線程的理解也是至關重要的,下面的這篇博文對操作系統的闡述比較形象易懂。操作系統基本原理總結如下： - 以多進程形式，允許多個任務同時運行 - 以多線程形式，允許單個任務分成不同的部分運行 - 提供協調機制，一方面防止進程之間和線程之間產生沖突，另一方面允許進程之間和線程之間共享資源

簡介

并發簡史

早期的計算機中不包含操作系統，它們從頭到尾執行一個程序，程序可訪問計算機上面的所有資源。這樣會造成極大的資源浪費。操作系統的出現使得計算機可以每次運行多個程序，操作系統為每個進程分配資源。之所以在計算機中加入操作系統來實現多個程序通知執行，主要基于下面三個原因。 - 資源利用率 - 公平性 - 便利性

線程優勢

那么引入多線程會有哪些優勢呢？ - 發揮多處理器的強大能力 - 使建模更具簡單性 - 異步事件的簡化處理不同類型的任務、負責自己的工作流。log4j的日志異步輸出，消息監控功能、這樣的線程可以專注自己的任務。

線程帶來的風險

線程也帶來了一些風險 - 安全性問題 - 活躍性問題 - 性能問題

基礎知識

線程安全性

什么是線程安全性？該書給出了自己的定義：當多個線程訪問某個類時，這個類始終都能表現正確的行為。

原子性加鎖機制用鎖保護狀態 java中與線程安全相關的關鍵詞：synchronized、Explicit Lock。synchronized 與 Explict Lock 大部分時候可以相互替代。

對象的共享

可見性線程封閉不變性安全發布

java中與對象共享相關的關鍵詞：Volatile、ThreadLocal、Final Volatile 變量具有 synchronized 的可見性特性，但是不具備原子特性。線程為了提高效率，將某成員變量(如A)拷貝了一份（如B），線程中對A的訪問其實訪問的是B。只在某些動作時才進行A和B的同步。因此存在A和B不一致的情況。volatile就是用來避免這種情況的。volatile告訴jvm，它所修飾的變量不保留拷貝，直接訪問主內存中的（也就是上面說的A)

安全發布的常用模式

在靜態初始化函數中初始化對象將對象的引用保存到volatile類型的域或AtomicReferance中將對象的應用保存到某個正確的構造對象的final域中將對象的引用保存到一個由鎖保護的域中

對象組合

設計線程安全的類實例封閉線程安全性的委托在現有的線程安全類中添加功能將同步策略文檔化在設計線程安全類的過程中，需要包含以下三個基本要素：找出構成對象狀態的所有變量。找出約束狀態變量的不變性條件。建立對象狀態的并發訪問管理策略。

舉例

實例封閉–PersonSet@ThreadSafepublic class PersonSet { @GuardedBy("this") private final Set<Person> mySet = new HashSet<Person>(); public synchronized void addPerson(Person p) { mySet.add(p); } public synchronized boolean containsPerson(Person p) { return mySet.contains(p); } interface Person { }}線程安全性的委托-DelegatingVehicleTracker（委托成功）NumberRange（委托失?。?** * DelegatingVehicleTracker * <p/> * Delegating thread safety to a ConcurrentHashMap * * @author Brian Goetz and Tim Peierls */@ThreadSafepublic class DelegatingVehicleTracker { private final ConcurrentMap<String, Point> locations; private final Map<String, Point> unmodifiableMap; public DelegatingVehicleTracker(Map<String, Point> points) { locations = new ConcurrentHashMap<String, Point>(points); unmodifiableMap = Collections.unmodifiableMap(locations); } public Map<String, Point> getLocations() { return unmodifiableMap; } public Point getLocation(String id) { return locations.get(id); } public void setLocation(String id, int x, int y) { if (locations.replace(id, new Point(x, y)) == null) throw new IllegalArgumentException("invalid vehicle name: " + id); } // Alternate version of getLocations (Listing 4.8) public Map<String, Point> getLocationsAsStatic() { return Collections.unmodifiableMap( new HashMap<String, Point>(locations)); }}

監視器模式

監視器是操作系統實現同步的重要基礎概念，同樣它也用在JAVA的線程同步中監視器可以看做是經過特殊布置的建筑，這個建筑有一個特殊的房間，該房間通常包含一些數據和代碼，但是一次只能一個消費者(thread)使用此房間，當一個消費者(線程)使用了這個房間，首先他必須到一個大廳(Entry Set)等待，調度程序將基于某些標準(e.g. FIFO)將從大廳中選擇一個消費者(線程)，進入特殊房間，如果這個線程因為某些原因被“掛起”，它將被調度程序安排到“等待房間”，并且一段時間之后會被重新分配到特殊房間，按照上面的線路，這個建筑物包含三個房間，分別是“特殊房間”、“大廳”以及“等待房間”。簡單來說，監視器用來監視線程進入這個特別房間，他確保同一時間只能有一個線程可以訪問特殊房間中的數據和代碼。（synchronized）這里寫圖片描述

基礎構建模塊

同步容器類并發容器阻塞隊列和生產-消費者模式阻塞方法與中斷方法同步工具

構建高效可用的緩存

java中早期的同步容器有： Collections.synchronizedXxx (早期jdk）、Vector、Hashtable java5.0版本之后開始出現效率更高的并發容器： java.util.concurrent(Java5.0)、ConcurrentHashMap、CopyOnWriteArrayList、 BlockingQueue

同步/并發容器對比

這里寫圖片描述從圖可以看出HashTable 和 ConcurrentHashMap兩者最大的差異在鎖的粒度，HashTable 的鎖是對整個散列表的全局鎖，而ConcurrentHashMap的鎖粒度更細。 ConcurrentHashMap 鎖的粒度是–Segment（桶）。查看ConcurrentHashMap 的源碼，get() 和 put()方法。

public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE); e != null; e = e.next) { K k; if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value; } } return null; }final V put(K key, int hash, V value, boolean onlyIfAbsent) { HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry<K,V>[] tab = table; int index = (tab.length - 1) & hash; HashEntry<K,V> first = entryAt(tab, index); for (HashEntry<K,V> e = first;;) { if (e != null) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { e.value = value; ++modCount; } break; } e = e.next; } else { if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { unlock(); } return oldValue; }

其中兩個內置類需要特別注意下，這兩個是構建分段加鎖和性能提升的關鍵點。 HashEntry: 可以看到除了value不是final的，其它值都是final的，這意味著不能從hash鏈的中間或尾部添加或刪除節點，因為這需要修改next 引用值，所有的節點的修改只能從頭部開始。對于put操作，可以一律添加到Hash鏈的頭部。但是對于remove操作，可能需要從中間刪除一個節點，這就需要將要刪除節點的前面所有節點整個復制一遍，最后一個節點指向要刪除結點的下一個結點。這在講解刪除操作時還會詳述。為了確保讀操作能夠看到最新的值，將value設置成volatile，這避免了加鎖。 Segment Hash表的一個很重要方面就是如何解決hash沖突，ConcurrentHashMap 和HashMap使用相同的方式，都是將hash值相同的節點放在一個hash鏈中。與HashMap不同的是，ConcurrentHashMap使用多個子Hash表，也就是段(Segment)。

java線程狀態轉換

線程的狀態圖這個圖是不是很熟悉？ Yield是一個靜態的原生(native)方法 Yield告訴當前正在執行的線程把運行機會交給線程池中擁有相同優先級的線程。 Yield不能保證使得當前正在運行的線程迅速轉換到可運行的狀態它僅能使一個線程從運行狀態轉到可運行狀態，而不是等待或阻塞狀態 join()運行著的線程將阻塞直到這個線程實例完成了執行關鍵詞：start、notify、yield、blocked、waiting、sleep、join

同步工具類

信號量（Semaphore）閉鎖（Latch）柵欄（Barrier）工具類：CountDownLatch、FutureTask、Semaphore、CyclicBarrier

閉鎖的作用相當與一扇門：在閉鎖到達結束狀態之前，這扇門一直關閉，沒有任何線程能通過，當達到狀態是，這扇門會打開允許所有線程通過。閉鎖舉例：TestHarness 、Preloader

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; }}

柵欄和閉鎖和類似，閉鎖是等待事件，柵欄是等待其他線程柵欄舉例：CellularAutomata

public class CellularAutomata { private final Board mainBoard; private final CyclicBarrier barrier; private final Worker[] workers; public CellularAutomata(Board board) { this.mainBoard = board; int count = Runtime.getRuntime().availableProcessors(); this.barrier = new CyclicBarrier(count, new Runnable() { public void run() { mainBoard.commitNewValues(); }}); this.workers = new Worker[count]; for (int i = 0; i < count; i++) workers[i] = new Worker(mainBoard.getSubBoard(count, i)); } private class Worker implements Runnable { private final Board board; public Worker(Board board) { this.board = board; } public void run() { while (!board.hasConverged()) { for (int x = 0; x < board.getMaxX(); x++) for (int y = 0; y < board.getMaxY(); y++) board.setNewValue(x, y, computeValue(x, y)); try { barrier.await(); } catch (InterruptedException ex) { return; } catch (BrokenBarrierException ex) { return; } } } private int computeValue(int x, int y) { // Compute the new value that goes in (x,y) return 0; } } public void start() { for (int i = 0; i < workers.length; i++) new Thread(workers[i]).start(); mainBoard.waitForConvergence(); } interface Board { int getMaxX(); int getMaxY(); int getValue(int x, int y); int setNewValue(int x, int y, int value); void commitNewValues(); boolean hasConverged(); void waitForConvergence(); Board getSubBoard(int numPartitions, int index); }}

構建高效可伸縮的緩存

個人覺得在項目中可以用到最多的就是FutureTask，下面舉例來說明試用FutureTask的好處。下面是一個緩存類的設計。

public class Memoizer1 <A, V> implements Computable<A, V> { @GuardedBy("this") private final Map<A, V> cache = new HashMap<A, V>(); private final Computable<A, V> c; public Memoizer1(Computable<A, V> c) { this.c = c; } /** * 帶來了性能問題，每次只有一個線程可以執行該方法 */ public synchronized V compute(A arg) throws InterruptedException { V result = cache.get(arg); if (result == null) { result = c.compute(arg); cache.put(arg, result); } return result; }}

這里寫圖片描述上面Memoizer1的設計是非常糟糕，帶來了性能問題，每次只有一個線程可以執行該方法.如上圖所示，下面對這個對象進行改進。

public class Memoizer2 <A, V> implements Computable<A, V> { private final Map<A, V> cache = new ConcurrentHashMap<A, V>(); private final Computable<A, V> c; public Memoizer2(Computable<A, V> c) { this.c = c; } public V compute(A arg) throws InterruptedException { V result = cache.get(arg); if (result == null) { result = c.compute(arg); cache.put(arg, result); } return result; }}

這里寫圖片描述大量請求同時調用時還是會有比較嚴重的性能問題，c.compute(arg)的計算結果時間較長時，只要結果還未計算出來，每個請求線程都會對結果重新計算一次，如上圖所示。下面引入TaskFutrue來實現。

public class Memoizer3 <A, V> implements Computable<A, V> { private final Map<A, Future<V>> cache = new ConcurrentHashMap<A, Future<V>>(); private final Computable<A, V> c; public Memoizer3(Computable<A, V> c) { this.c = c; } /** * 該方法大大減少了并發多次計算的過程，但是還是會有可以重復計算，在獲取 */ public V compute(final A arg) throws InterruptedException { Future<V> f = cache.get(arg); if (f == null) { Callable<V> eval = new Callable<V>() { public V call() throws InterruptedException { return c.compute(arg); } }; FutureTask<V> ft = new FutureTask<V>(eval); f = ft; cache.put(arg, ft); ft.run(); // call to c.compute happens here 這里調用 c.compute(arg) } try { return f.get(); } catch (ExecutionException e) { throw LaunderThrowable.launderThrowable(e.getCause()); } }}

這里寫圖片描述該方法大大減少了并發多次計算的過程，但是還是會有可以重復計算，在獲取Future f = cache.get(arg)之后判斷if (f == null) 與put(arg, ft) 并不是原子性的操作，在這期間可能會計算多次，如上圖所示。最終的解決方案引入了 ConcurrentMap.putIfAbsent(arg, ft)方法，只在不存在的時候才寫放入Future去計算。最終的優化解決方案如下：

public class Memoizer <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 Memoizer(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 InterruptedException { return c.compute(arg); } }; FutureTask<V> ft = new FutureTask<V>(eval); f = cache.putIfAbsent(arg, ft); if (f == null) { f = ft; ft.run(); } } try { return f.get(); } catch (CancellationException e) { cache.remove(arg, f); } catch (ExecutionException e) { throw LaunderThrowable.launderThrowable(e.getCause()); } } }}