精品免费视频999,久久精品福利视频

線程互斥同步除了使用最基本的 synchronized 關(guān)鍵字外（關(guān)于 synchronized 關(guān)鍵字的實現(xiàn)原理，請看之前寫的線程安全之 synchronized 關(guān)鍵字）， Java 5 之后還提供了 API 可以實現(xiàn)同樣的功能，java.util.concurrent（簡稱 J.U.C）下的重入鎖 ReentrantLock 不僅實現(xiàn)可重入的互斥鎖，還有幾個高級功能：等待可中斷、可實現(xiàn)公平鎖、鎖可綁定多個條件、可限定最大等待時間。下面從基本使用到內(nèi)部實現(xiàn)，層層分析 ReentrantLock 原理。

1. ReentrantLock 的用法

ReentrantLock 文檔中寫明了在 lock() 方法后，用 try 把同步代碼塊包起來，然后在 finally 中調(diào)用 unlock()。這樣做的目的是保證解鎖操作一定會被調(diào)用，防止死鎖。

class X {
    private final ReentrantLock lock = new ReentrantLock();
    // ...

    public void m() {
        // block until condition holds
        lock.lock(); 
        try {
            // ... method body
        } finally {
            lock.unlock();
        }
    }
}

ReentrantLock 還可以綁定多個條件，下面使用 Condition 文檔中的例子來說明：

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    // notFull 是 buffer 沒有到最大值的條件
    final Condition notFull = lock.newCondition();
    // notEmpty 是 buffer 不為空的條件
    final Condition notEmpty = lock.newCondition();

    // buffer 最大值為 100
    final Object[] items = new Object[100];
    int putptr, takeptr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                // buffer 滿了就掛起，直到收到 notFull 的信號
                notFull.await();
            items[putptr] = x;
            if (++ putptr == items.length) putptr = 0;
            ++ count;
            // buffer 新增 item，發(fā)送 notEmpty 信號
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                // buffer 為空就掛起，直到收到 notEmpty 的信號
                notEmpty.await();
            Object x = items[takeptr];
            if (++ takeptr == items.length) takeptr = 0;
            -- count;
            // buffer 取走 item，發(fā)送 notFull 信號
            notFull.signal();
            return x; 
        } finally {
            lock.unlock();
        }
    }
}

2. ReentrantLock 的 API

ReentrantLock 實現(xiàn)了 Lock 和 Serializable 接口，下面是它的一些關(guān)鍵 API。

ReentrantLock() -- 默認使用非公平鎖

ReentrantLock(boolean fair) -- 是否使用公平鎖

void lock() -- 獲取鎖，如果鎖被其他線程持有，則阻塞該線程

void lockInterruptibly() -- 獲取鎖，如果鎖被其他線程持有，則阻塞該線程，直到獲取鎖或被其他線程中斷；如果獲取鎖之前或者在獲取過程的過程中線程中斷，則拋出中斷異常

boolean tryLock() -- 如果直接獲取鎖成功則返回 true；如果鎖被其他線程持有，返回 false

boolean tryLock(long timeout, TimeUnit unit) -- 在等待時間內(nèi)獲取到鎖并且線程沒有被中斷，返回 true；否則返回 false

void unlock() -- 釋放鎖，如果該線程沒有持有鎖，則拋出異常

Condition newCondition() -- 返回一個與鎖關(guān)聯(lián)的 Condition 實例

boolean isHeldByCurrentThread() -- 當前線程是否持有鎖

boolean isLocked() -- 鎖是否被任意線程持有

3. ReentrantLock 的內(nèi)部實現(xiàn)

先總體描述下 ReentrantLock 的大致實現(xiàn)，有一個成員屬性 sync，所有的方法都是調(diào)用該屬性的方法。Sync 繼承 AbstractQueuedSynchronizer（簡稱 AQS），AQS 封裝了鎖和線程等待隊列的基本實現(xiàn)。Sync 有兩個子類 NonfairSync 和 FairSync，分別對應非公平鎖和公平鎖。AQS 內(nèi)部使用volatile int state表示同步狀態(tài)，在 ReentrantLock 中 state 表示占有線程對鎖的持有數(shù)量，為 0 表示鎖未被持有，為 1 表示鎖被某個線程持有，> 1 表示鎖被某個線程持有多次（即重入）。

3.1 默認非公平鎖的 lock()

非公平鎖的 lock() 的方法路線如下：

lock() -> NonfairSync.lock() -> AQS.compareAndSetState(0, 1)
                             -> AQS.acquire(1) -> NonfairSync.tryAcquire(1) ->  Sync.nonfairTryAcquire(1)
                                               -> AQS.acquireQueued(addWaiter(Node.EXCLUSIVE), 1)

下面一步步分析源碼：

final void NonfairSync.lock() {
    // 鎖未被持有，則獲取鎖，并將當前線程設置為鎖的獨占線程
    // 這里可能為其他線程剛剛釋放鎖，還有其他線程在等待，但這時直接獲取，所以是不公平的
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    // 若鎖被持有，則調(diào)用 AQS.acquire(1) 方法
    else
        acquire(1);
}

protected final boolean AQS.compareAndSetState(int expect, int update) {
    // 利用 sun.misc.Unsafe 的 CAS 原子操作
    // 如果 state 的當前值為 expect，則修改為 update，返回 true
    // 如果 state 的當前值不為 expect，返回 false
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

public final void AQS.acquire(int arg) {
    // NonfairSync.tryAcquire(1) 方法只是調(diào)用了 Sync.nonfairTryAcquire(1)
    // 先嘗試獲取鎖
    if (!tryAcquire(arg) &&
        // 獲取失敗則把線程添加到等待隊列中，并阻塞該線程直到獲取成功
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

final boolean Sync.nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 如果鎖未被持有，則直接獲取
        // 這里可能為其他線程剛剛釋放鎖，還有其他線程在等待，但這時直接獲取，所以是不公平的
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        // 鎖被當前線程持有，屬于重入，state ++
        int nextc = c + acquires;
        // 如果 state > 2 ^ 31 - 1, 則拋出異常，這也是最大重入次數(shù)
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

所以非公平鎖的 lock() 的大致邏輯為：如果鎖未被持有，不管等待隊列中的線程直接獲?。蝗绻i被自己（當前線程）持有，則把 state 加 1；否則將當前線程加入到等待隊列中，并阻塞該線程直到獲取成功。

關(guān)于AQS.acquireQueued()的內(nèi)部實現(xiàn)在下一篇文章中專門分析 AQS 的內(nèi)部原理，阻塞線程是調(diào)用LockSupport.park()方法實現(xiàn)的。

LockSupport.park() 與線程中斷的關(guān)系

使用 Object.wait() 阻塞線程后，中斷阻塞線程會喚醒它并且清除中斷狀態(tài)然后拋出 InterruptedException。而 LockSupport.park() 阻塞線程后，線程中斷只會喚醒被阻塞的線程，沒有其他行為，和 unpark() 行為一致，所以需要判斷 Thread.interrupted() 來確定是否由中斷喚醒的。

3.2 公平鎖的 lock()

公平鎖的 lock() 方法路線如下：

lock() -> FairSync.lock() -> AQS.acquire(1) -> FairSync.tryAcquire(1)
                                            -> AQS.acquireQueued(addWaiter(Node.EXCLUSIVE), 1)

公平鎖與非公平鎖的主要區(qū)別在于 FairSync.tryAcquire(1) 這一步：

protected final boolean FairSync.tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 如果鎖未被持有，并且當前線程在等待隊列的頭部或者等待隊列為空，則獲取鎖
        // 保證了沒有線程等待時間超過當前線程，所以是公平的
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        // 鎖被當前線程持有，屬于重入，state ++
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

所以公平鎖的 lock() 的大致邏輯為：如果鎖未被持有，并且當前線程在等待隊列的頭部或者等待隊列為空，則獲取鎖；如果鎖被自己（當前線程）持有，則把 state 加 1；否則將當前線程加入到等待隊列中，并阻塞該線程。

3.3 可中斷的 lockInterruptibly()

lockInterruptibly() 方法的文檔介紹是獲取鎖除非線程中斷，首先看它的方法路線：

lockInterruptibly() -> AQS.acquireInterruptibly(1) -> throw new InterruptedException()
                                                   -> NonfairSync.tryAcquire(1) or FairSync.tryAcquire(1)
                                                   -> AQS.doAcquireInterruptibly(1)

下面看 acquireInterruptibly() 的源碼：

public final void AQS.acquireInterruptibly(int arg)
        throws InterruptedException {
    // 如果當前線程是中斷的，拋出 InterruptedException
    if (Thread.interrupted())
        throw new InterruptedException();
    // 嘗試獲取鎖，即如果鎖未被持有或者已被當前線程持有，直接獲取
    if (!tryAcquire(arg))
        // 獲取鎖失敗，把線程添加到等待隊列，阻塞線程，直到獲取成功或者線程中斷，線程中斷也會拋出 InterruptedException
        doAcquireInterruptibly(arg);
}

從上面實現(xiàn)，可以發(fā)現(xiàn) lockInterruptibly() 與 lock() 的主要區(qū)別有兩點：（1）如果此時線程是中斷的，那么直接拋出 InterruptedException 異常；（2）如果線程被阻塞，在等待過程中線程中斷，拋出 InterruptedException 并取消獲取，從等待隊列中刪除。該方法可以用線程中斷防止長時間阻塞，也可以以此退出死鎖。

3.4 非公平的 tryLock()

不管是公平鎖或者非公平鎖，tryLock() 方法都是使用非公平策略來嘗試獲取鎖，看它的路線圖：

tryLock() -> Sync.nonfairTryAcquire(1)

Sync.nonfairTryAcquire(1) 方法在默認非公平鎖的 lock() 中分析過了，如果鎖未被其他線程持有（兩種情況：1. 未被持有 2. 被自己持有），則獲取鎖并返回 true，否則返回 false。tryLock() 方法只是嘗試獲取鎖，獲取失敗就會返回不會阻塞線程，而使用 synchronized 關(guān)鍵字則會阻塞直到獲取鎖。

3.5 在限定時間內(nèi)的 tryLock(long timeout, TimeUnit unit)

先從方法實現(xiàn)看看與 tryLock() 的區(qū)別：

tryLock(long timeout, TimeUnit unit) -> AQS.tryAcquireNanos(1, unit.toNanos(timeout)) -> throw new InterruptedException()
                                                                                      -> NonfairSync.tryAcquire(1) or FairSync.tryAcquire(1)
                                                                                      -> AQS.doAcquireNanos(1, nanosTimeout)

AQS 的 tryAcquireNanos 方法源碼如下：

public final boolean AQS.tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    // 如果當前線程是中斷的，拋出 InterruptedException
    if (Thread.interrupted())
        throw new InterruptedException();
    // 嘗試獲取鎖，即如果鎖未被持有或者已被當前線程持有，直接獲取
    return tryAcquire(arg) ||
        // 獲取失敗，把線程添加到等待隊列，阻塞線程，直到限定時間、線程中斷或者在此之前獲取成功，線程中斷也會拋出 InterruptedException
        doAcquireNanos(arg, nanosTimeout);
}

可以看出 AQS.tryAcquireNanos(arg, nanosTimeout) 方法與 AQS.acquireInterruptibly(arg) 類似，都支持線程中斷，還加上了一個限定時間。如果限定時間為 0，那么就相當于調(diào)用 tryAcquire(1) 方法。上面的 tryLock() 方法在公平鎖中還是使用非公平策略，但是 tryLock(0, TimeUnit.SECONDS) 在公平鎖中可以實現(xiàn)公平的 tryLock() 方法。

3.6 unlock()

unlock() 釋放持有的鎖，從獲取鎖的過程可以猜測到其中肯定會將 state 減 1，但是具體的方法路線是如何呢？

unlock() -> AQS.release(1) -> Sync.tryRelease(1)
                           -> AQS.unparkSuccessor(head)

下面具體源碼：

public final boolean AQS.release(int arg) {
    // 嘗試釋放鎖
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 釋放成功，并等待隊列的第一個節(jié)點不為空，使用 LockSupport.unpark() 喚醒第一個節(jié)點的線程
            unparkSuccessor(h);
        return true;
    }
    return false;
}

protected final boolean Sync.tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        // 當前線程沒有持有鎖，拋出異常
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        // state 為 0，鎖才是自由的，否則只是退出一次重入，鎖的被持有線程不變
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    // 返回釋放后鎖是否自由，即未被持有
    return free;
}

所以 unlock() 方法實際是將 state 減 1，之后如果鎖是自由的，則會喚起等待隊列的頭節(jié)點中的線程。不過在兩者中間，如果有其他線程獲取鎖的話，公平鎖會判斷是否有線程等待，而非公平鎖則直接獲取該鎖。

3.7 isHeldByCurrentThread() 與 isLocked()

這兩個方法就比較簡單了，直接看對應的源碼：

public boolean isHeldByCurrentThread() {
    return sync.isHeldExclusively();
}

protected final boolean Sync.isHeldExclusively() {
    // 判斷鎖的被持有線程是否為當前線程
    return getExclusiveOwnerThread() == Thread.currentThread();
}

public boolean isLocked() {
    return sync.isLocked();
}

final boolean Sync.isLocked() {
    return getState() != 0;
}

newCondition() 方法會在下面單獨描述，而其他方法不是很重要，這里就不再分析了。

4. Condition

Condition 的作用與 Object 的 wait、notify、notifyAll 類似，用以線程間協(xié)作。調(diào)用 Condition.await() 或 Object.wait() 將阻塞線程等待其他線程的通知，調(diào)用 Conditon.signal()、Condition.signalAll()、Object.nofity()、Object.notifyAll() 將喚起 wait 的線程。

下面有幾個相關(guān)疑問，可以仔細琢磨下。

為什么 Condition 與鎖相關(guān)，Object 的 wait、notify、notifyAll 與對象相關(guān)

先思考為什么 wait、notify、notifyAll 是 Object 的方法，如果它們不和對象相關(guān)聯(lián)，wait() 阻塞線程后，notify() 喚起線程時不知道究竟喚醒哪些 wait 的線程，所以與某一對象對應可以幫助 notify() 時喚醒的也是與該對象相關(guān)的等待線程。

為什么 await、signal 方法需要先獲取鎖，wait、notify 方法需要先獲取對象鎖

這樣做的好處是保證 wait 和 notify 的過程是互斥的，而它們又要與某一個東西相關(guān)聯(lián)，所以直接的方法與對象鎖相關(guān)聯(lián)，實際不是與對象相關(guān)。所以 Condition 和 lock 相關(guān)聯(lián)。

4.1 Condition 的 API

await() -- 釋放相關(guān)的鎖，然后阻塞當前線程直到被 singal 通知或者線程中斷

awaitUninterruptibly() -- 釋放相關(guān)的鎖，阻塞當前線程直到被 singal 通知

awaitNanos(long nanosTimeout)、await(long time, TimeUnit unit)、awaitUntil(Date deadline) -- 釋放相關(guān)的鎖，阻塞當前線程直到被 singal 通知、線程中斷或限定時間到

signal() -- 喚醒一個等待的線程，被喚醒的線程返回 await() 方法前需要重新獲取鎖

singalAll() -- 喚醒所有等待的線程，所有被喚醒的線程返回 await() 方法前需要重新獲取鎖

4.2 ReentrantLock 的 Condition 的內(nèi)部實現(xiàn)

下面看 await()、signal() 兩個方法的實現(xiàn)細節(jié)，ReentrantLock 返回的 Condition 是 AQS.ConditionObject 實例。

public final void AQS.ConditionObject.await() throws InterruptedException {
    // 如果線程中斷，直接拋出 InterruptedException
    if (Thread.interrupted())
        throw new InterruptedException();
    // 先把當前線程添加到 condition 的等待隊列中
    Node node = addConditionWaiter();
    // 釋放線程當前持有的鎖
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 判斷線程是否被通知想重新獲取鎖
    while (!isOnSyncQueue(node)) {
        // 阻塞線程
        LockSupport.park(this);
        // 阻塞線程被喚醒后，如果此時線程中斷，則跳出循環(huán)
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 所以阻塞線程被 signal 喚醒后，或者線程中斷后可以跳出循環(huán)

    // 重新獲取鎖，獲取失敗則阻塞加入阻塞隊列直到獲取成功
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        // 如果獲取的過程中線程中斷，設置 interruptMode 為 REINTERRUPT
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        // 清楚等待隊列中的取消的節(jié)點
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        // 如果 interruptMode 為 REINTERRUPT, 再次中斷線程
        // 如果 interruptMode 為 THROW_IE，拋出 InterruptedException
        reportInterruptAfterWait(interruptMode);
}

所以 awit() 的大致邏輯為：釋放鎖，并且阻塞自己并添加到 condition 的等待隊列，被 signal 通知或線程中斷后喚醒線程，重新獲取鎖。

下面再看 signal() 方法的實現(xiàn)：

public final void AQS.ConditionObject.signal() {
    // 鎖不是互斥獨占鎖時，拋出 IllegalMonitorStateException 異常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        // 如果等待隊列不為空，
        doSignal(first);
}

private void AQS.ConditionObject.doSignal(Node first) {
    do {
        // 把 first 節(jié)點從隊列中移除
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
        // 循環(huán)找到第一個未取消的節(jié)點，把該節(jié)點從 condition 隊列添加到 sync 等待隊列（lock 隊列）
    } while (!transferForSignal(first) &&
                (first = firstWaiter) != null);
}

可以看到 signal() 并沒有喚起 wait 的線程，只是把等待時間最長的未取消線程添加到 sync 等待隊列，等待獲取鎖。

而 signalAll() 方法的區(qū)別時將 condition 等待隊列中所有節(jié)點移到 sync 等待隊列。

現(xiàn)在再來分析下，一開始提供的 Condition 的 BoundedBuffer 示例，假設現(xiàn)在 BoundedBuffer 中 items 為空：

public Object take() throws InterruptedException {
    lock.lock();
    // 此時，線程 A 獲取到 lock
    try {
        while (count == 0)
            // 因為 buffer 為空，釋放獲取的 lock，阻塞線程，添加到 notEmpty 等待隊列
            notEmpty.await();
        Object x = items[takeptr];
        if (++ takeptr == items.length) takeptr = 0;
        -- count;
        // buffer 取走 item，發(fā)送 notFull 信號
        notFull.signal();
        return x; 
    } finally {
        lock.unlock();
    }
}

public void put(Object x) throws InterruptedException {
    lock.lock();
    // 然后，線程 B 獲取到 lock
    try {
        while (count == items.length)
            // buffer 滿了就掛起，直到收到 notFull 的信號
            notFull.await();
        items[putptr] = x;
        if (++ putptr == items.length) putptr = 0;
        ++ count;
        // 把 notEmpty 等待隊列中的線程 A 移到 lock 的等待隊列
        notEmpty.signal();
    } finally {
        // 線程 B 釋放鎖，喚醒 lock 等待隊列中的線程 A，線程 A 獲取到 lock 然后從 await() 方法返回
        lock.unlock();
    }
}

5. 總結(jié)

ReentrantLock 是 API 的重入鎖，相對 synchronized 關(guān)鍵字來說，額外支持公平鎖（synchronized 是非公平的）、獲取鎖可中斷、可以限定獲取的最大時間、可以關(guān)聯(lián)多個 Condition。內(nèi)部主要實現(xiàn)細節(jié)是基于 AQS 的，等待隊列是用鏈表結(jié)構(gòu)存儲的，阻塞隊列使用 LockSupport.park() 實現(xiàn)。

什么時候用 ReentrantLock?

JDK 1.6 之后，synchronized 的性能優(yōu)化得和 ReentrantLock 差不多，所以在 synchronized 可以滿足條件的情況話，優(yōu)先使用 synchronized。

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

線程安全之 ReentrantLock 完全解析

線程安全之 ReentrantLock 完全解析

1. ReentrantLock 的用法

2. ReentrantLock 的 API

3. ReentrantLock 的內(nèi)部實現(xiàn)

3.1 默認非公平鎖的 lock()

3.2 公平鎖的 lock()

3.3 可中斷的 lockInterruptibly()

3.4 非公平的 tryLock()

3.5 在限定時間內(nèi)的 tryLock(long timeout, TimeUnit unit)

3.6 unlock()

3.7 isHeldByCurrentThread() 與 isLocked()

4. Condition

4.1 Condition 的 API

4.2 ReentrantLock 的 Condition 的內(nèi)部實現(xiàn)

5. 總結(jié)

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

線程安全之 ReentrantLock 完全解析

1. ReentrantLock 的用法

2. ReentrantLock 的 API

3. ReentrantLock 的內(nèi)部實現(xiàn)

3.1 默認非公平鎖的 lock()

3.2 公平鎖的 lock()

3.3 可中斷的 lockInterruptibly()

3.4 非公平的 tryLock()

3.5 在限定時間內(nèi)的 tryLock(long timeout, TimeUnit unit)

3.6 unlock()

3.7 isHeldByCurrentThread() 與 isLocked()

4. Condition

4.1 Condition 的 API

4.2 ReentrantLock 的 Condition 的內(nèi)部實現(xiàn)

5. 總結(jié)

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av