學(xué)習(xí)JUC源碼（2）——自定義同步組件

丹楓無跡 2021-06-22

展開全文

前言

　　在之前的博文（學(xué)習(xí)JUC源碼（1）——AQS同步隊(duì)列（源碼分析結(jié)合圖文理解））中，已經(jīng)介紹了AQS同步隊(duì)列的相關(guān)原理與概念，這里為了再加深理解ReentranLock等源碼，模仿構(gòu)造同步組件的基本模式，編寫不可重入的互斥鎖Mutex與指定共享線程數(shù)量的共享鎖。MySharedLock。
　　主要參考資料《Java并發(fā)編程藝術(shù)》（有需要的小伙伴可以找我，我這里只有電子PDF）同時結(jié)合ReentranLock、AQS等源碼。

一、構(gòu)造同步組件的模式

叢概念方層面，在中，我們知道鎖與同步器的相關(guān)概念：

同步器是實(shí)現(xiàn)鎖的關(guān)鍵，在鎖的實(shí)現(xiàn)中聚合同步器，利用同步器實(shí)現(xiàn)鎖的語義；
鎖是面向使用者的，提供鎖交互的實(shí)現(xiàn)；
同步器是面向鎖的實(shí)現(xiàn)者，簡化了鎖的實(shí)現(xiàn)方式，屏蔽了同步狀態(tài)管理、線程排隊(duì)、等待/喚醒等底層操作。

從代碼層面，同步器是基于模板模式實(shí)現(xiàn)的，可以通過可重寫的方法中的隨便一個窺探：

　　/** * 模板方法：
     *  protected關(guān)鍵字
     *  沒有任何實(shí)現(xiàn)
     * @param arg
     * @return */protected boolean tryAcquire(int arg) {throw new UnsupportedOperationException();
    }

也就是需要進(jìn)行以下幾步：

1）繼承同步器重寫指定方法（idea中extends AQS點(diǎn)擊快捷鍵ctrl+O即可顯示）

tryAcquire(int arg)：獨(dú)占式獲取同步狀態(tài)；
tryRelease(int arg)：獨(dú)占式釋放同步狀態(tài)；
tryAcquireShared(int arg)：共享式獲取同步狀態(tài)，返回大于0的值表示獲取成功，否則失敗
tryReleaseShared(int arg)：共享式釋放鎖
isHeldExclusively()：當(dāng)前線程是否在獨(dú)占模式下被線程占用，一般該方法表示是否被當(dāng)前線程占用

2）隨后將同步器組合在自定義同步組件的實(shí)現(xiàn)中，即定義內(nèi)部類Syn繼承AQS，在Syn中重寫AQS方法：

public class Sync extends AbstractQueuedSynchronizer{
        @Overrideprotected boolean tryAcquire(int arg) {final Thread current = Thread.currentThread();if (compareAndSetState(0, 1)) {// 獲取成功之后，當(dāng)前線程是該鎖的持有者，不需要再可重入數(shù)                setExclusiveOwnerThread(current);return true;
            }return false;
        }

        @Overrideprotected boolean tryRelease(int arg) {if (getState() == 0) {throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);return true;
        }

        @Overrideprotected boolean isHeldExclusively() {              return getState() == 1;
        }// 返回Condition，每個Condition都包含了一個隊(duì)列        Condition newCondition() {return new ConditionObject();
        }
    }

3）最后調(diào)用同步器提供的模板方法，即同步組件類實(shí)現(xiàn)Lock方法之后，在lock/unlock方法中調(diào)用內(nèi)部類Syn的方法acquire(int arg)等方法

public class Mutex implements Lock {
    
   ........
    @Overridepublic void lock() {
        sync.acquire(1);
    }
    @Overridepublic void unlock() {
        sync.release(1);
    }
    ........

}

具體請看下面的實(shí)踐部分

二、互斥不可重入鎖

在我之前寫過的博文中（詳解Java鎖的升級與對比（1）——鎖的分類與細(xì)節(jié)（結(jié)合部分源碼））介紹可重入鎖與不可重入鎖的區(qū)別時，就寫到JUC中沒有不可重入鎖的具體實(shí)現(xiàn)，但是可以類比，現(xiàn)在呢，我們可以做到實(shí)現(xiàn)了，具體看下面代碼，模式完全符合依賴Lock與AQS構(gòu)造同步組件模式。

（1）Mutex代碼實(shí)現(xiàn)（核心關(guān)鍵實(shí)現(xiàn)已經(jīng)在代碼中注釋）

public class Mutex implements Lock {private final Sync sync = new Sync();public class Sync extends AbstractQueuedSynchronizer{
        @Overrideprotected boolean tryAcquire(int arg) {final Thread current = Thread.currentThread();if (compareAndSetState(0, 1)) {// 獲取成功之后，當(dāng)前線程是該鎖的持有者，不需要再可重入數(shù)                setExclusiveOwnerThread(current);return true;
            }return false;
        }

        @Overrideprotected boolean tryRelease(int arg) {if (getState() == 0) {throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            setState(0);return true;
        }

        @Overrideprotected boolean isHeldExclusively() {              return getState() == 1;
        }// 返回Condition，每個Condition都包含了一個隊(duì)列        Condition newCondition() {return new ConditionObject();
        }
    }


    @Overridepublic void lock() {
        sync.acquire(1);
    }
    @Overridepublic void unlock() {
        sync.release(1);
    }

    @Overridepublic void lockInterruptibly() throws InterruptedException {

    }

    @Overridepublic boolean tryLock() {return false;
    }

    @Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return false;
    }



    @Overridepublic Condition newCondition() {return null;
    }
}

其中核心代碼就是重寫的兩個方法：

tryAcquire(int arg)方法：主要是設(shè)置同獨(dú)占式更新同步狀態(tài)，CAS實(shí)現(xiàn)state+1
tryRelease(int arg)方法：獨(dú)占式釋放同步狀態(tài)，釋放鎖持有

（2）測試Demo

public class MutexDemo {

    @Testpublic void test(){final Mutex lock = new Mutex();class Worker extends Thread {
            @Overridepublic void run() {// 一直不停在獲取鎖while (true) {
                    lock.lock();try {
                        System.out.println(Thread.currentThread().getName() +" hold lock, "+new Date());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        lock.unlock();
                        System.out.println(Thread.currentThread().getName() +" release lock, "+new Date());
                    }
                }
            }

        }for (int i = 0; i < 10; i++) {
            Worker worker = new Worker();// 以守護(hù)進(jìn)程運(yùn)行，VM退出不影響運(yùn)行，這里只是為了一個打印效果，去掉注釋一直打印worker.setDaemon(true);
            worker.start();
        }// 每隔一秒換行for (int j = 0; j < 10; j++) {try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println();
        }
    }
}

（3）運(yùn)行結(jié)果

Thread-0 hold lock, Tue Dec 08 16:26:42 CST 2020Thread-0 release lock, Tue Dec 08 16:26:43 CST 2020Thread-1 hold lock, Tue Dec 08 16:26:43 CST 2020Thread-2 hold lock, Tue Dec 08 16:26:44 CST 2020Thread-1 release lock, Tue Dec 08 16:26:44 CST 2020Thread-3 hold lock, Tue Dec 08 16:26:45 CST 2020Thread-2 release lock, Tue Dec 08 16:26:45 CST 2020Thread-3 release lock, Tue Dec 08 16:26:46 CST 2020Thread-4 hold lock, Tue Dec 08 16:26:46 CST 2020Thread-4 release lock, Tue Dec 08 16:26:47 CST 2020Thread-6 hold lock, Tue Dec 08 16:26:47 CST 2020Thread-7 hold lock, Tue Dec 08 16:26:48 CST 2020Thread-6 release lock, Tue Dec 08 16:26:48 CST 2020Thread-7 release lock, Tue Dec 08 16:26:49 CST 2020Thread-5 hold lock, Tue Dec 08 16:26:49 CST 2020Thread-8 hold lock, Tue Dec 08 16:26:50 CST 2020Thread-5 release lock, Tue Dec 08 16:26:50 CST 2020Thread-8 release lock, Tue Dec 08 16:26:51 CST 2020Thread-9 hold lock, Tue Dec 08 16:26:51 CST 2020

（4）結(jié)果分析

互斥鎖的核心就是同一個同步狀態(tài)只能被一個線程持有，其它線程等待持有線程釋放才能競爭獲取。截圖一開始的運(yùn)行結(jié)果分析：

Thread-0 hold lock, Tue Dec 08 16:26:42 CST 2020Thread-0 release lock, Tue Dec 08 16:26:43 CST 2020Thread-1 hold lock, Tue Dec 08 16:26:43 CST 2020Thread-2 hold lock, Tue Dec 08 16:26:44 CST 2020Thread-1 release lock, Tue Dec 08 16:26:44 CST 2020

10個線程不斷競爭鎖，一開始Thread-0在08 16:26:42獲取到鎖，持有鎖1秒后在釋放16:26:43時釋放，同時Thread-1立馬獲取到鎖，1秒后于16:26:44釋放鎖，同時Thread-2立馬獲取到了鎖......

根據(jù)輸出結(jié)果來說，完全符合Mutex作為互斥鎖這個功能：同一時刻只有一個線程持有鎖（同步狀態(tài)），其它線程等待釋放后才能獲取。

三、指定共享線程數(shù)目的共享鎖

（1）代碼實(shí)現(xiàn)（核心關(guān)鍵實(shí)現(xiàn)已經(jīng)在代碼中注釋）

public class MyShareLock implements Lock {// 可以看到共享等待隊(duì)列中的線程public Collection<Thread> getSharedQueuedThreads(){return syn.getSharedQueuedThreads();
    }private final Syn syn = new Syn(2);private static final class Syn extends AbstractQueuedSynchronizer{int newShareCount=0;
        Syn(int shareCount){if (shareCount <= 0) {throw new IllegalArgumentException("share count must large than zero");
            }// 設(shè)置初始共享同步狀態(tài)            setState(shareCount);
        }/** * 共享鎖指定數(shù)目
         * @param reduceShareCount
         * @return */@Overrideprotected int tryAcquireShared(int reduceShareCount) {for (;;){int currentShareCount = getState();
                newShareCount = currentShareCount- reduceShareCount;if (newShareCount < 0 ||compareAndSetState(currentShareCount,newShareCount)) {// newShareCount大于等于０才說明獲取鎖成功if (newShareCount >= 0) {//                        System.out.println(Thread.currentThread().getName()+" hold lock, current share count is "+newShareCount+", "+new Date());                    }// newShareCount小于0表示獲取失敗所以需要返回// compareAndSetState(currentShareCount,newShareCount)為true自然表示成功需要返回return newShareCount;
                }
            }
        }

        @Overrideprotected boolean tryReleaseShared(int returnShareCount) {for (;;){int currentShareCount = getState();
                newShareCount = currentShareCount + returnShareCount;if (compareAndSetState(currentShareCount,newShareCount)) {//                    System.out.println(Thread.currentThread().getName() +" release lock, current share count is "+newShareCount+", "+new Date());return true;
                }
            }
        }protected int getShareCount(){return getState();
        }
    }/** * 調(diào)用內(nèi)部同步器Syn的acquireShare方法     */@Overridepublic void lock() {
        syn.acquireShared(1);
    }/** * 調(diào)用內(nèi)部同步器Syn的releaseShared方法     */@Overridepublic void unlock() {
        syn.releaseShared(1);
    }

    @Overridepublic void lockInterruptibly() throws InterruptedException {if (Thread.interrupted()) {throw new IllegalStateException();
        }
        syn.acquireInterruptibly(1);
    }

    @Overridepublic boolean tryLock() {return false;
    }

    @Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return false;
    }


    @Overridepublic Condition newCondition() {return null;
    }
}

（2）測試Demo

public class ShareLockTest {

    @Testpublic void test(){final MyShareLock lock = new MyShareLock();class Worker extends Thread {
            @Overridepublic void run() {// 一直不停在獲取鎖while (true) {
                lock.lock();try {
                    System.out.println(Thread.currentThread().getName() +" hold lock, "+new Date());//                    System.out.println(lock.getSharedQueuedThreads());Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                    System.out.println(Thread.currentThread().getName() +" release lock, "+new Date());
                }
            }
            }

        }for (int i = 0; i < 10; i++) {
            Worker worker = new Worker();// 以守護(hù)進(jìn)程運(yùn)行，VM退出不影響運(yùn)行，這里只是為了一個打印效果，去掉注釋一直打印worker.setDaemon(true);
            worker.start();
        }// 每隔一秒換行for (int j = 0; j < 10; j++) {try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println();
        }
    }
}

（3）運(yùn)行結(jié)果（結(jié)果可能不同）

Thread-1 hold lock, Tue Dec 08 16:36:05 CST 2020Thread-0 hold lock, Tue Dec 08 16:36:05 CST 2020Thread-0 release lock, Tue Dec 08 16:36:06 CST 2020Thread-4 hold lock, Tue Dec 08 16:36:06 CST 2020Thread-1 release lock, Tue Dec 08 16:36:06 CST 2020Thread-2 hold lock, Tue Dec 08 16:36:06 CST 2020Thread-4 release lock, Tue Dec 08 16:36:07 CST 2020Thread-2 release lock, Tue Dec 08 16:36:07 CST 2020Thread-5 hold lock, Tue Dec 08 16:36:07 CST 2020Thread-8 hold lock, Tue Dec 08 16:36:07 CST 2020Thread-8 release lock, Tue Dec 08 16:36:08 CST 2020Thread-3 hold lock, Tue Dec 08 16:36:08 CST 2020Thread-9 hold lock, Tue Dec 08 16:36:08 CST 2020Thread-5 release lock, Tue Dec 08 16:36:08 CST 2020Thread-6 hold lock, Tue Dec 08 16:36:09 CST 2020Thread-7 hold lock, Tue Dec 08 16:36:09 CST 2020Thread-3 release lock, Tue Dec 08 16:36:09 CST 2020Thread-9 release lock, Tue Dec 08 16:36:09 CST 2020Thread-6 release lock, Tue Dec 08 16:36:10 CST 2020Thread-1 hold lock, Tue Dec 08 16:36:10 CST 2020Thread-0 hold lock, Tue Dec 08 16:36:10 CST 2020Thread-7 release lock, Tue Dec 08 16:36:10 CST 2020Thread-1 release lock, Tue Dec 08 16:36:11 CST 2020Thread-2 hold lock, Tue Dec 08 16:36:11 CST 2020Thread-0 release lock, Tue Dec 08 16:36:11 CST 2020Thread-4 hold lock, Tue Dec 08 16:36:11 CST 2020Thread-2 release lock, Tue Dec 08 16:36:12 CST 2020Thread-8 hold lock, Tue Dec 08 16:36:12 CST 2020Thread-5 hold lock, Tue Dec 08 16:36:12 CST 2020Thread-4 release lock, Tue Dec 08 16:36:12 CST 2020Thread-5 release lock, Tue Dec 08 16:36:13 CST 2020Thread-9 hold lock, Tue Dec 08 16:36:13 CST 2020Thread-3 hold lock, Tue Dec 08 16:36:13 CST 2020Thread-8 release lock, Tue Dec 08 16:36:13 CST 2020Thread-3 release lock, Tue Dec 08 16:36:14 CST 2020Thread-7 hold lock, Tue Dec 08 16:36:14 CST 2020Thread-9 release lock, Tue Dec 08 16:36:14 CST 2020Thread-6 hold lock, Tue Dec 08 16:36:14 CST 2020

（4）結(jié)果分析

該指定共享線程數(shù)量N的共享鎖的最終目的就是多個線程可以持有鎖（同步狀態(tài)），達(dá)到共享線程數(shù)量N（代碼中默認(rèn)為2）時，其它線程將進(jìn)入Queue等待獲取同步結(jié)果，同一時刻只能最多有N個線程持有鎖。

同樣地，我們分析開頭運(yùn)行結(jié)果：

Thread-1 hold lock, Tue Dec 08 16:36:05 CST 2020Thread-0 hold lock, Tue Dec 08 16:36:05 CST 2020Thread-0 release lock, Tue Dec 08 16:36:06 CST 2020Thread-4 hold lock, Tue Dec 08 16:36:06 CST 2020Thread-1 release lock, Tue Dec 08 16:36:06 CST 2020Thread-2 hold lock, Tue Dec 08 16:36:06 CST 2020

10個線程不停競爭鎖，一開始Thread-0與Thread-1在16:36:05時刻同時獲取到了鎖，此時已經(jīng)達(dá)到共享數(shù)量的最大值，即N，之后持有鎖1秒，Thread-0與Thread-1在16:36:06時刻立馬釋放鎖，同時Thread-4與Thread-2立馬退出等待隊(duì)列立馬競爭持有鎖。

從結(jié)果來看，完全是符合ShareLock共享鎖功能的：同一時刻最多允許N個線程持有鎖，其它線程等待持有線程釋放鎖！

本站是提供個人知識管理的網(wǎng)絡(luò)存儲空間，所有內(nèi)容均由用戶發(fā)布，不代表本站觀點(diǎn)。請注意甄別內(nèi)容中的聯(lián)系方式、誘導(dǎo)購買等信息，謹(jǐn)防詐騙。如發(fā)現(xiàn)有害或侵權(quán)內(nèi)容，請點(diǎn)擊一鍵舉報(bào)。

轉(zhuǎn)藏 分享

QQ空間 QQ好友新浪微博微信

獻(xiàn)花（0） +1

來自：丹楓無跡 > 《待分類》

舉報(bào)/認(rèn)領(lǐng)