synchronized實現原理及ReentrantLock源碼

時間 2020-12-18

標籤 html java node c# 併發 jvm 函數性能優化欄目 Java 简体版

原文原文鏈接

synchronized

synchronized的做用範圍

public class SynchronizedTest {
    // 實例方法，方法訪問標誌ACC_SYNCHRONIZED,鎖對象是對象實例
    public synchronized void test1(){}
    // 靜態方法，方法訪問標誌ACC_SYNCHRONIZED,鎖對象是MetaSpace中的Class
    // 至關於類的全局鎖，會鎖住全部調用該方法的線程
    public synchronized static void test2(){}

    public void test3() {
        //同步代碼塊，在代碼塊前增長monitorenter指令，代碼塊後增長monitorexit指令
        SynchronizedTest synchronizedTest = new SynchronizedTest();
        synchronized (synchronizedTest) {}
        // 類鎖，效果等同於鎖靜態方法。代碼塊先後增長monitorenter、monitorexit指令
        synchronized (SynchronizedTest.class) {}
    }
}

可jclasslib查看Acc_SYNCHRONIZED標誌和monitorenter、monitorexit指令html

test1 方法：java

Access flags： 0x0021[public synchronized]

test2 方法：node

Access flags： 0x0029[public static synchronized]

test3方法Code操做碼：c#

0 new #2 <com/java/study/jvm/SynchronizedTest>
 3 dup
 4 invokespecial #3 <com/java/study/jvm/SynchronizedTest.<init>>
 7 astore_1
 8 aload_1
 9 dup
10 astore_2
11 monitorenter
12 aload_2
13 monitorexit
14 goto 22 (+8)
17 astore_3
18 aload_2
19 monitorexit
20 aload_3
21 athrow
22 ldc #2 <com/java/study/jvm/SynchronizedTest>
24 dup
25 astore_2
26 monitorenter
27 aload_2
28 monitorexit
29 goto 39 (+10)
32 astore 4
34 aload_2
35 monitorexit
36 aload 4
38 athrow
39 return

synchronized實現

核心組件併發

Wait Set：哪些調用 wait方法被阻塞的線程被放置在這裏
Contention List：競爭隊列，全部請求鎖的線程首先被放在這個競爭隊列中
Entry List： Contention List 中那些有資格成爲候選資源的線程被移動到 Entry List 中
OnDeck：任意時刻，最多隻有一個線程正在競爭鎖資源，該線程被成爲 OnDeck
Owner：當前已經獲取到所資源的線程被稱爲 Owner
!Owner：當前釋放鎖的線程

圖示過程：jvm

解釋：函數

JVM 每次從隊列的尾部取出一個數據用於鎖競爭候選者（OnDeck），可是併發狀況下，ContentionList 會被大量的併發線程進行 CAS 訪問，爲了下降對尾部元素的競爭， JVM 會將一部分線程移動到 EntryList 中做爲候選競爭線程。
Owner 線程會在 unlock 時，將 ContentionList 中的部分線程遷移到 EntryList 中，並指定EntryList 中的某個線程爲 OnDeck 線程（通常是最早進去的那個線程）。
Owner 線程並不直接把鎖傳遞給 OnDeck 線程，而是把鎖競爭的權利交給 OnDeck，OnDeck 須要從新競爭鎖。這樣雖然犧牲了一些公平性，可是能極大的提高系統的吞吐量，在JVM 中，也把這種選擇行爲稱之爲「競爭切換」。
OnDeck 線程獲取到鎖資源後會變爲 Owner 線程，而沒有獲得鎖資源的仍然停留在 EntryList中。若是 Owner 線程被 wait 方法阻塞，則轉移到 WaitSet 隊列中，直到某個時刻經過 notify或者 notifyAll 喚醒，會從新進去 EntryList 中。
處於 ContentionList、 EntryList、 WaitSet 中的線程都處於阻塞狀態，該阻塞是由操做系統來完成的（Linux 內核下采用 pthread_mutex_lock 內核函數實現的）。
Synchronized 是非公平鎖。 Synchronized 在線程進入 ContentionList 時，等待的線程會先嚐試自旋獲取鎖，若是獲取不到就進入 ContentionList，這明顯對於已經進入隊列的線程是不公平的，還有一個不公平的事情就是自旋獲取鎖的線程還可能直接搶佔 OnDeck 線程的鎖資源。
參考： http://www.javashuo.com/article/p-eawupvdg-ba.html
每一個對象都有個 monitor 對象，加鎖就是在競爭 monitor 對象，代碼塊加鎖是在先後分別加上 monitorenter 和 monitorexit 指令來實現的，方法加鎖是經過一個標記位來判斷的
synchronized 是一個重量級操做，須要調用操做系統相關接口，性能是低效的，有可能給線程加鎖消耗的時間比有用操做消耗的時間更多。
Java1.6， synchronized 進行了不少的優化，有適應自旋、鎖消除、鎖粗化、輕量級鎖及偏向鎖等，效率有了本質上的提升。在以後推出的 Java1.7 與 1.8 中，均對該關鍵字的實現機理作了優化。引入了偏向鎖和輕量級鎖。都是在對象頭中有標記位，不須要通過操做系統加鎖。
鎖能夠從偏向鎖升級到輕量級鎖，再升級到重量級鎖。這種升級過程叫作鎖膨脹；
JDK 1.6 中默認是開啓偏向鎖和輕量級鎖，能夠經過-XX:-UseBiasedLocking 來禁用偏向鎖

ReentrantLock

ReentrantLock初始化時，會new一個同步類（默認非公平NonfairSync，當傳入公平參數fair=true時，則new公平類FairSync）；而FairSync 和NonfairSync都繼承ReentrantLock中內部類Sync，Sync則繼承同步器AbstractQueuedSynchronizer。UML圖以下（http://www.javashuo.com/article/p-ahbodinp-ny.html 截取）：性能

Lock流程圖(非公平鎖示例)

源碼

ReentrantLock$NonfairSync#lock(),當state爲0，即compareAndSetState(0, 1)爲true時，得到鎖；不然進行下一步
ReentrantLock$NonfairSync#acquire() ——> AbstractQueuedSynchronizer#acquire() --> ReentrantLock$NonfairSync#tryAcquire() -->
ReentrantLock$Sync#nonfairTryAcquire(), 第2次嘗試獲取鎖
在上面acquire方法中，還會調用addWaiter方法，將一個排他鎖加入隊列

public class ReentrantLock implements Lock, java.io.Serializable {
    
    abstract static class Sync extends AbstractQueuedSynchronizer {
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                //第2次嘗試獲取鎖
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }
        
    static final class NonfairSync extends Sync {
    
        final void lock() {
            // 可不進入隊列，直接搶鎖
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
    
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }
}

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    public final void acquire(int arg) {
        // 步驟3，加入等待隊列，默認排他鎖
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

而繼續addWaiter、enq和acquireQueued則是實現如下圖示過程：
優化

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //前置節點爲null的臨界條件，第一個線程進入等待隊列
        enq(node);
        return node;
    }

前置節點爲null的臨界條件，第一個線程進入等待隊列，進行初始化ui

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                //隊列初始化
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //雙向鏈表添加元素
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

node屬性值介紹：

對應源碼：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    static final class Node {
        
        static final Node EXCLUSIVE = null;
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;

        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter;

        final boolean isShared() {
            return nextWaiter == SHARED;
        }

        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }

        Node() {    // Used to establish initial head or SHARED marker
        }

        Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
        }

        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }
}

重入鎖的實現

重入鎖的可重複進入在如下代碼中實現(非公平鎖示例，公平鎖代碼同樣)：

c > 0, 即有鎖，而且獲取鎖的線程就是當前線程，則將state加1，並更新

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        ...
    }
    // c > 0
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

公平鎖和非公平鎖

第一處不公平地方(lock方法)：

非公平鎖lock時，若是發現沒有鎖了，即state爲0，能夠無論隊列，直接compareAndSetState，若是獲取true了(搶到鎖)，直接得到鎖，不用進同步器中的隊列。
而公平鎖沒有此邏輯。

static final class NonfairSync extends Sync {

    final void lock() {
        // 可不進入隊列，直接搶鎖
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
}

static final class FairSync extends Sync {
    final void lock() {
        acquire(1);
    }
}

第二處不公平的地方(tryAcquire)：

非公平鎖tryAcquire方法會調用Sync#nonfairTryAcquire()，當state爲0，發現鎖被釋放時，可直接搶鎖
公平鎖則必須知足!hasQueuedPredecessors()條件，也即必須同步器中隊列沒有線程在等待，纔去獲取鎖

static final class NonfairSync extends Sync {
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

abstract static class Sync extends AbstractQueuedSynchronizer {
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            //發現鎖被釋放時，可直接搶鎖
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        ...
    }
}

公平鎖

static final class FairSync extends Sync {
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 必須同步器中隊列沒有線程在等待，纔去獲取鎖
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        ...
    }
}

第三處不公平地方，加入隊列時，前置節點是頭節點：

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                ...
                }
            }
    }