深刻理解Java併發框架AQS系列（二）：AQS框架簡介及鎖概念

時間 2021-03-14

標籤 html java 架構併發框架工具性能學習測試優化欄目 Java 简体版

原文原文鏈接

深刻理解Java併發框架AQS系列（一）：線程
 深刻理解Java併發框架AQS系列（二）：AQS框架簡介及鎖概念html

1、AQS框架簡介

AQS誕生於Jdk1.5，在當時低效且功能單一的synchronized的年代，某種意義上講，她拯救了Javajava

注：本系列文章全部測試用例均基於jdk1.8，操做系統爲macOS架構

1.一、思考

咱們去學習一個知識點或開啓一個新課題時，最好是帶着問題去學習，這樣針對性比較強，且印象比較深入，主動思考帶給咱們帶來了無窮的好處併發

拋開AQS，設想如下問題：框架

Q：若是咱們遇到 thread 沒法獲取所需資源時，該如何操做？
A：不斷重試唄，一旦資源釋放可快速嘗試獲取
Q：那若是資源持有時長較長，不斷循環獲取，是否比較浪費CPU ？
A：的確，那就讓線程休息1秒鐘，再嘗試獲取，這樣就不會致使CPU空轉了
Q：那若是資源在第0.1秒時被釋放，那線程豈不是要白白等待0.9秒了？
A：實在不行就讓當前線程掛起，等釋放資源的線程去通知當前線程，這樣就不存在等待時間長短的問題了
Q：但若是資源持有時間很短，每次都掛起、喚醒線程成爲了一個很大的開銷
A：那就依狀況而定，lock時間短的，就不斷循環重試，時間長的就掛起
Q：如何界定lock的時間長短？還有就是若是lock的時間不固定，也沒法預期呢？
A：唔。。。這是個問題
Q：若是線程等待期間，我想放棄呢？
A：。。。。。。
Q：還有不少問題
- 若是我想動態增長資源呢？
- 如何我不想產生飢餓，而保證加鎖的有序性呢？
- 或者我要支持/不支持可重入特性呢？
- 我要查看全部等待資源的線程狀態呢？
- 。。。。。。

咱們發現，一個簡單的等待資源的問題，牽扯出後續諸多龐雜且無頭緒的問題；加鎖不只依賴一套完善的框架體系，還要具體根據使用場景而定，才能接近最優解；那咱們即將要引出的AQS能完美解決上述這些問題嗎？工具

答案是確定的：不能性能

其實Doug Lea也意識到問題的複雜性，不可能出一個超級工具來解決全部問題，因此他把AQS設計爲一個abstract類，並提供一系列子類去解決不一樣場景的問題，例如ReentrantLock、Semaphore等；當咱們發現這些子類也不能知足咱們加鎖需求時，咱們能夠定義本身的子類，經過重寫兩三個方法，寥寥幾行代碼，實現強大的功能，這一切都得益於AQS做者敏銳的前瞻性學習

指的一提的是，雖然咱們能夠用某個子類去實現另外一個子類所提供的功能（例如使用Semaphore替代CountDownLatch），但其易用、簡潔、高效性等可否達到理想效果，都值得商榷；就比如在陸地上穿着雪橇走路，雖能前進，卻低效易摔跤測試

1.二、併發框架

本小節僅帶你們對AQS架構有個初步瞭解，在後文的獨佔鎖、共享鎖等中會詳細闡述。下圖爲AQS框架的主體結構
優化

從上圖中咱們看到了AQS中很是關鍵的一個概念：「阻塞隊列」。即AQS的理念是當線程沒法獲取資源時，提供一個FIFO類型的有序隊列，用來維護全部處於「等待中」的線程。看似無解可擊的框架設計，同時也牽出另外的一個問題：阻塞隊列必定高效嗎?

當「同步塊邏輯」執行很快時，咱們列出兩種場景

場景1：直接使用AQS框架，例如試用其子類ReentrantLock，遇到資源爭搶，放阻塞隊列
場景2：由於鎖佔用時間短，無限重試

針對這2種場景，咱們寫測試用例比較一下

package org.xijiu.share.aqs.compare;

import org.junit.Test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author likangning
 * @since 2021/3/9 上午8:58
 */
public class CompareTest {

  private class MyReentrantLock extends AbstractQueuedSynchronizer {
    protected final boolean tryAcquire(int acquires) {
      final Thread current = Thread.currentThread();
      while (true) {
        int c = getState();
        if (c == 0) {
          if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
          }
        }
      }
    }

    protected final boolean tryRelease(int releases) {
      int c = getState() - releases;
      if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
      boolean free = false;
      if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
      }
      setState(c);
      return free;
    }
  }

  /**
   * 使用AQS框架
   */
  @Test
  public void test1() throws InterruptedException {
    ReentrantLock reentrantLock = new ReentrantLock();
    long begin = System.currentTimeMillis();
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 2; i++) {
      executorService.submit(() -> {
        for (int j = 0; j < 50000000; j++) {
          reentrantLock.lock();
          doBusiness();
          reentrantLock.unlock();
        }
      });
    }
    executorService.shutdown();
    executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    System.out.println("ReentrantLock cost : " + (System.currentTimeMillis() - begin));
  }

  /**
   * 無限重試
   */
  @Test
  public void test2() throws InterruptedException {
    MyReentrantLock myReentrantLock = new MyReentrantLock();
    long begin = System.currentTimeMillis();
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 2; i++) {
      executorService.submit(() -> {
        for (int j = 0; j < 50000000; j++) {
          myReentrantLock.tryAcquire(1);
          doBusiness();
          myReentrantLock.tryRelease(1);
        }
      });
    }
    executorService.shutdown();
    executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    System.out.println("MyReentrantLock cost : " + (System.currentTimeMillis() - begin));
  }

  private void doBusiness() {
    // 空實現，模擬程序快速運行
  }
}

上例，雖然MyReentrantLock繼承了AbstractQueuedSynchronizer，但沒有使用其阻塞隊列。咱們每種狀況跑5次，看下二者在耗時層面的表現

類	耗時1	耗時2	耗時3	耗時4	耗時5	平均耗時（ms）
`ReentrantLock`	11425	12301	12289	10262	11461	11548
`MyReentrantLock`	8717	8957	10283	8445	8928	9066

上例只是拿獨佔鎖舉例，共享鎖也同理。能夠簡單歸納爲：線程掛起、喚醒的時間佔整個加鎖週期比重較大，致使每次掛起、喚醒已經成爲一種負擔。固然此處並非說AQS設計有什麼缺陷，只是想表達並無一種萬能的框架能應對全部狀況，一切都要靠使用者靈活理解、應用

1.三、拓撲結構及如何使用

咱們經常使用的鎖併發類，基本上都是AQS的子類或經過組合方式實現，可見AQS在Java併發體系的重要性

至於如何使用，是須要區分子類是想實現獨佔鎖仍是共享鎖

獨佔鎖
- tryAcquire()
- tryRelease()
- isHeldExclusively() -- 可不實現
共享鎖
- tryAcquireShared()
- tryReleaseShared()

AQS自己是一個abstract類，將主要併發邏輯進行了封裝，咱們定義本身的併發控制類，僅須要實現其中的兩三個方法便可。而在對外(public方法)表現形式上，可依據本身的業務特性來定義；例如Semaphore定義爲acquire、release，而ReentrantLock定義爲lock、unlock

2、鎖

相信你們常常會被各類各樣鎖的定義搞亂，叫法兒也五花八門，爲了後續行文的方便，此章咱們把一些鎖概念闡述一下

2.一、獨佔鎖

獨佔鎖，顧名思義，即在同一時刻，僅容許一個線程執行同步塊代碼。比如一夥兒人想要過河，但只有一根獨木橋，且只能承受一人的重量

JDK支持的典型獨佔鎖：ReentrantLock、ReentrantReadWriteLock

2.二、共享鎖

共享鎖實際上是相對獨佔鎖而言的，涉及到共享鎖就要聊到併發度，即同一時刻最多容許同時執行線程的數量。上圖所述的併發度爲3，即在同一時刻，最多可有3我的在同時過河。

但共享鎖的併發度也能夠設置爲1，此時它能夠看做是獨佔鎖

JDK支持的典型獨佔鎖：Semaphore、CountDownLatch

2.三、公平鎖

雖然叫作公平鎖，但咱們知道任何事情都是相對的，此處也不例外，咱們也只能作到相對公平，後文會涉及，此處再也不贅述

線程在進入時，首先要檢查阻塞隊列中是否爲空，若是發現已有線程在排隊，那麼主動添加至隊尾並等待被逐一喚起；若是發現阻塞隊列爲空，纔會嘗試去獲取資源。公平鎖相對非公平鎖效率較低，一般來說，加鎖時間越短，表現越明顯

2.四、非公平鎖

任何一個剛進入的線程，都會嘗試去獲取資源，釋放資源後，還會通知頭節點去嘗試獲取資源，這樣可能致使飢餓發生，即某一個阻塞隊列中的線程一直得不到調度。

那爲何咱們會說，非公平鎖的效率要高於公平鎖呢？假設一個獨佔鎖，阻塞隊列中已經有10個線程在排隊，線程A搶到資源並執行完畢後，去喚醒頭結點head，head線程喚醒須要時間，head喚醒後才嘗試去獲取資源，而在整個過程當中，沒有線程在執行加鎖代碼

由於線程喚起須要引起用戶態及內核態的切換，故是一個相對比較耗時的操做。

咱們再舉一個不恰當的例子：行政部在操場上爲同窗們辦理業務，由於天氣炎熱，故讓排隊的同窗在場邊一個涼亭等待，涼亭距離業務點約300米，且沒法直接看到業務點，須要等待上一個辦理完畢的同窗來通知。假定平均辦理一個業務耗時約30秒

公平鎖：全部新來辦理業務的同窗都被告知去排隊，上一個辦理完業務的同窗須要去300米外通知下一個同窗，來回600米的路程（線程喚醒）預估耗時2分鐘，在這2分鐘裏，由於沒有同窗過來辦理業務，業務點處於等待狀態

非公平鎖：新來辦理業務的同窗首先看一下業務點是否有人正在在辦理，若是有人正在辦理，那麼主動進入排隊，若是辦理點空閒，那麼直接開始辦理業務。明顯非公平鎖更高效，隊首的同窗接到通知，過來辦理的時間片內，業務點可能已經處理了2個同窗的業務

AQS框架是支持公平、非公平兩種模式的，使用者能夠根據自身的狀況作選擇，而Java中的內置鎖synchronized是非公平鎖

2.五、可重入鎖

即某個線程獲取到鎖後、在釋放鎖以前，再次嘗試獲取鎖，能成功獲取到，不會出現死鎖，即是可重入鎖；須要注意的是，加鎖次數須要跟釋放次數同樣

synchronized、ReentrantLock均爲可重入鎖

2.六、偏向鎖 / 輕量級鎖 / 重量級鎖

之因此將這三個鎖放在一塊兒論述，是由於它們都是synchronized引入的概念，爲了描述流暢，咱們把它們放在一塊兒

偏向鎖：JVM設計者發現，在大多數場景中，在同一時刻爭搶synchronized鎖只有一個線程，並且老是被這一個線程反覆加鎖、解鎖；故引入偏向鎖，且向對象頭的MarkWord部分中，標記上線程id，值得一提的是，在線程加鎖結束後，並無解鎖的動做，這樣帶來的好處首先是少了一次CAS操做，其次當這個線程再次嘗試加鎖時，僅僅比較MarkWord部分中的線程id與當前線程的id是否一致，若是一致則加鎖成功。偏向鎖所以而得名，它偏向於佔有它的線程，對其很是友好。當上一個線程釋放鎖後，若是有另外一個線程嘗試加鎖，偏向鎖會從新偏向新的線程。而當一個線程正佔有鎖，又有一個新的線程試圖加鎖時，便進入了輕量級鎖
輕量級鎖：所謂輕量級鎖，是針對重量級鎖而言的，這個階段也有人叫自旋鎖。其本質就是不會立刻掛起線程，而是反覆重試10（可以使用參數-XX:PreBlockSpin來修改）次。由於線程掛起、喚醒也是至關耗時的，在鎖併發不高、加鎖時間短時，採用自旋能夠獲得更好的效果，具體能夠參考1.2章的測試用例
重量級鎖：線程掛起並進入阻塞隊列，等待被喚醒

這3層鎖是逐級膨脹的，且過程不可回逆，即某個鎖一旦進入重量級鎖，便不可回退至輕量級鎖或偏向鎖。雖然synchronized不是本文的重點，但既然提起來了，咱們能夠把其特性簡單羅列一下

synchronized 獨佔鎖、非公平鎖、可重入；內部作了不少優化

那synchronized鎖的性能究竟如何呢？咱們跟AQS框架中的ReentrantLock作個簡單對比

public class SynchronizedAndReentrant {

  private static int THREAD_NUM = 5;

  private static int EXECUTE_COUNT = 30000000;

  /**
   * 模擬ReentrantLock處理業務
   */
  @Test
  public void test() throws InterruptedException {
    ReentrantLock reentrantLock = new ReentrantLock();
    long begin = System.currentTimeMillis();
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < THREAD_NUM; i++) {
      executorService.submit(() -> {
        for (int j = 0; j < EXECUTE_COUNT; j++) {
          reentrantLock.lock();
          doBusiness();
          reentrantLock.unlock();
        }
      });
    }
    executorService.shutdown();
    executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    System.out.println("ReentrantLock cost : " + (System.currentTimeMillis() - begin));
  }

  private void doBusiness() {
  }

  /**
   * 模擬synchronized處理業務
   */
  @Test
  public void test2() throws InterruptedException {
    long begin = System.currentTimeMillis();
    ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < THREAD_NUM; i++) {
      executorService.submit(() -> {
        for (int j = 0; j < EXECUTE_COUNT; j++) {
          synchronized (SynchronizedAndReentrant.class) {
            doBusiness();
          }
        }
      });
    }
    executorService.shutdown();
    executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    System.out.println("synchronized cost : " + (System.currentTimeMillis() - begin));
  }

}