Semaphore源码解析

今天药忘吃喽~ 2022-05-08 09:22 175阅读 0赞

`Semaphore` 用来控制同时访问特定资源的线程数量，使用 `AQS` 状态值 `state` 代表许可数 `permits`。

构造方法参数：

*  `permits`：许可数
 *  `fair`：是否公平信号量

public Semaphore(int permits) { 
        sync = new NonfairSync(permits);
    }
    
    public Semaphore(int permits, boolean fair) { 
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

首先来分析 `acquire` 获取锁方法

// Semaphore
    public void acquire() throws InterruptedException { 
        sync.acquireSharedInterruptibly(1);
    }

以共享可中断方式获取锁

// AbstractQueuedSynchronizer
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException { 
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

非公平方式获取共享锁

// Semaphore.NonfairSync
    protected int tryAcquireShared(int acquires) { 
        return nonfairTryAcquireShared(acquires);
    }

// Semaphore.Sync
    final int nonfairTryAcquireShared(int acquires) { 
        // 循环
        for (;;) { 
            int available = getState(); // 获取当前状态值(许可数)
            int remaining = available - acquires; // 剩余可用许可数减去需要许可数
            // 如果剩余可用许可数小于0 或 CAS设置当前剩余可用许可数成功，则返回剩余可用许可数
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

公平方式获取共享锁，添加 `hasQueuedPredecessors` 方法判断是否有先与当前线程获取锁的线程

// Semaphore.FairSync
    protected int tryAcquireShared(int acquires) { 
        for (;;) { 
            // 添加判断是否有先与当前线程获取锁的线程
            if (hasQueuedPredecessors())
                return -1;
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

分析 `release` 释放锁方法

// Semaphore
    public void release() { 
        sync.releaseShared(1);
    }

// AbstractQueuedSynchronizer
    public final boolean releaseShared(int arg) { 
        if (tryReleaseShared(arg)) { 
            doReleaseShared();
            return true;
        }
        return false;
    }

// Semaphore.Sync
    protected final boolean tryReleaseShared(int releases) { 
        for (;;) { 
            int current = getState(); // 获取当前状态值(许可数)
            int next = current + releases; // 当前剩余可用许可数加上本地释放许可数
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            if (compareAndSetState(current, next)) // CAS设置状态值(许可数)
                return true;
        }
    }