【Java進階】併發編程

PS：整理自極客時間《Java併發編程》

1. 概述

• 三種性質
  • 可見性：一個線程對共享變量的修改，另外一個線程能馬上看到。緩存可致使可見性問題。
  • 原子性：一個或多個CPU執行操做不被中斷。線程切換可致使原子性問題。
  • 有序性：編譯器優化可能致使指令順序發生改變。編譯器優化可能致使有序性問題。
• 三個問題
  • 安全性問題：線程安全
  • 活躍性問題：死鎖、活鎖、飢餓
  • 性能問題：
    • 使用無鎖結構：TLS，Copy-On-Write，樂觀鎖；Java的原子類，Disruptor無鎖隊列
    • 減小鎖的持有時間：讓鎖細粒度。如ConcurrentHashmap；再如讀寫鎖，讀無鎖寫有鎖

2. Java內存模型

• volatile
  
  • C語言中的原意：禁用CPU緩存，從內存中讀出和寫入。
  • Java語言的引伸義：
    • Java會將變量馬上寫入內存，其餘線程讀取時直接從內存讀（普通變量改變後，何時寫入內存是不必定的）
    • 禁止指令重排序
  • 解決問題：
    • 保證可見性
    • 保證有序性
    • 不能保證原子性
• Happens-Before規則（H-B）
  • 程序順序性規則：前面執行的語句對後面語句可見
  • volatile變量規則：volatile變量的寫操做對後續的讀操做可見
  • 傳遞性規則：A H-B B，B H-B C，那麼A H-B C
  • 管程中鎖的規則：對一個鎖的解鎖 H-B於 後續對這個鎖的加鎖

3. 互斥鎖sychronized

• 鎖對象：非靜態this，靜態Class，括號Object參數
• 預防死鎖：
  • 互斥：不能破壞
  • 佔有且等待：同時申請全部資源
  • 不可搶佔：sychronized解決不了，Lock能夠解決
  • 循環等待：給資源設置id字段，每次都是按順序申請鎖
• 等待通知機制：
  • wait、notify、notifyAll

class Allocator {
  private List<Object> als;
  // 一次性申請全部資源
  synchronized void apply(
    Object from, Object to){
    // 經典寫法
    while(als.contains(from) ||
         als.contains(to)){
      try{
        wait();
      }catch(Exception e){
      }   
    } 
    als.add(from);
    als.add(to);  
  }
  // 歸還資源
  synchronized void free(
    Object from, Object to){
    als.remove(from);
    als.remove(to);
    notifyAll();
  }
}

 

4. 線程的生命週期

• 通用線程的生命週期：
• Java線程的生命週期：
• 狀態流轉：
  • RUNNABLE -- BLOCKED：線程獲取和等待sychronized隱式鎖
    • ps：調用阻塞式API時，不會進入BLOCKED狀態，但對於操做系統而言，線程實際上進入了休眠態，只不過JVM不關心。
  • RUNNABLE -- WAITING：
    • Object.wait()
    • Thread.join()
    • LockSupport.park()
  • RUNNABLE -- TIMED-WAITING：調用各類帶超時參數的線程方法
  • NEW -- RUNNABLE：Thread.start()
  • RUNNABLE -- TERMINATED：線程運行完畢，有異常拋出，或手動調用線程stop()

6. 線程的性能指標

• 延遲：發出請求到收到響應
• 吞吐量：單位時間內處理的請求數量
• 最佳線程數：
  • CPU密集型：線程數 = CPU核數 + 1
  • IO密集型：線程數 = （IO耗時/CPU耗時 + 1）* CPU核數

7. JDK併發包

• Lock：lock、unlock
  • 互斥鎖，和sychronized同樣的功能，裏面能保證可見性
• Condition：await、signal
  • 條件，相比於sychronized的Object.wait，Condition能夠實現多條件喚醒等待機制
• Semaphore：acquire、release
  • 信號量，能夠用來實現多個線程訪問一個臨界區，如實現對象池設計中的限流器
• ReadWriteLock：readLock、writeLock
  • 寫鎖、讀鎖，容許多線程讀，一個線程寫，寫鎖持有時全部讀鎖和寫鎖的獲取都阻塞（寫鎖的獲取要等全部讀寫鎖釋放）
  • 適用於讀多寫少的場景
• StampedLock：tryOptimisticRead、validate
  • 寫鎖、讀鎖（分悲觀讀鎖、樂觀讀鎖）：
• 線程同步：
  • CountDownLatch：一個線程等待多個線程
    • 初始化 --> countDown（減1） --> await（等待爲0）
  • CyclicBarrier：一組線程之間相互等待
    • 初始化 --> 設置回調函數（爲0時執行，並返回原始值） -->  await（減1並等待爲0）
• 併發容器：
  • List：
    • CopyOnWriteArrayList：適用寫少的場景，要容忍可能的讀不一致
  • Map：
    • ConcurrentHashMap：分段鎖
    • ConcurrentSkipListMap：跳錶
  • Set：
    • CopyOnWriteArraySet：同上
    • ConcurrentSkipListSet：同上
  • Queue：
    • 分類：阻塞Blocking、單端Queue、雙端Deque
    • 單端阻塞（BlockingQueue）：Array～、Linked～、Sychronized～、LinkedTransfer～、Priority～、Delay～
    • 雙端阻塞（BlockingDeque）：Linked～
    • 單端非阻塞（Queue）：ConcurrentLinked～
    • 雙端非阻塞（Deque）：ConcurrentLinked～
• 原子類：
  • 無鎖方案原理：增長了硬件支持，即CPU的CAS指令
  • ABA問題：有解決ABA問題的需求時，增長一個遞增的版本號緯度化解
  • 分類：原子化基本數據類型，原子化引用類型、原子化數組、原子化對象屬性更新器、原子化累加器
• Future：
  • Future：cancel、isCanceled、isDone、get
  • FutureTask：實現了Runnable和Future接口
• 強大工具類
  • CompletableFuture：一個強大的異步編程工具類（任務之間有聚合關係），暫時略
  • CompletionService：批量並行任務，暫時略

8. 線程池

• 設計原理：
  • 用生產者消費者模型，線程池是消費者，調用者是生產者。
  • 線程池對象裏維護一個阻塞隊列，一個已經跑起來的工做線程組ThreadsList
  • ThreadList裏面循環從隊列中去Runnable任務，並調用run方法

  •  1 // 簡化的線程池，僅用來講明工做原理
     2 class MyThreadPool{
     3   // 利用阻塞隊列實現生產者 - 消費者模式
     4   BlockingQueue<Runnable> workQueue;
     5   // 保存內部工做線程
     6   List<WorkerThread> threads 
     7     = new ArrayList<>();
     8   // 構造方法
     9   MyThreadPool(int poolSize, 
    10     BlockingQueue<Runnable> workQueue){
    11     this.workQueue = workQueue;
    12     // 建立工做線程
    13     for(int idx=0; idx<poolSize; idx++){
    14       WorkerThread work = new WorkerThread();
    15       work.start();
    16       threads.add(work);
    17     }
    18   }
    19   // 提交任務
    20   void execute(Runnable command){
    21     workQueue.put(command);
    22   }
    23   // 工做線程負責消費任務，並執行任務
    24   class WorkerThread extends Thread{
    25     public void run() {
    26       // 循環取任務並執行
    27       while(true){ ①
    28         Runnable task = workQueue.take();
    29         task.run();
    30       } 
    31     }
    32   }  
    33 }
    34 
    35 /** 下面是使用示例 **/
    36 // 建立有界阻塞隊列
    37 BlockingQueue<Runnable> workQueue = 
    38   new LinkedBlockingQueue<>(2);
    39 // 建立線程池  
    40 MyThreadPool pool = new MyThreadPool(
    41   10, workQueue);
    42 // 提交任務  
    43 pool.execute(()->{
    44     System.out.println("hello");
    45 });

• ThreadPoolExcutor
  • 參數
    • corePoolSize：線程池保有的最小線程數
    • maximumPoolSize：線程池建立的最大線程數
    • keepAliveTime：工做線程多久沒收到任務，被認爲是閒的
    • workQueue：工做隊列
    • threadFactory：經過這個參數自定義如何建立線程
    • handler：任務拒絕策略
      • 默認爲AbortPolicy，會拋出RejectedExecutionException，這是個運行時異常，要注意
  • 方法
    • void execute()
    • Future submit(Runnable task | Callable task)

9. 鳥瞰並行任務分類