從 EventBus 看透 synchronized

參考

最近又看了一遍 EventBus 的源碼，感嘆優秀的庫每次學習都能得到更多的知識。安全

先貼一段代碼：markdown

public class HandlerPoster extends Handler implements Poster {

    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;
    private final EventBus eventBus;
    private boolean handlerActive;

    protected HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
        super(looper);
        this.eventBus = eventBus;
        this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
        queue = new PendingPostQueue();
    }

    public void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        //①
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!handlerActive) {
                handlerActive = true;
                if (!sendMessage(obtainMessage())) {
                    throw new EventBusException("Could not send handler message");
                }
            }
        }
    }

    @Override
    public void handleMessage(Message msg) {
        boolean rescheduled = false;
        try {
            long started = SystemClock.uptimeMillis();
            while (true) {
                PendingPost pendingPost = queue.poll();
                if (pendingPost == null) {
                    //②
                    synchronized (this) {
                        // Check again, this time in synchronized
                        pendingPost = queue.poll();
                        if (pendingPost == null) {
                            handlerActive = false;
                            return;
                        }
                    }
                }
                eventBus.invokeSubscriber(pendingPost);
                long timeInMethod = SystemClock.uptimeMillis() - started;
                if (timeInMethod >= maxMillisInsideHandleMessage) {
                    if (!sendMessage(obtainMessage())) {
                        throw new EventBusException("Could not send handler message");
                    }
                    rescheduled = true;
                    return;
                }
            }
        } finally {
            handlerActive = rescheduled;
        }
    }
}
複製代碼

看過 EventBus 源碼的同窗都清楚 HandlerPoster 用於向主線程發送消息，在 handleMessage() 中作消息處理。能夠看到在 ①② 兩處各有一個 synchronized 用於保證數據線程安全。到這裏能夠思考一個問題，若是拿掉 ② 處的 synchronized 有沒有問題，它是否是多餘的？數據結構

① 處的關鍵字保證了數據入棧的操做的安全性。並且咱們已知的是這裏的 handlerMessage() 方法必定是在主線程中調用的，也就是出棧操做只涉及到一個線程，間接保證了數據出棧的安全性。這樣來看的話 ② 處的關鍵字就好像是多餘的了。多線程

若是你真的這麼以爲話那就掉進陷阱裏了，問題在於調用的時機，若是 handlerMessage 方法中第一次判空返回 true 而 ② 處沒有 synchronized 的話這時若是剛好另外一個線程調用了 enqueue() 方法就能夠入棧新的消息，結果就是新入棧的消息得不到及時執行。因此 ② 處 synchronized 關鍵字不只很少餘仍是必要的。併發

看到這裏若是你對 synchronized 的用法和原理還有疑問的話，那就跟我一塊兒深刻學習吧。app

如下內容是對深刻分析Synchronized原理、synchronized 實現原理兩篇文章的摘錄和調整。

synchronized 使用

Synchronized 是 Java 中解決併發問題的一種最經常使用的方法，也是最簡單的一種方法。Synchronized 的做用主要有三個：

原子性：確保線程互斥的訪問同步代碼；
可見性：保證共享變量的修改可以及時可見，實際上是經過 Java 內存模型中的 對一個變量unlock操做以前，必需要同步到主內存中；若是對一個變量進行lock操做，則將會清空工做內存中此變量的值，在執行引擎使用此變量前，須要從新從主內存中load操做或assign操做初始化變量值 來保證的；
有序性：有效解決重排序問題，即 一個unlock操做先行發生(happen-before)於後面對同一個鎖的lock操做。

synchronized 有三種用法：

修飾實例方法：

public synchronized void method1() {
    System.out.println("method 1");
}
複製代碼

修飾靜態方法：

public static synchronized void method2() {
    System.out.println("method 2");
}
複製代碼

修飾代碼塊：

public static void method3() {
    synchronized (this) {
        System.out.println("method 3");       
    }
}
複製代碼

三種用法的區別除了做用範圍外最大的區別在於做用對象的不一樣：

修飾實例方法：當前訪問此方法的實例對象；
修飾靜態方法：當前類的 class 對象，所以靜態方法鎖也至關於該類的一個全局鎖；
修飾代碼塊：() 中傳入的對象。

synchronized 實現原理

當一個線程訪問 synchronized 同步代碼塊時，首先是須要獲得鎖才能執行同步代碼，當退出或者拋出異常時必需要釋放鎖，那麼它是如何來實現這個機制的呢？咱們先看一段簡單的代碼：

package com.paddx.test.concurrent;
public class SynchronizedDemo {
    public void method() {
        synchronized (this) {
            System.out.println("Method 1 start");
        }
    }
}
複製代碼

查看反編譯後結果：

添加 synchronized 關鍵字後，比普通方法多了兩個指令：

monitorenter：線程執行 monitorenter 指令時嘗試獲取 monitor 的全部權，當 monitor 被佔用時對象就會處於鎖定狀態。

若是 monitor 的進入數爲0，則該線程進入 monitor，而後將進入數設置爲1，該線程即爲 monitor 的全部者；

若是線程已經佔有該 monitor，只是從新進入，則進入 monitor 的進入數加1；

若是其餘線程已經佔用了 monitor，則該線程進入阻塞狀態，直到 monitor 的進入數爲0，再從新嘗試獲取 monitor 的全部權；

monitorexit：執行 monitorexit 指令的必須是 monitor 的全部者。指令執行時，monitor的進入數減1，若是減1後進入數爲0，那線程退出 monitor，再也不是這個monitor 的全部者。其餘被這個 monitor 阻塞的線程能夠嘗試去獲取這個 monitor 的全部權。

能夠看到編譯結果中 monitorexit 指令出現了兩次，第一次是程序正常退出執行，而後直接執行 return 指令，第二次是程序異常退出執行，這樣就保證了鎖必定會釋放。

子類同步方法調用了父類同步方法，如沒有可重入的特性，則會發生死鎖；

在看下同步方法：

package com.paddx.test.concurrent;

public class SynchronizedMethod {
    public synchronized void method() {
        System.out.println("Hello World!");
    }
}
複製代碼

查看反編譯後結果：

從編譯的結果來看，方法的同步並無經過指令 monitorenter 和 monitorexit 來完成（理論上其實也能夠經過這兩條指令來實現），不過相對於普通方法，其常量池中多了 ACC_SYNCHRONIZED 標示符。JVM就是根據該標示符來實現方法的同步的：

當方法調用時，調用指令將會檢查方法的 ACC_SYNCHRONIZED 訪問標誌是否被設置，若是設置了，執行線程將先獲取 monitor，獲取成功以後才能執行方法體，方法執行完後再釋放monitor。在方法執行期間，其餘任何線程都沒法再得到同一個 monitor 對象。

兩種同步方式本質上沒有區別，只是方法的同步是一種隱式的方式來實現，無需經過字節碼來完成。

經過對反編譯代碼的解讀咱們知道 synchronized 的實現原理是經過 monitor 的對象來完成，其實 wait/notify 等方法也依賴於 monitor 對象，這就是爲何只有在同步的塊或者方法中才能調用 wait/notify 等方法，不然會拋出異常的緣由。

Monitor

那咱們屢次提到的 monitor 是什麼呢？monitor 被翻譯作管程或監視器。它是 synchronized 實現線程同步的基礎。monitor 有兩個做用：

互斥：即同一時刻只容許一個線程訪問共享資源；
同步：即線程之間如何通訊、協做。

在 Java 虛擬機（HotSpot）中，Monitor 是由 ObjectMonitor 實現的，其主要數據結構以下（位於 HotSpot 虛擬機源碼 ObjectMonitor.hpp 文件，C++ 實現的）：

ObjectMonitor() {
    _header       = NULL;
    _count        = 0; // 記錄個數
    _waiters      = 0,
    _recursions   = 0;
    _object       = NULL;
    _owner        = NULL;
    _WaitSet      = NULL; // 處於wait狀態的線程，會被加入到_WaitSet
    _WaitSetLock  = 0 ;
    _Responsible  = NULL ;
    _succ         = NULL ;
    _cxq          = NULL ;
    FreeNext      = NULL ;
    _EntryList    = NULL ; // 處於等待鎖block狀態的線程，會被加入到該列表
    _SpinFreq     = 0 ;
    _SpinClock    = 0 ;
    OwnerIsThread = 0 ;
  }
複製代碼

ObjectMonitor 中有兩個隊列，_WaitSet 和 _EntryList，用來保存 ObjectWaiter 對象列表（每一個等待鎖的線程都會被封裝成 ObjectWaiter 對象）， _owner指向持有 ObjectMonitor 對象的線程，當多個線程同時訪問一段同步代碼時：

首先會進入 _EntryList 集合，當線程獲取到對象的 monitor 後，進入 _Owner 區域並把 monitor 中的 owner 變量設置爲當前線程，同時 monitor 中的計數器 count 加1；

若線程調用 wait() 方法，將釋放當前持有的 monitor，owner 變量恢復爲null，count 自減1，同時該線程進入 _WaitSet 集合中等待被喚醒；

若當前線程執行完畢，也將釋放 monitor（鎖）並復位 count 的值，以便其餘線程進入獲取 monitor(鎖)；

如上圖所示，一個線程經過1號門進入Entry Set(入口區)，若是在入口區沒有線程等待，那麼這個線程就會獲取監視器成爲監視器的Owner，而後執行監視區域的代碼。若是在入口區中有其它線程在等待，那麼新來的線程也會和這些線程一塊兒等待。線程在持有監視器的過程當中，有兩個選擇，一個是正常執行監視器區域的代碼，釋放監視器，經過5號門退出監視器；還有可能等待某個條件的出現，因而它會經過3號門到Wait Set（等待區）休息，直到相應的條件知足後再經過4號門進入從新獲取監視器再執行。注意：當一個線程釋放監視器時，在入口區和等待區的等待線程都會去競爭監視器，若是入口區的線程贏了，會從2號門進入；若是等待區的線程贏了會從4號門進入。只有經過3號門才能進入等待區，在等待區中的線程只有經過4號門才能退出等待區，也就是說一個線程只有在持有監視器時才能執行wait操做，處於等待的線程只有再次得到監視器才能退出等待狀態。

咱們知道 synchronized 依賴於 monitor 實現，而 monitor 實際上是依賴於 JVM 的 Mutex Lock 來實現的，可是使用 Mutex Lock 被阻塞的線程會被掛起、等待從新調度並從用戶態切換到內核態，對性能有較大影響。並且 HotSpot 的做者發現 大多數鎖只會由同一線程併發申請，基於此在 JDK 6 中對鎖進行了重要改進，優化了其性能引入了偏向鎖、輕量級鎖、適應性自旋等實現。

因此目前鎖主要存在四種狀態：無鎖狀態、偏向鎖狀態、輕量級鎖狀態、重量級鎖狀態。鎖能夠從偏向鎖升級到輕量級鎖，再升級的重量級鎖。可是鎖的升級是單向的，也就是說只能從低到高升級，不會出現鎖的降級。其實 monitor 機制只是鎖升級到重量鎖後的工做機制。那鎖是如何升級的呢？偏向鎖和輕量級鎖是如何實現的呢？咱們接着看。

在 JDK 1.6 中默認是開啓偏向鎖和輕量級鎖的，能夠經過-XX:-UseBiasedLocking來禁用偏向鎖。

Java 對象頭

以前咱們介紹過，當線程進入 synchronized 同步代碼塊時會去獲取 monitor 鎖，獲取成功就能夠執行此同步代碼塊。那是如何獲取 monitor 的呢？其實每個 Java 對象都默認攜帶 monitor 對象，線程獲取 monitor 鎖其實就是獲取對象內部的 monitor。

在 JVM 中，對象在內存中的佈局分爲三塊區域，以下圖：

實例數據：存放類的屬性數據信息，包括父類的屬性信息；
對齊填充：因爲虛擬機要求對象起始地址必須是8字節的整數倍。填充數據不是必須存在的，僅僅是爲了字節對齊；
對象頭：Java對象頭通常佔有2個機器碼（在32位虛擬機中，1個機器碼等於4字節，也就是32bit，在64位虛擬機中，1個機器碼是8個字節，也就是64bit），可是若是對象是數組類型，則須要3個機器碼，由於JVM虛擬機能夠經過Java對象的元數據信息肯定Java對象的大小，可是沒法從數組的元數據來確認數組的大小，因此用一塊來記錄數組長度。

Hotspot 虛擬機的對象頭主要包括兩部分數據：

Mark Word：存儲對象自身的運行時數據，它是實現輕量級鎖和偏向鎖的關鍵；
Klass Pointer：類型指針，指向它的類元數據的指針，虛擬機經過這個指針來肯定這個對象是哪一個類的實例；
Array Length：數組長度。非必須，若是數據對象是數組用來保存數組長度。

64 位虛擬機 Mark Word 是 64 bit，在不一樣狀態下其存儲數據結構以下：

對象頭的最後兩位存儲了鎖的標誌位，01是初始狀態，未加鎖，其對象頭裏存儲的是對象自己的哈希碼，隨着鎖級別的不一樣，對象頭裏會存儲不一樣的內容。偏向鎖存儲的是當前佔用此對象的線程ID；而輕量級則存儲指向線程棧中鎖記錄的指針。從這裏咱們能夠看到，「鎖」這個東西，多是個鎖記錄+對象頭裏的引用指針（判斷線程是否擁有鎖時將線程的鎖記錄地址和對象頭裏的指針地址比較)，也多是對象頭裏的線程ID（判斷線程是否擁有鎖時將線程的ID和對象頭裏存儲的線程ID比較）。

下面來看 synchronized 鎖狀態升級流程：

偏向鎖

流程

當線程訪問同步塊並獲取鎖時處理流程以下：

檢查 Mark Word 的 線程id。
若是爲空則設置 CAS 替換當前 線程id。若是替換成功則獲取鎖成功，若是失敗則撤銷偏向鎖。
若是不爲空則檢查 線程id 爲是否爲本線程。若是是則獲取鎖成功，若是失敗則撤銷偏向鎖。

持有偏向鎖的線程之後每次進入這個鎖相關的同步塊時，只需比對一下 Mark Word 的 線程id 是否爲本線程，若是是則獲取鎖成功。

若是發生線程競爭發生二、3 步失敗的狀況則須要撤銷偏向鎖。

偏向鎖的撤銷

偏向鎖的撤銷動做必須等待全局安全點
暫停擁有偏向鎖的線程，判斷鎖對象是否處於被鎖定狀態
撤銷偏向鎖恢復到無鎖（標誌位爲 01）或輕量級鎖（標誌位爲 00）的狀態

優缺點

優勢：

只有一個線程執行同步塊時進一步提升性能，適用於一個線程反覆得到同一鎖的狀況。偏向鎖能夠提升帶有同步但無競爭的程序性能。

缺點：

若是存在競爭會帶來額外的鎖撤銷操做。

輕量級鎖

加鎖

多個線程競爭偏向鎖致使偏向鎖升級爲輕量級鎖

JVM 在當前線程的棧幀中建立 Lock Reocrd，並將對象頭中的 Mark Word 複製到 Lock Reocrd 中。（Displaced Mark Word）
線程嘗試使用 CAS 將對象頭中的 Mark Word 替換爲指向 Lock Reocrd 的指針。若是成功則得到鎖，若是失敗則先檢查對象的 Mark Word 是否指向當前線程的棧幀若是是則說明已經獲取鎖，不然說明其它線程競爭鎖則膨脹爲重量級鎖。

解鎖

使用 CAS 操做將 Mark Word 還原
若是第 1 步執行成功則釋放完成
若是第 1 步執行失敗則膨脹爲重量級鎖。

優缺點

優勢

其性能提高的依據是對於絕大部分的鎖在整個生命週期內都是不會存在競爭。在多線程交替執行同步塊的狀況下，能夠避免重量級鎖引發的性能消耗。

缺點

在有多線程競爭的狀況下輕量級鎖增長了額外開銷。

自旋鎖

自旋是一種獲取鎖的機制並非一個鎖狀態。在膨脹爲重量級鎖的過程當中或重入時會屢次嘗試自旋獲取鎖以免線程喚醒的開銷，可是它會佔用 CPU 的時間所以若是同步代碼塊執行時間很短自旋等待的效果就很好，反之則浪費了 CPU 資源。默認狀況下自旋次數是 10 次用戶可使用參數 -XX : PreBlockSpin 來更改。那麼如何優化來避免此狀況發生呢？咱們來看適應性自旋。

適應性自旋鎖

JDK 6 引入了自適應自旋鎖，意味着自旋的次數不在固定，而是由前一次在同一個鎖上的自旋時間及鎖的擁有者的狀態來決定。若是對於某個鎖不多自旋成功那麼之後有可能省略掉自旋過程以免資源浪費。有了自適應自旋隨着程序運行和性能監控信息的不斷完善，虛擬機對程序鎖的情況預測就會愈來愈準確，虛擬機就會變得愈來愈聰明了。

優缺點

優勢

競爭的線程不會阻塞掛起，提升了程序響應速度。避免重量級鎖引發的性能消耗。

缺點

若是線程始終沒法獲取鎖，自旋消耗 CPU 最終會膨脹爲重量級鎖。

重量級鎖

在重量級鎖中沒有競爭到鎖的對象會 park 被掛起，退出同步塊時 unpark 喚醒後續線程。喚醒操做涉及到操做系統調度會有額外的開銷。也就是上面介紹的 monitor 機制了。

清楚了 synchronized 同步代碼塊是如何工做的以及和對象之間的關係，再來看最開始的問題就很清晰了。在 BackgroundPoster 中使用兩個 synchronized 代碼塊，()中傳入的是 this 實例對象，也就保證了當經過同一實例對象訪問數據入隊出隊的安全性。

多線程問題是任何操做系統中都至關複雜的一部分，這裏只摘錄了一小部分，可是基本也能夠保證咱們在平常開發中能作到心中有數了 ^_^。