【JUC源码】并发工具：Semaphore 源码分析&应用示例

小灰灰 2023-01-13 04:17 203阅读 0赞

Semaphore 字面意思是信号量的意思，它的作用是控制访问特定资源的线程数目。

## 1.应用示例 ##

场景：资源访问，服务限流

public class SemaphoreDemo {
        
    
        public static void main(String[] args) {
        
        	// 限流，同一时刻最多能有两个线程执行
            Semaphore semaphore = new Semaphore(2);
            for (int i = 0; i < 5; i ++) {
        
                new Thread(new Task(semaphore, "zhangsan +" + i)).start();
            }
        }
    
        static class Task extends Thread {
        
            Semaphore semaphore;
            public Task(Semaphore semaphore, String threadname) {
        
                this.semaphore = semaphore;
                this.setName(threadname);
            }
    
            public void run() {
        
                try {
        
                	// 申请资源（同时最多两个thread）
                    semaphore.acquire();
                    System.out.println(Thread.currentThread().getName() + ": acquire() at time:" + System.currentTimeMillis());
                    Thread.sleep(1000);
                    // 释放资源
                    semaphore.release();
                    System.out.println(Thread.currentThread().getName() + ": release() at time:" + System.currentTimeMillis());
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
    
            }
        }
    
    }

结果如下：

Thread-3: acquire() at time:1618487257823
    Thread-1: acquire() at time:1618487257823
    Thread-3: release() at time:1618487258824
    Thread-5: acquire() at time:1618487258824
    Thread-1: release() at time:1618487258824
    Thread-9: acquire() at time:1618487258824
    Thread-7: acquire() at time:1618487259825
    Thread-9: release() at time:1618487259825
    Thread-5: release() at time:1618487259825
    Thread-7: release() at time:1618487260825

可以看到两点

*  同一时刻最多只能有两个线程同时执行
 *  只有一个线程 release() 后，才能有一个新线程 acquire()

## 2.源码分析 ##

Semaphore 的源码实现其实跟 CountDownLatch 很相似，都是通过直接复用 AQS 来实现的（内部类 Sync）；

> PS：关于 CountDownLatch 的使用示例及源码分析，请看[这篇文章][Link 1]…

我们先来看看 Semaphore 的构造函数做了什么

public Semaphore(int permits) {
        
        sync = new NonfairSync(permits);
    }

Semaphore 的构造函数实际上是构造了一个 NonfairSync（非公平锁），并传入了我们预设的最大允许执行线程数  
![在这里插入图片描述][20210416142729818.png]  
可以看到，NonfairSync 实际上是调用了 AQS 的构造函数，是将 AQS 的 state 从默认的 1 改成了我们预设的 permits。这也正是 Semaphore 能同时多个线程执行的关键所在。

### 2.1 acquire() ###

当使用 Semaphore 后，要申请资源时调用 acquire() 方法

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

再来看 acquireSharedInterruptibly() 方法，该方法是在 AQS 中的，Semaphore 的内部类 Sync 并没有自己实现/重写：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]  
这里有两个核心方法，tryAcquireShared() 和 doAcquireInterruptibly()，它俩的关系是，当 tryAcquireShared() 判断无法拿到锁时，就会进入 AQS 的 doAcquireInterruptibly() 方法。

**tryAcquireShared()**

这个方法最终会走到内部类 Sync 的 nonfairTryAcquireShared() 方法

// Semaphore.Sync
    final int nonfairTryAcquireShared(int acquires) {
        
    	// 自旋
        for (;;) {
        
            int available = getState();
            // 计算修改后的 AQS 的 state（减去当前线程要申请的资源数）
            int remaining = available - acquires;
            //  当修改后的 state 大于 0，且 CAS 能够修改成功时，可以拿锁
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

**doAcquireSharedInterruptibly()**

// AQS
    private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
        
        	// 将当前线程封装为node（共享模式），并加到同步队列队尾
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
        
            	// 自旋，保证所有被唤醒的线程都能依次恢复运行
                for (;;) {
        
                    final Node p = node.predecessor();
                    // 当前node前进到队二 && tryAcquire成功（state减到0），就可以执行了
                    if (p == head) {
        
                    	// 判断是否能拿到锁
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
        
                        	// setHeadAndPropagate 会调用 doReleaseShared 去唤醒后续 Shared 节点
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    // 其余线程阻塞（最后也是在此处醒来）
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
        
                if (failed)
                    cancelAcquire(node);
            } 
    }

> 注：虽然是共享锁模式，一个线程拿锁之后会唤醒同步队列中后续的 Shared 节点，但醒了后还要判断是否能拿到锁（tryAcquireShared），所以还会再次陷入休眠。（其实我觉得这里或许可以用互斥锁模式，欢迎讨论。。。）

### 2.2 release() ###

当使用 Semaphore，某个线程要释放资源时，调用 release()

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

**releaseShared()**

// AQS
    public final boolean releaseShared(int arg) {
        
    	// 判断能否释放锁（修改state）
        if (tryReleaseShared(arg)) {
        
        	// 唤醒后续线程
            doReleaseShared();
            return true;
        }
        return false;
    }

**tryReleaseShared()**

// Semaphore.Sync
    protected final boolean tryReleaseShared(int releases) {
        
    	// 自旋，保证 state 修改一定能成功
        for (;;) {
        
            int current = getState();
            // 将 state 加上当前线程锁持有的资源数
            int next = current + releases;
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            // CAS 修改 state
            if (compareAndSetState(current, next))
                return true;
        }
    }

**doReleaseShared()**

// AQS
    private void doReleaseShared() {
        
        // 自旋，保证所有线程正常的线程都能被唤醒
        for (;;) {
        
            Node h = head;
            // 还没有到队尾，此时队列中至少有两个节点
            if (h != null && h != tail) {
        
                int ws = h.waitStatus;
                // 如果头结点状态是 SIGNAL ，说明后续节点都需要唤醒
                if (ws == Node.SIGNAL) {
        
                    // CAS 保证只有一个节点可以运行唤醒的操作
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 进行唤醒操作
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            // 退出自旋条件 h==head，一般出现于以下两种情况
            // 第一种情况，头节点没有发生移动，结束。
            // 第二种情况，因为此方法可以被两处调用，一次是获得锁的地方，一处是释放锁的地方，
            // 加上共享锁的特性就是可以多个线程获得锁，也可以释放锁，这就导致头节点可能会发生变化，
            // 如果头节点发生了变化，就继续循环，一直循环到头节点不变化时，结束循环。
            if (h == head)                   // loop if head changed
                break;
        }
    }

[Link 1]: https://yzhyaa.blog.csdn.net/article/details/108718739
[20210416142729818.png]: /images/20221022/42077812ca1e483eaf22119d29930e07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221022/37292061107d48f4b58efb561a8c4bac.png