利用LockSupport實(shí)現(xiàn)互斥鎖和共享鎖

前言

首先說(shuō)說(shuō)LockSupport吧,它的作用是提供一組直接block或unblock線(xiàn)程的方法,其底層實(shí)現(xiàn)利用了Unsafe(前面文章有講過(guò)Unsafe)。LockSupport是一個(gè)非常底層的API,我們利用其可以做很多事情,本文將利用LockSupport實(shí)現(xiàn)互斥鎖和共享鎖。

Lock

在JDK中已經(jīng)提供了很多種鎖的實(shí)現(xiàn),原生的synchronized(優(yōu)先推薦使用),juc中的ReentrantLock等,本文不糾結(jié)synchronized和ReentrantLock的實(shí)現(xiàn),本文只從Lock的語(yǔ)義出發(fā)實(shí)現(xiàn)兩種鎖。

Lock的語(yǔ)義

juc中對(duì)于Lock接口的定義如下:

    void lock();

    void lockInterruptibly() throws InterruptedException;

    boolean tryLock();

    boolean tryLock(long var1, TimeUnit var3) throws InterruptedException;

    void unlock();

    Condition newCondition();
  • void lock():獲取鎖的語(yǔ)義,如果沒(méi)有獲取到鎖一直阻塞當(dāng)前線(xiàn)程(不響應(yīng)中斷interrupt)
  • void lockInterruptibly() throws InterruptedException; 獲取鎖,但是當(dāng)前線(xiàn)程在阻塞期間可以響應(yīng)中斷(后面稍微會(huì)扯一下InterruptedException)
  • boolean tryLock(); 嘗試獲取鎖,不阻塞;獲取到鎖返回true,沒(méi)有獲取到返回false
  • boolean tryLock(long var1, TimeUnit var3) throws InterruptedException; 嘗試獲取鎖,并嘗試阻塞等待一定時(shí)間,阻塞期間可以響應(yīng)中斷
  • void unlock(); 釋放鎖;
  • Condition newCondition();在鎖上新建Condition

以上的關(guān)于鎖的語(yǔ)義稍微復(fù)雜了點(diǎn),特別是相應(yīng)中斷部分和newCondition部分,所以這次實(shí)現(xiàn)上簡(jiǎn)化了Lock的語(yǔ)義如下:

    void lock();

    void unLock();

    boolean tryLock();

    boolean tryLock(long maxWaitInMills);

基本功能和上面保持一致,但是都不響應(yīng)中斷

分析鎖的實(shí)現(xiàn)

  • Lock有可重入的語(yǔ)義,一個(gè)線(xiàn)程擁有鎖之后再次調(diào)用lock應(yīng)該完全沒(méi)有任何問(wèn)題,所以鎖的實(shí)現(xiàn)中需要維護(hù)一個(gè)已經(jīng)獲取鎖的線(xiàn)程隊(duì)列;
  • Lock未成功需要阻塞當(dāng)前線(xiàn)程,所以需要底層阻塞原語(yǔ)(LockSupport)等的支持,并且在有線(xiàn)程釋放鎖之后需要喚起阻塞線(xiàn)程進(jìn)行鎖的競(jìng)爭(zhēng),所以需要維護(hù)等待鎖的線(xiàn)程隊(duì)列
  • Lock需要維護(hù)當(dāng)前鎖的狀態(tài)(是否可以被獲取等)

互斥鎖

public class MutexLock implements Lock {


    private volatile Thread threadOwnsTheLock;

    private final AtomicInteger state = new AtomicInteger(0);

    private final ConcurrentLinkedQueue<Thread> waitThreadsQueue = new ConcurrentLinkedQueue<Thread>();


    //一直等待
    public void lock() {
        tryLock(-1L);
    }

    //invoke all的語(yǔ)義,也可以做invokeNext
    public void unLock() {
        tryRelease(-1);
        threadOwnsTheLock = null;
        if (!waitThreadsQueue.isEmpty()) {
            for (Thread thread : waitThreadsQueue) {
                LockSupport.unpark(thread);
            }
        }
    }

    public boolean tryLock() {
        if (threadOwnsTheLock != null && (threadOwnsTheLock == Thread.currentThread())) {
            return true;
        }
        if (tryAcquire(1)) {
            threadOwnsTheLock = Thread.currentThread();
            return true;
        }

        return false;
    }

    //沒(méi)有實(shí)現(xiàn)interrupt的語(yǔ)義,不能打斷
    public boolean tryLock(long maxWaitInMills) {
        Thread currentThread = Thread.currentThread();
        try {
            waitThreadsQueue.add(currentThread);
            if (maxWaitInMills > 0) {
                boolean acquired = false;
                long left = maxWaitInMills * 1000L * 1000L;
                long cost = 0;
                while (true) {
                    //需要判斷一次interrupt

                    if (tryAcquire(1)) {
                        threadOwnsTheLock = currentThread;
                        acquired = true;
                        break;
                    }

                    left = left - cost;
                    long mark = System.nanoTime();
                    if (left <= 0) {
                        break;
                    }
                    LockSupport.parkNanos(left);
                    cost = mark - System.nanoTime();
                }
                return acquired;
            }else {
                while (true) {
                    if (tryAcquire(1)) {
                        threadOwnsTheLock = currentThread;
                        break;
                    }
                    LockSupport.park();
                }
                return true;
            }
        } finally {
            waitThreadsQueue.remove(currentThread);
        }

    }

    protected boolean tryAcquire(int acquire) {
        return state.compareAndSet(0, 1);
    }

    protected void tryRelease(int release) {
        if (threadOwnsTheLock == null || (threadOwnsTheLock != Thread.currentThread())) {
            System.out.println("Wrong state, this thread don't own this lock.");
        }
        while (true) {
            if (state.compareAndSet(1, 0)) {
                return;
            }
        }
    }
}

以上互斥鎖使用了一個(gè)AtomicInteger,利用了CAS來(lái)維持鎖的狀態(tài)

共享鎖

public class ShareLock implements Lock {

    private volatile Set<Thread> threadsOwnsLock = Sets.newConcurrentHashSet();

    private final AtomicInteger state;

    private final ConcurrentLinkedQueue<Thread> waitThreadsQueue = new ConcurrentLinkedQueue<Thread>();

    public ShareLock(int shareNum) {
        this.state = new AtomicInteger(shareNum);
    }


    //一直等待
    public void lock() {
        tryLock(-1L);
    }

    public void unLock() {
        tryRelease(-1);
        threadsOwnsLock.remove(Thread.currentThread());
        if (!waitThreadsQueue.isEmpty()) {
            for (Thread thread : waitThreadsQueue) {
                LockSupport.unpark(thread);
            }
        }
    }

    public boolean tryLock() {
        if ( !(threadsOwnsLock.contains(Thread.currentThread()))) {
            return true;
        }
        if (tryAcquire(1)) {
            threadsOwnsLock.add(Thread.currentThread());
            return true;
        }

        return false;
    }

    public boolean tryLock(long maxWaitInMills) {


        Thread currentThread = Thread.currentThread();
        try {
            waitThreadsQueue.add(currentThread);
            if (maxWaitInMills > 0) {
                boolean acquired = false;
                long left = TimeUnit.MILLISECONDS.toNanos(maxWaitInMills);
                long cost = 0;
                while (true) {
                    if (tryAcquire(1)) {
                        threadsOwnsLock.add(Thread.currentThread());
                        acquired = true;
                        break;
                    }

                    left = left - cost;
                    long mark = System.nanoTime();
                    if (left <= 0) {
                        break;
                    }
                    LockSupport.parkNanos(left);
                    cost = mark - System.nanoTime(); //有可能是被喚醒重新去獲取鎖,沒(méi)獲取到還得繼續(xù)等待剩下的時(shí)間(并不精確)
                }
                return acquired;
            }else {
                while (true) {
                    if (tryAcquire(1)) {
                        threadsOwnsLock.add(Thread.currentThread());
                        break;
                    }
                    LockSupport.park();
                }
                return true;
            }
        } finally {
            waitThreadsQueue.remove(currentThread);
        }

    }

    protected boolean tryAcquire(int acquire) {
        if (state.getAndDecrement() > 0) {
            return true;
        } else {
            state.getAndIncrement();//恢復(fù)回來(lái)
            return false;
        }
    }

    protected void tryRelease(int release) {
        if (!(threadsOwnsLock.contains(Thread.currentThread()))) {
            System.out.println("Wrong state, this thread don't own this lock.");
        }
        state.getAndIncrement();
    }
}

總結(jié)

以上利用了LockSupport來(lái)實(shí)現(xiàn)了互斥鎖和共享鎖,但是實(shí)現(xiàn)中并沒(méi)有完成中斷響應(yīng)。后面應(yīng)該會(huì)有文章單獨(dú)說(shuō)明關(guān)于InterruptedException的注意點(diǎn)。下篇文章將講述如何利用LockSupport實(shí)現(xiàn)Future語(yǔ)義

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀(guān)點(diǎn),簡(jiǎn)書(shū)系信息發(fā)布平臺(tái),僅提供信息存儲(chǔ)服務(wù)。

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

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