前提概要

之前的文章中會(huì)涉及到了相關(guān)AQS的原理和相關(guān)源碼的分析，所謂實(shí)踐是檢驗(yàn)真理的唯一標(biāo)準(zhǔn)！接下來(lái)就讓我們活化一下AQS技術(shù)，主要針對(duì)于自己動(dòng)手實(shí)現(xiàn)一個(gè)AQS同步器。

定義MyLock實(shí)現(xiàn)Lock

Doug Lea大神在JDK1.5編寫了一個(gè)Lock接口，里面定義了實(shí)現(xiàn)一個(gè)鎖的基本方法，我們只需編寫一個(gè)MyLock類實(shí)現(xiàn)這個(gè)接口就好。

class MyLock implements Lock {
    /**
     * 加鎖。如果不成功則進(jìn)入等待隊(duì)列
     */
    @Override
    public void lock() {}
    /**
    * 加鎖(可被interrupt)
    */
    @Override
    public void lockInterruptibly() throws InterruptedException {}
    /**
     * 嘗試加鎖
     */
    @Override
    public boolean tryLock() {}
    /**
     * 加鎖 帶超時(shí)的
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {}
    /**
    * 釋放鎖
    */
    @Override
    public void unlock() {}
    /**
    * 返回一個(gè)條件變量(不在本案例談?wù)?
    */
    @Override
    public Condition newCondition() {}
}
復(fù)制代碼

定義好MyLock后，接下來(lái)就是實(shí)現(xiàn)各個(gè)方法的邏輯，達(dá)到真正的用于線程間sync互斥的需求。

自定義一個(gè)MySync繼承自AQS

接下來(lái)我們需要自定義一個(gè)繼承自AQS的MySync。實(shí)現(xiàn)自定義的MySync前，先了解AQS內(nèi)部的一些基本概念。在AQS中主要的一些成員屬性如下：

• state：用于標(biāo)記資源狀態(tài)，如果為0表示資源沒(méi)有被占用，可以加鎖成功。如果大于0表示資源已經(jīng)被占用，然后根據(jù)自己的定義去實(shí)現(xiàn)是否允許對(duì)共享資源進(jìn)行操作。 比如：ReentrantLock的實(shí)現(xiàn)方式是當(dāng)state大于0，那么表示已經(jīng)有線程獲得鎖了，我們都知道ReentrantLock是可重入的，其原理就是當(dāng)有線程次進(jìn)入同一個(gè)lock標(biāo)記的臨界區(qū)時(shí)。先判斷這個(gè)線程是否是獲得鎖的那個(gè)線程，如果是，state會(huì)+1，此時(shí)state會(huì)等于2。 當(dāng)unlock時(shí)，會(huì)一層一層地減1，直到state等于0則表示完全釋放鎖成功。
• head、tail：用于存放獲得鎖失敗的線程。在AQS中，每一個(gè)線程會(huì)被封裝成一個(gè)Node節(jié)點(diǎn)，這些節(jié)點(diǎn)如果獲得鎖資源失敗會(huì)鏈在head、tail中，成為一個(gè)雙向鏈表結(jié)構(gòu)。
• exclusiveOwnerThread：用于存放當(dāng)前獲得鎖的線程，正如在state說(shuō)明的那樣。ReentrantLock判斷可重入的條件就是用這個(gè)exclusiveOwnerThread線程跟申請(qǐng)獲得鎖的線程做比較，如果是同一個(gè)線程，則state+1,并重入加鎖成功。

知道這些概念后我們就可以自定義一個(gè)AQS：

public final class MySync extends AbstractQueuedSynchronizer {
    /**
    * 嘗試加鎖
    */
    @Override
    protected boolean tryAcquire(int arg) {
        if (compareAndSetState(0, 1)) {
            // 修改state狀態(tài)成功后設(shè)置當(dāng)前線程為占有鎖資源線程
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
    /**
    * 釋放鎖
    */
    @Override
    protected boolean tryRelease(int arg) {
        setExclusiveOwnerThread(null);
        // state有volatile修飾，為了保證解鎖后其他的一些變量對(duì)其他線程可見(jiàn)，把setExclusiveOwnerThread(null)放到上面 hAppens-before中定義的 volatile規(guī)則
        setState(0);
        return true;
    }
    /**
    * 判斷是否是獨(dú)占鎖
    */
    @Override
    protected boolean isHeldExclusively() {
        return getState() == 1;
    }
}
復(fù)制代碼

將MySync組合進(jìn)MyLock

最后一步就是將第一步中的所有方法邏輯完成

class MyLock implements Lock {

    // 組合自定義sync器
    private MySync sync = new MySync();

    /**
     * 加鎖。如果不成功則進(jìn)入等待隊(duì)列
     */
    public void lock() {
        sync.acquire(1);
    }
    /**
    * 加鎖(可被interrupt)
    */
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    /**
     * 嘗試加鎖
     */
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }
    /**
     * 加鎖 帶超時(shí)的
     */
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toMillis(time));
    }
    /**
    * 釋放鎖
    */
    public void unlock() {
        sync.release(0);
    }
    /**
    * 返回一個(gè)條件變量(不在本案例談?wù)?
    */
    @Override
    public Condition newCondition() {
        return null;
    }
}
復(fù)制代碼

完成整個(gè)MyLock的邏輯后，發(fā)現(xiàn)在lock()、unlock()中調(diào)用的自定義sync的方法tryAcquire()和tryRelease()方法。我們就以在lock()方法中調(diào)用acquire()方法說(shuō)明模板設(shè)計(jì)模式在AQS中的應(yīng)用。

點(diǎn)進(jìn).acquire()方法后，發(fā)現(xiàn)該該方法是來(lái)自
AbstractQueuedSynchronizer中：

• 在這里面可以看到tryAcquire方法，繼續(xù)點(diǎn)進(jìn)去看看tryAcquire()，發(fā)現(xiàn)該方法是一個(gè)必須被重寫的方法，否則拋出一個(gè)運(yùn)行時(shí)異常。
• 模板方法設(shè)計(jì)模式在這里得以體現(xiàn)，再回到我們第二部中自定義的MySync中，就是重寫了AQS中的tryAcquire()方法。

因此整個(gè)自定義加鎖的流程如下：

• 調(diào)用MyLock的lock()，lock()方法調(diào)用AQS的acquire()方法
• 在acquire()方法中調(diào)用了tryAcquire()方法進(jìn)行加鎖
• 而tryAcquire()方法在AQS中是一個(gè)必須讓子類自定義重寫的方法，否則會(huì)拋出一個(gè)異常
• 因此調(diào)用tryAcquire()時(shí)實(shí)際上是調(diào)用了我們自定義的MySync類中tryAcquire()方法

總結(jié)

AQS作為JAVA并發(fā)體系下的關(guān)鍵類，在各種并發(fā)工具中都有它的身影，如ReentrantLock、Semaphore等。這些并發(fā)工具用于控制sync互斥的手段都是采用AQS，外加Cas機(jī)制。AQS采用了模板方法設(shè)計(jì)模式讓子類們自定義sync互斥的條件，比如本案例中MySync類重寫了tryAcquire方法。

下面實(shí)現(xiàn)一個(gè)自定義的sync：

public class SelfSynchronizer {

    private final Sync sync = new Sync();

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

    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    public boolean unLock() {
        return sync.release(1);
    }

    static class Sync extends AbstractQueuedSynchronizer {
        //是否處于占用狀態(tài)
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        /**
         * 獲取sync資源
         * @param acquires
         * @return
         */
        @Override
        public boolean tryAcquire(int acquires) {
            if(compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            //這里沒(méi)有考慮可重入鎖
            /*else if (Thread.currentThread() == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }*/
            return false;
        }

        /**
         * 釋放sync資源
         * @param releases
         * @return
         */
        @Override
        protected boolean tryRelease(int releases) {
            int c = getState() - releases;
            boolean free = false;
            if (c == 0) {
                free = true;
            }
            setState(c);
            return free;
        }
    }
}

復(fù)制代碼

ReentrantLock源碼和上面自定義的sync很相似，測(cè)試下該sync，i++在多線程下執(zhí)行情況：

public class TestSelfSynchronizer {
    private static int a = 0;
    private static int b = 0;
    private static SelfSynchronizer selfSynchronizer = new SelfSynchronizer();
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 50, 1, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>());
    private static ExecutorService ec = Executors.newFixedThreadPool(20);
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 20 ; i++) {
            executor.submit(new Task());
        }
        for (int j = 0; j < 20 ; j++) {
            ec.submit(new TaskSync());
        }
        Thread.sleep(10000);
        System.out.println("a的值："+ a);
        System.out.println("b的值" + b);
        executor.shutdown();
        ec.shutdown();
    }
    static class Task implements Runnable {
        @Override
        public void run() {
            for(int i=0;i<10000;i++) {
                a++;
            }
        }
    }
    static class TaskSync implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10000; i++) {
            	//使用sync器加鎖
                selfSynchronizer.lock();
                b++;
                selfSynchronizer.unLock();
            }
        }
    }
}
復(fù)制代碼

開(kāi)啟兩個(gè)線程池，對(duì)int型變量自增10000次，如果不加sync器，最后值小于200000，使用了自定義sync器則最后值正常等于200000，這是因?yàn)槊看巫栽霾僮骷渔i

作者：李浩宇A(yù)lex
鏈接：
https://juejin.cn/post/6989937347429302280
來(lái)源：掘金
著作權(quán)歸作者所有。商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請(qǐng)注明出處。