ReentrantLock 類

mykurisu 發布于2019-08-16 16:16 / 1163人閱讀

摘要：在多線程中可以使用關鍵字來實現多線程之間同步互斥但在中新增加了類也能達到同樣的效果并且在擴展功能上也更加強大比如具有嗅探鎖定多路分支通知公平鎖和非公平鎖等默認功能而且在使用上也比更加的靈活使用實現同步調用對象的方法獲取鎖調用方法釋放鎖從運行

在 Java 多線程中, 可以使用 synchronized 關鍵字來實現多線程之間同步互斥, 但在 JDK 1.5 中新增加了 ReentrantLock 類也能達到同樣的效果, 并且在擴展功能上也更加強大, 比如具有嗅探鎖定, 多路分支通知, 公平鎖和非公平鎖等(默認)功能, 而且在使用上也比 synchronized 更加的靈活.

使用 ReentrantLock 實現同步

public class MyService {

    private Lock lock = new ReentrantLock();

    public void testMethod() {
        lock.lock();

        for (int i = 0; i < 10; i++){
            System.out.println("ThreadName=" + Thread.currentThread().getName() + (" " + (i + 1)));
        }

        lock.unlock();

    }

}

public class MyThread extends Thread {

    private MyService myService;
    public MyThread(MyService myService) {
        this.myService = myService;
    }

    @Override
    public void run() {
        myService.testMethod();
    }
}

    public static void main(String[] args) throws IOException, InterruptedException {

        MyService myService = new MyService();

        MyThread myThreadA = new MyThread(myService);
        MyThread myThreadB = new MyThread(myService);
        MyThread myThreadC = new MyThread(myService);
        MyThread myThreadD = new MyThread(myService);
        MyThread myThreadE = new MyThread(myService);

        myThreadA.start();
        myThreadB.start();
        myThreadC.start();
        myThreadD.start();
        myThreadE.start();

    }

調用 ReentrantLock 對象的 lock() 方法獲取鎖, 調用 unLock() 方法釋放鎖.

從運行結果來看, 當前線程打印完畢之后將鎖進行釋放, 其他的線程才可以繼續打印. 線程打印的數據是分組打印, 因為當前線程已經持有鎖, 但線程之間打印的順序是隨機的.

使用 Condition 實現等待/通知

關鍵字 synchronized 與 wait() 和 notify() / notifyall() 方法結合可以實現等待/通知模式, 只不過在使用時, 調用 notify() 方法 JVM 會隨機選擇一個 WAITNG 狀態的線程來執行.

而使用 Condition 則可以更加靈活, 可以實現 "選擇性通知", 可以指定的選擇喚醒哪些線程, 哪些線程繼續等待.

public class MyService {

    private Lock lock = new ReentrantLock();
    public Condition conditionA = lock.newCondition();
    public Condition conditionB = lock.newCondition();

    public void awaitA() throws InterruptedException {
        lock.lock();

        System.out.println("begin awaitA 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        conditionA.await();

        System.out.println("end awaitA 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        lock.unlock();
    }

    public void awaitB() throws InterruptedException {
        lock.lock();

        System.out.println("begin awaitB 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        conditionB.await();

        System.out.println("end awaitB 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        lock.unlock();
    }

    public void  signalAll_A() throws InterruptedException {
        lock.lock();
        System.out.println("begin signalAll_A 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        conditionA.signalAll();

        lock.unlock();
    }

    public void  signalAll_B() throws InterruptedException {
        lock.lock();
        System.out.println("begin signalAll_B 時間" + System.currentTimeMillis() + "ThreadName=" + Thread.currentThread().getName());

        conditionB.signalAll();

        lock.unlock();
    }
}

public class ThreadA extends Thread {

    private MyService myService;
    public ThreadA(MyService myService) {
        this.myService = myService;
    }

    @Override
    public void run() {
        try {
            myService.awaitA();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class ThreadB extends Thread {

    private MyService myService;

    public ThreadB(MyService myService) {
        this.myService = myService;
    }

    @Override
    public void run() {
        try {
            myService.awaitB();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

    public static void main(String[] args) throws IOException, InterruptedException {

        MyService myService = new MyService();

        ThreadA threadA = new ThreadA(myService);
        threadA.setName("a");
        threadA.start();

        ThreadB threadB = new ThreadB(myService);
        threadB.setName("b");
        threadB.start();

        Thread.sleep(3000);
        myService.signalAll_A();

    }

Object 類中的 wait() 方法相當于 Condition 類中的 await() 方法.

Object 類中的 wait(long timeout) 方法相當于 Condition 類中的 await(long time, TimeUnit unit) 方法.

Object 類中的 notify() 方法相當于 Condition 類中的 signal() 方法.

Object 類中的 notifyAll() 方法相當于 Condition 類中的 signalAll() 方法.

從執行結果來看, a 和 b 線程被暫停, 當執行 myService.signalAll_A() 方法時, a 線程繼續執行, 而 b 線程仍然是等待狀態.

源碼

ReentrantLock 類實現了 Lock, java.io.Serializable

    public ReentrantLock() {
        sync = new NonfairSync();
    }
    
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

兩個構造方法, 默認是非公平鎖, 如果為 true 表明是公平鎖.

可以看到 NonfairSync 和 FairSync 都是繼承了 Sync 這個抽象類, 而 Sync 則繼承了AQS. Sync、NonfairSync、FairSync 都是 ReentrantLock 的靜態內部類, ReentrantLock 的許多方法都是Sync類代為實現.

AbstractQueuedSynchronizer 核心方法

AQS最核心的數據結構是一個 volatile int state 和一個 FIFO 線程等待對列.

state 代表共享資源的數量, 如果是互斥訪問, 一般設置為1, 而如果是共享訪問, 可以設置為N(N為可共享線程的個數);

而線程等待隊列是一個雙向鏈表, 無法立即獲得鎖而進入阻塞狀態的線程會加入隊列的尾部. 當然對 state 以及隊列的操作都是采用了 volatile + CAS + 自旋的操作方式, 采用的是樂觀鎖的概念.

acquire 方法

此方法是獨占模式下線程獲取共享資源的頂層入口.

public void lock() {
    sync.acquire(1);
}

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&    //嘗試獲取鎖，若獲取成功，則state減1，返回true
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //若獲取鎖不成功，調用addWaiter方法使線程進入等待隊列，acquireQueued方法讓線程進入阻塞狀態
            selfInterrupt(); //檢查在等待過程中是否有中斷，若有中斷，則在此時再響應
}

tryAcquire方法

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
}

為什么要拋出異常而不是聲明為抽象類呢？
因為AQS是可選模式的, 我們選擇的是獨占模式, 就不需要去重寫 tryAcquireShared 方法, 如果我們選的是共享模式, 也不需要重寫 tryAcquire 方法, 因此AQS雖然是抽象類, 但是沒有抽象方法, 而是用拋出異常的方式代替.

addWaiter 方法

addWaiter方法的主要是把當前線程加入到FIFO等待隊列隊尾.

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);//創建節點
        // 首先嘗試快速插入隊尾
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {//CAS操作
                pred.next = node;
                return node;
            }
        }
        enq(node);//若不成功，則嘗試以自旋方式插入隊尾
        return node;
}

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;
                }
            }
        }
}

acquireQueued 方法

acquireQueued 方法, 主要是讓加入隊尾的線程進入等待狀態, 等到前面的進程執行完了, 再喚醒該線程, 去執行同步代碼在這里是檢測是否應該park()(park是一個Unsafe包中的native方法), 以及檢測在隊列的等待過程中是否有中斷, 在等待過程中是不響應中斷的, 等到等待結束被喚醒時, 才去向上傳遞是否中斷過的值.

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);
        }
}

常用方法 ReentrantLock 類

int getHoldCount() 查詢調用 lock() 方法的次數.

final int getQueueLength() 估計等待鎖的線程數. 比如有5個線程, 1個線程首先執行 await() 方法, 那么在調用此方法后返回值是4, 說明有4個線程同時在等待lock的釋放.

int getWaitQueueLength(Condition condition) 返回與此鎖相關聯給定條件等待的線程數的估計. 比如有5個線程, 每個線程都執行了同一個 condition 對象的 await() 方法, 則調用此方法時返回的值是5.

final boolean hasQueuedThreads() 判斷是否有線程等待此鎖.

final boolean hasQueuedThread(Thread thread) 判斷指定線程是否等待獲取此鎖.

boolean hasWaiters(Condition condition) 判斷線程有沒有調用 await() 方法.

void lockInterruptibly() throws InterruptedException 獲取鎖, 除非當前線程為interrupted.

Condition 類

void awaitUninterruptibly() 和 await() 區別就是當調用 interrupt() 方法時不會拋出 InterrputedException 異常.

GPU云服務器云服務器 ReentrantLock 分頁類異常類 java類內部類 1類2類3類模塊化機房

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/74007.html

Java多線程進階（三）—— J.U.C之locks框架：ReentrantLock

摘要：公平策略在多個線程爭用鎖的情況下，公平策略傾向于將訪問權授予等待時間最長的線程。使用方式的典型調用方式如下二類原理的源碼非常簡單，它通過內部類實現了框架，接口的實現僅僅是對的的簡單封裝，參見原理多線程進階七鎖框架獨占功能剖析 showImg(https://segmentfault.com/img/remote/1460000016012582); 本文首發于一世流云的專欄：https...

jasperyang 2019-08-16 12:41 評論0 收藏0
Java 重入鎖 ReentrantLock 原理分析

摘要：的主要功能和關鍵字一致，均是用于多線程的同步。而僅支持通過查詢當前線程是否持有鎖。由于和使用的是同一把可重入鎖，所以線程可以進入方法，并再次獲得鎖，而不會被阻塞住。公平與非公平公平與非公平指的是線程獲取鎖的方式。 1.簡介可重入鎖ReentrantLock自 JDK 1.5 被引入，功能上與synchronized關鍵字類似。所謂的可重入是指，線程可對同一把鎖進行重復加鎖，而不會被阻...

lx1036 2019-08-15 15:54 評論0 收藏0
J.U.C|可重入鎖ReentrantLock

摘要：二什么是重入鎖可重入鎖，顧名思義，支持重新進入的鎖，其表示該鎖能支持一個線程對資源的重復加鎖。將由最近成功獲得鎖，并且還沒有釋放該鎖的線程所擁有?？梢允褂煤头椒▉頇z查此情況是否發生。一、寫在前面前幾篇我們具體的聊了AQS原理以及底層源碼的實現，具體參見《J.U.C|一文搞懂AQS》《J.U.C|同步隊列（CLH）》《J.U.C|AQS獨占式源碼分析》《J.U.C|AQS共享式源...

wangdai 2019-08-16 17:00 評論0 收藏0
Lock鎖子類了解一下

摘要：前言回顧前面多線程三分鐘就可以入個門了源碼剖析多線程基礎必要知識點看了學習多線程事半功倍鎖機制了解一下簡簡單單過一遍只有光頭才能變強上一篇已經將鎖的基礎簡單地過了一遍了，因此本篇主要是講解鎖主要的兩個子類那么接下來我們就開始吧一鎖首先我們來前言回顧前面：多線程三分鐘就可以入個門了！ Thread源碼剖析多線程基礎必要知識點！看了學習多線程事半功倍 Java鎖機制了解一下 AQ...

時飛 2019-08-15 15:43 評論0 收藏0
Java同步機制的底層實現

摘要：在多線程編程中我們會遇到很多需要使用線程同步機制去解決的并發問題，而這些同步機制就是多線程編程中影響正確性和運行效率的重中之重。這五個方法之所以能指定同步器的行為，則是因為中的其他方法就是通過對這五個方法的調用來實現的。在多線程編程中我們會遇到很多需要使用線程同步機制去解決的并發問題，而這些同步機制就是多線程編程中影響正確性和運行效率的重中之重。這不禁讓我感到好奇，這些同步機制是如何...

yintaolaowanzi 2019-08-16 17:40 評論0 收藏0