并发编程（九）：AQS之Semaphore

朱雀 2023-06-04 10:59 94阅读 0赞

# 一，底层AQS源码分析：[并发编程（四）：AbstractQueuedSynchronizer源码分析][AbstractQueuedSynchronizer] #

## 二，Semaphore介绍 ##

##     1，线程访问控制 ##

*Semaphore* 也就是我们常说的信号灯，通过初始化的信号量控制同时访问线程的个数。在线程调用 *acquire()* 方法时，会获取到一个许可，作为通行证继续进行线程执行，等线程执行完成后，通过调用 *release()* 方法释放许可，交给后续线程继续执行。如果线程在获取信号量时没有获取到，则会添加到AQS同步队列，并将线程挂起，等待线程唤醒。

##     2，类图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70][]

\* 从类图中可以看到，*Semaphore* 内部定义了内部类 *Sync*。与之前分析的其他线程同步工具一样，*Sync* 类继承自AQS，并派生出两个子类 *FairSync*（公平锁）和 *NonfairSync*（非公平锁）。所以，Semaphore 的线程通信也是基于AQS完成

##     3，常用API ##

// 初始化信号灯，并定义许可数量，默认非公平锁
    public Semaphore(int permits);
    // 初始化信号灯，定义许可数量，并指定使用非公平锁还是公平锁
    public Semaphore(int permits, boolean fair);
    // 尝试获取线程许可，默认获取一个
    public boolean tryAcquire();
    // 尝试获取线程许可，一次性获取指定个
    public boolean tryAcquire(int permits);
    // 尝试限时获取线程许可，默认获取一个
    public boolean tryAcquire(long timeout, TimeUnit unit);
    // 尝试限时获取线程许可，一次性获取指定个
    public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
    // 获取线程许可，默认获取一个
    public void acquire() throws InterruptedException;
    // 获取线程许可，一次性获取指定个
    public void acquire(int permits) throws InterruptedException;
    // 释放线程需求，默认释放一个
    public void release();
    // 释放线程需求，一次性释放多个
    public void release(int permits);

##     4，功能DEMO ##

package com.gupao.concurrent;
    
    import java.util.concurrent.Semaphore;
    
    /**
     * Semaphore信号灯
     *
     * @author pj_zhang
     * @create 2019-10-11 21:34
     **/
    public class SemaphoreTest {
    
        public static void main(String[] args) throws InterruptedException {
            // 初始化 Semaphore 信号灯，且信号量为3
            Semaphore semaphore = new Semaphore(3);
            for (int i = 0; i < 6; i++) {
                new Thread(() -> {
                    try {
                        // 获取一个信号，并执行
                        semaphore.acquire();
                        // 线程沉睡1S再释放，
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "执行；时间：" + System.currentTimeMillis());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        semaphore.release();
                    }
                }, "THREAD_" + i).start();
    
            }
    
        }
    
    }

![20191011215827131.png][]

# 二，源码分析（统一基于多许可分析） #

##     1，初始化源码分析 ##

###         1.1，不指定锁类型 ###

\* Semaphore(int permits)

public Semaphore(int permits) {
    	// Sync默认初始化为非公平锁
    	// 并且传递许可数量permits
    	sync = new NonfairSync(permits);
    }

\* NonfairSync(int permits)

NonfairSync(int permits) {
    	// 调用父类初始化方法，及Semaphore.Sync
    	super(permits);
    }
    
    Sync(int permits) {
    	// 最终将permits值设置到state中
    	setState(permits);
    }

###         1.2，指定锁类型 ###

\* Semaphore(int permits, boolean fair)

public Semaphore(int permits, boolean fair) {
    	// 判断初始化公平锁还是分公平锁
    	sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

##     2，尝试获取许可源码分析 ##

###         2.1，不限时尝试源码分析 ###

\* tryAcquire(int permits)

public boolean tryAcquire(int permits) {
    	// 参数合法性校验
    	if (permits < 0) throw new IllegalArgumentException();
    	// 直接尝试获取锁，此处不做公平锁和非公平锁区分，统一按照非公平锁处理
    	// 该方法返回当前线程获取后的许可数量，>=0说明许可数量足够，获取成功
    	return sync.nonfairTryAcquireShared(permits) >= 0;
    }

\* nonfairTryAcquireShared(int acquires)

final int nonfairTryAcquireShared(int acquires) {
    	// 自旋处理，此处自旋主要考虑CAS失败情况
    	for (;;) {
    		// 获取当前信号量
    		int available = getState();
    		// 获取当前线程后获取剩余的信号量
    		int remaining = available - acquires;
    		// 返回信号量，因为线程竞争，CAS可能会失败
            // ||会进行短路，所以如果remaining<0，则不会进行CAS替换
    		if (remaining < 0 ||
    			compareAndSetState(available, remaining))
    			return remaining;
    	}
    }

###         2.2，限时尝试源码分析 ###

\* tryAcquire(int permits, long timeout, TimeUnit unit)

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
            throws InterruptedException {
    	// 参数合法性校验
    	if (permits < 0) throw new IllegalArgumentException();
    	// 限时尝试获取共享锁
    	return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }

\* tryAcquireSharedNanos(int arg, long nanosTimeout)

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
    		throws InterruptedException {
    	// 线程中断判断
    	if (Thread.interrupted())
    		throw new InterruptedException();
    	// tryAcquireShared：尝试获取共享锁，该方法在Semaphore中重新定义
    	// doAcquireSharedNanos：线程等待获取锁
    	return tryAcquireShared(arg) >= 0 ||
    		doAcquireSharedNanos(arg, nanosTimeout);
    }

\* tryAcquireShared(int acquires)：公平锁

protected int tryAcquireShared(int acquires) {
    	for (;;) {
    		// 判断AQS同步队列中是否有等待节点，有则直接失败，添加AQS队列尾部
    		if (hasQueuedPredecessors())
    			return -1;
    		// AQS同步队列中不存在节点，再进行许可量判断获取
    		int available = getState();
    		int remaining = available - acquires;
    		if (remaining < 0 ||
    			compareAndSetState(available, remaining))
    			return remaining;
    	}
    }

\* tryAcquireShared(int acquires)：非公平锁

protected int tryAcquireShared(int acquires) {
    	// 直接调用获取，不再分析
    	return nonfairTryAcquireShared(acquires);
    }

##     3，获取许可源码分析 ##

\* acquire(int permits)

public void acquire(int permits) throws InterruptedException {
    	// 参数合法性校验
    	if (permits < 0) throw new IllegalArgumentException();
    	// 获取共享锁
    	sync.acquireSharedInterruptibly(permits);
    }

\* acquireSharedInterruptibly(int arg)

public final void acquireSharedInterruptibly(int arg)
    		throws InterruptedException {
    	// 线程中断校验
    	if (Thread.interrupted())
    		throw new InterruptedException();
    	// 尝试获取共享锁
    	if (tryAcquireShared(arg) < 0)
    		// 获取失败后添加CAS队列
    		doAcquireSharedInterruptibly(arg);
    }

##     4，释放许可源码分析 ##

\* release(int permits)

public void release(int permits) {
    	// 参数合法性校验
    	if (permits < 0) throw new IllegalArgumentException();
    	// 释放共享锁
    	sync.releaseShared(permits);
    }

\* releaseShared(int arg)

public final boolean releaseShared(int arg) {
    	// 尝试释放共享锁，在Semaphore.Sync中重定义
    	if (tryReleaseShared(arg)) {
    		// 释放共享锁，AQS释放，参考AQS源码分析
    		doReleaseShared();
    		return true;
    	}
    	return false;
    }

\* tryReleaseShared(int releases)

protected final boolean tryReleaseShared(int releases) {
    	// 自旋处理，线程竞争可能导致CAS失败
    	for (;;) {
    		// 获取当前许可数量
    		int current = getState();
    		// 添加当前线程释放的数量
    		int next = current + releases;
    		if (next < current) // overflow
    			throw new Error("Maximum permit count exceeded");
    		// 对许可数量进行CAS替换
    		// 则许可数量+N，后续线程可以继续获取许可执行
    		if (compareAndSetState(current, next))
    			return true;
    	}
    }

[AbstractQueuedSynchronizer]: https://blog.csdn.net/u011976388/article/details/101717658
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70]: /images/20230601/4f6a2109d5414c62a7fffa8edf8a3ee6.png
[20191011215827131.png]: /images/20230601/8a37dc15e6cf47f797423669d178ed83.png