十一：深入理解 Semaphore —— 信号量

青旅半醒 2024-05-05 07:45 67阅读 0赞

#### 目录 ####

*  1、Semaphore 入门
 *   *  1.1、概念
     *  1.2、案例
 *  2、Semaphore 源码解析
 *   *  2.1、类结构
     *  2.2、\`availablePermits()\` 方法 —— Semaphore
     *  2.3、\`acquire()\` 方法 —— Semaphore【可中断】
     *   *  2.3.1、\`acquireSharedInterruptibly()\` 方法 —— AQS【共享模式；可中断】
         *   *  2.3.1.1、\`tryAcquireShared()\` 方法 —— AQS【由子类实现】
             *  2.3.1.2、\`tryAcquireShared()\` 方法 —— Semaphore.NonfairSync
             *   *  2.3.1.2.1、\`nonfairTryAcquireShared()\` 方法 —— Semaphore.Sync【非公平】
                 *  2.3.1.2.2、\`tryAcquireShared()\` 方法 —— Semaphore.FairSync【公平】
             *  2.3.1.3、\`doAcquireSharedInterruptibly()\` 方法 —— AQS【共享模式；可中断】
             *   *  2.3.1.3.1、\`setHeadAndPropagate()\` 方法 —— AQS
     *  2.4、\`release()\` 方法 —— Semaphore
     *   *  2.4.1、\`releaseShared()\` 方法 —— AQS
         *   *  2.4.1.1、\`tryReleaseShared()\` 方法 —— Semaphore.Sync
             *  2.4.1.2、\`doReleaseShared()\` 方法 —— AQS
             *   *  2.4.1.2.1、\`unparkSuccessor()\` 方法 —— AQS

## 1、Semaphore 入门 ##

### 1.1、概念 ###

**`Semaphore`**：一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做完自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。主要用于控制并发访问共享资源的线程数量。

> 底层基于 AQS 共享模式，并依赖 AQS 的变量 `state` 作为许可证 `permit`，通过控制许可证的数量，来保证线程之间的配合

### 1.2、案例 ###

> 公司大楼来了 6 辆车，怎么办？保安大哥手里只有 2 张停车卡（一个车位一张卡）

public static void main(String[] args) throws InterruptedException {
        
        // 车位
        int parkLot = 2;
        // 车数量
        int cars = 6;
        // 信号量
        Semaphore semaphore = new Semaphore(parkLot);
        for (int i = 1; i < cars + 1; i++) {
        
            int finalI = i;
            Runnable task = () -> {
        
                try {
        
                    // 1.看看有没有空车位
                    if (semaphore.availablePermits() == 0) {
        
                        System.out.println("第" + finalI + "辆司机看了看，哎，还没有空停车位，继续排队");
                    }
                    // 2.尝试进入停车位
                    semaphore.acquire();
                    System.out.println("第" + finalI + "成功进入停车场");
                    // 3.模拟车辆在停车场停留的时间
                    Thread.sleep(new Random().nextInt(10000));
                    System.out.println("第" + finalI + "驶出停车场");
                    // 4.离开停车场
                    semaphore.release();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            };
            new Thread(task).start();
        }
    }

运行结果：

第1成功进入停车场
    第5辆司机看了看，哎，还没有空停车位，继续排队
    第2成功进入停车场
    第4辆司机看了看，哎，还没有空停车位，继续排队
    第3辆司机看了看，哎，还没有空停车位，继续排队
    第6辆司机看了看，哎，还没有空停车位，继续排队
    第2驶出停车场
    第5成功进入停车场
    第1驶出停车场
    第4成功进入停车场
    第4驶出停车场
    第3成功进入停车场
    第5驶出停车场
    第6成功进入停车场
    第3驶出停车场
    第6驶出停车场

## 2、Semaphore 源码解析 ##

### 2.1、类结构 ###

public class Semaphore {
        
    	private final Sync sync;
    	// 构造方法：赋值
    	public Semaphore(int permits) {
        
            sync = new NonfairSync(permits);
        }
        public Semaphore(int permits, boolean fair) {
        
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }
        
    	// 内部类
    	abstract static class Sync extends AbstractQueuedSynchronizer {
        
    		// 构造方法：给 state 赋值
    		Sync(int permits) {
        
                setState(permits);
            }
    		// 返回 state 值
    		final int getPermits() {
        
                return getState();
            }
    	}
    	// 非公平
    	static final class NonfairSync extends Sync {
        
    		NonfairSync(int permits) {
        
                super(permits);
            }
    	}
    	// 公平
    	static final class FairSync extends Sync {
        
    		 FairSync(int permits) {
        
                super(permits);
            }
    	}
    }

从源码可知：

1.  `Semaphore` 有属性 `Sync`，且分为**公平、非公平。默认为非公平**
2.  通过构造方法给 state 赋值：在 `Semaphore` 构造方法中调用公平/非公平的构造方法，然后再调用其父类 `Sync` 的构造方法，在此构造方法中通过调用 `setState()` 方法给 state 赋值

### 2.2、`availablePermits()` 方法 —— Semaphore ###

public int availablePermits() {
        
    	// 返回 state 值
        return sync.getPermits();
    }

**`availablePermits()` 方法**：返回信号量中剩余的可用许可证数量【state 值】

### 2.3、`acquire()` 方法 —— Semaphore【可中断】 ###

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

**`acquire()` 方法**：从信号量中获取许可证。如果有许可证，则立即返回，并将可用的许可证数 -1；否则，会将当前线程进行阻塞睡眠，直到发生以下两种情况：

1.  其它线程调用了 `release()` 方法，并且将许可证分配给当前线程
2.  其它线程中断了当前线程

如果当前线程在进入 `acquire()` 方法时/之前，设置了中断状态【`thread.interrupt()`】；或者在阻塞过程中被中断，则会抛出异常 `InterruptedException`

#### 2.3.1、`acquireSharedInterruptibly()` 方法 —— AQS【共享模式；可中断】 ####

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
        
        if (Thread.interrupted()) {
        
            throw new InterruptedException();
        }
        // 判断 state 是否小于 0
        if (tryAcquireShared(arg) < 0) {
        
            // 如果 state 小于 0，获取许可证失败，则需要放入阻塞队列
            doAcquireSharedInterruptibly(arg);
        }
    }

**`acquireSharedInterruptibly()` 方法**：先判断当前线程的中断状态，如果已被中断，则抛异常 `InterruptedException`。如果未中断，则判断是否获取到许可证。如果已获取，则将可用的许可证数 -1，并直接返回；否则，调用 `doAcquireSharedInterruptibly()` 方法添加到同步队列中，阻塞起来

##### 2.3.1.1、`tryAcquireShared()` 方法 —— AQS【由子类实现】 #####

子类分为公平、非公平。

##### 2.3.1.2、`tryAcquireShared()` 方法 —— Semaphore.NonfairSync #####

protected int tryAcquireShared(int acquires) {
        
    	// 调用父类 Sync 的方法
        return nonfairTryAcquireShared(acquires);
    }

###### 2.3.1.2.1、`nonfairTryAcquireShared()` 方法 —— Semaphore.Sync【非公平】 ######

final int nonfairTryAcquireShared(int acquires) {
        
        for (;;) {
        
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 || compareAndSetState(available, remaining)) {
        
            	return remaining;
           	}
        }
    }

**`nonfairTryAcquireShared()` 方法**：由于是【共享模式】下获取，可能会存在多个线程同时执行，所以，使用了 【自旋 + CAS 】操作。要么 state < 0，要么 CAS 更新 state 成功，才会返回 state；否则，一直循环。如果返回的值 >= 0，则获取许可证成功；否则，获取失败

> 如：现有 2 个线程 A、B 同时执行这块代码。假设：state = 1，acquires = 1。

![在这里插入图片描述][282be794a9b04abb8318bc4b97719966.png]

线程 A、B 按照代码执行结果：

> available = state = 1；  
> remaining = state - 1 = 1 - 1 = 0

都要执行 if 语句。假设线程 A 先执行CAS 操作，那么线程 A 是能执行成功并返回 0，此时：`state = 0`。线程 B 也开始执行 CAS 操作，它会失败。因为 state 值被改变了【1 => 0】。所以，线程 B 会在循环一次：

> available = state = 0；  
> remaining = state - 1 = 0 - 1 = -1

所以，线程 B 会在 if 语句的第一个条件不满足返回【 -1】

###### 2.3.1.2.2、`tryAcquireShared()` 方法 —— Semaphore.FairSync【公平】 ######

![在这里插入图片描述][c8fce3a0393d4f23a776d469adc2e8b2.png]

如果有线程正在同步等待队列中排队，那么当前线程就不会去竞争许可证，而是阻塞排队；如果没有线程排队，那么，再去竞争许可证

##### 2.3.1.3、`doAcquireSharedInterruptibly()` 方法 —— AQS【共享模式；可中断】 #####

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
        
        // 【共享模式】节点入同步队列
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
        
        	// 自旋
            for (;;) {
        
                // 获取 node 的前驱
                final Node p = node.predecessor();
                if (p == head) {
        
                    // 再次获取许可证，返回的是剩下的许可证数【>= 0 获取成功】
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
        
                 		// 如果获取许可证成功，则把当前节点设置为 head 节点，并唤醒后继共享节点
                        setHeadAndPropagate(node, r);
                        p.next = null;
                        failed = false;
                        return;
                    }
                }
                // 自旋两次后，阻塞线程
                //第一次，waitStatus 默认为 0，shouldParkAfterFailedAcquire() 方法将 waitStatus 赋值为 SIGNAL并返回 false；
                //第二次 for 循环，shouldParkAfterFailedAcquire() 方法返回 true，通过调用 parkAndCheckInterrupt() 将自己阻塞
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
        
                     throw new InterruptedException();           
                }
            }
        } finally {
        
            if (failed)
                cancelAcquire(node);
        }
    }

**`doAcquireSharedInterruptibly()` 方法**：将当前线程封装为【共享模式】节点入同步队列，再自旋，如果当前节点前驱是 head 节点，再次获取许可证，如果返回结果 >= 0，则表示获取许可证成功，则调用 `setHeadAndPropagate()` 方法进行共享锁传播并返回；否则，自旋两次后，进行阻塞。

使用案例可能理解起来更清楚：

> 假设：现在有 5 个大卡车【线程 A、B、C、D、E】，2 个车位

执行 `doAcquireSharedInterruptibly()` 方法后，线程 A、B 成功获取；而线程 C、D、E 阻塞在同步队列中，等待被唤醒。【唤醒后会执行 `setHeadAndPropagate()` 方法】

![在这里插入图片描述][62338291a8634ca29ee8ea13efccd814.png]

然后，线程 A 执行完，调用 `release()` 方法进行唤醒。【此处，我们直接跳转到 `doReleaseShared()` 方法】

###### 2.3.1.3.1、`setHeadAndPropagate()` 方法 —— AQS ######

private void setHeadAndPropagate(Node node, int propagate) {
        
        // 旧 head 节点
        Node h = head;
        // 将当前节点设置为 head 节点
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
        
            Node s = node.next;
            // 只有一个节点或者存在多个节点且是共享模式，则释放所有等待的线程，各自尝试抢占锁
            if (s == null || s.isShared()) {
        
                doReleaseShared();        
            }
        }
    }

分析 if 语句：

*  `propagate > 0`：`propagate` 对于 `Semaphore` 是 `>= 0` 。如果是 `propagate > 0`，表示还有许可证，后面条件被短路；如果是 `propagate = 0`，表示没有许可证，所以，条件 `propagate > 0` 不满足，往后判断
 *  `h == null`：一般情况下是不可能是等于 null，除非旧 head 刚好被 gc 了，`h == null` 不满足
 *  `h.waitStatus < 0`：h.waitStatus 可能等于 0，可能等于 -3
    
     *  `h.waitStatus = 0`：线程 A 执行完释放了许可证，调用了 `doReleaseShared()` 方法【-1 => 0】或者 被唤醒的线程 C 调用了 `setHeadAndPropagate()` 方法【它也会去调用 `doReleaseShared()` 方法】，-1 => 0
     *  `h.waitStatus = -3`：线程 A 执行完释放了许可证，调用了 `doReleaseShared()` 方法【-1 => 0】，也会唤醒其它线程 C，线程 C 如果还没有更新 head 节点，那么就会【0 => -3】；或者线程 B 也执行完释放了许可证，那么线程 D 被唤醒了，也会【0 => -3】
 *  `(h = head) == null`：判断新 head 节点是否为空，可能为空，也可能不为空
 *  `h.waitStatus < 0`：判断新 head 的 waitStatue。可能为 0、-1、-3
    
     *  `h.waitStatus = 0`：节点入队列之前，它是尾节点，节点入队之后，还未调用 `shouldParkAfterFailedAcquire()` 方法 或者调用了 doReleaseShared() 方法
     *  `h.waitStatus = -1` 节点入队列之前，它是尾节点，节点入队之后，已调用 `shouldParkAfterFailedAcquire()` 方法

> 不管线程 D 有没有进入阻塞队列【如果没有进入，那么此时，线程 D 和 线程 C 竞争许可证】，这里假设是线程 C 获取许可证成功，没有多余的许可证（propagate = 0），那么线程 D 就会进入到同步队列【此时，有可能阻塞（线程C.waitStatus = -1），也有可能没阻塞（线程C.waitStatus = 0）】

此时，`propagate = 0`，且 head 节点状态是 `0`

> 如果 `propagate > 0`：还有剩余许可证可以获取，那么短路后面条件

如果线程 D 未阻塞：`if` 条件是不会成立的，不会唤醒线程 D；如果线程 D 阻塞：`if` 条件成立，将线程 C 设置为新 head 节点，会调用 `doReleaseShared()` 方法尝试唤醒线程 D ，但是因为 `propagate = 0`，没有多余的许可证，所以，线程 D 会再次阻塞【引起不必要的唤醒】

如果 `h.waitStatus < 0` 成立：线程 A 释放许可证时，会先将 head 节点置为 0，再调用 `unparkSuccessor()` 方法唤醒线程 C。因为 head 节点的 waitStatus 为 0 代表一种中间状态 —— head 的后继节点对应的线程已经唤醒，但它还没有做完工作。而这里旧 head 的 waitStatus < 0，只能是由于 `doReleaseShared()` 方法的 `compareAndSetWaitStatus(h, 0, Node.PROPAGATE)` 的操作。而且由于当前执行 `setHeadAndPropagate()` 的线程 C 只会在最后一句才执行 `doReleaseShared()` 方法，所以出现这种情况，一定是因为有另一个线程在调用 `doReleaseShared()` 方法才能造成，而这很可能是因为在中间状态时，又有人释放了共享锁。即：线程 B 释放了许可证，执行了 `compareAndSetWaitStatus(h, 0, Node.PROPAGATE)` 的操作，将旧 head 节点状态修改为 -3；此时，线程 C 节点调用 `setHeadAndPropagate()` 方法，修改了 head 节点。

过程如下图：

![在这里插入图片描述][126b670bd65a4ab1b69e395576364347.png]

如果 `propagate > 0` 不成立，且 `h.waitStatus < 0` 不成立，而第二个 `h.waitStatus < 0` 成立。第一个`h.waitStatus < 0` 不成立很正常，因为它一般为 0【别的线程可能不会那么碰巧读到一个中间状态】。第二个 `h.waitStatus < 0` 成立也很正常，因为只要新 head 不是队尾，那么新 head 的 waitStatus 肯定是 SIGNAL【同步等待队列入队时，会将前驱节点置为 SIGNAL，然后再阻塞自己】。同步队列中，最后一个节点的 waitStatue = 0。线程 C 设置为新 head 节点，会调用 `doReleaseShared()` 方法尝试唤醒线程 D ，但是因为 `propagate = 0`，没有多余的许可证，所以，线程 D 会再次阻塞【引起不必要的唤醒】

![在这里插入图片描述][0c019e2b78994ecbb539d75d8806d362.png]

第 2 个 if 判断：`s == null || s.isShared()`：

*  `s == null`：当 node 为队尾时，当前条件就成立，此时会调用 `doReleaseShared()` 方法，但此方法中会判断 `if (h != null && h != tail)` 条件成立，才会唤醒后驱，很显然，不会成立，所以，不会唤醒后驱。当然，由于当前节点是尾节点，所以，不需要唤醒
 *  `s.isShared()`：当 node 为队尾时，当前条件就成立，此时也会调用 `doReleaseShared()` 方法

### 2.4、`release()` 方法 —— Semaphore ###

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

#### 2.4.1、`releaseShared()` 方法 —— AQS ####

public final boolean releaseShared(int arg) {
        
        if (tryReleaseShared(arg)) {
        
        	// 如果添加许可证数成功，则唤醒同步队列中阻塞的线程
            doReleaseShared();
            return true;
        }
        return false;
    }

##### 2.4.1.1、`tryReleaseShared()` 方法 —— Semaphore.Sync #####

protected final boolean tryReleaseShared(int releases) {
        
         for (;;) {
        
             int current = getState();
             int next = current + releases;
             if (next < current) {
        
             	throw new Error("Maximum permit count exceeded");
           	 }   
             if (compareAndSetState(current, next)) {
        
             	return true;
           	}
         }
     }

`tryReleaseShared()` 方法：自旋 + CAS 操作，添加许可证数

##### 2.4.1.2、`doReleaseShared()` 方法 —— AQS #####

private void doReleaseShared() {
        
        for (;;) {
        
            Node h = head;
            if (h != null && h != tail) {
        
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
        
                	// 如果 head 节点的 waitStatue = -1，则 CAS 置为 0
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)){
        
                    	// CAS 失败，继续执行
                    	continue;
                   	}
                   	// 唤醒后驱节点【一个】
                    unparkSuccessor(h);
                } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
        
                	continue;
               	}
            }
            // 判断 head 节点是否是原 head 节点；如果未改变，则跳出循环
            if (h == head) {
        
            	break;
           	}
        }
    }

> 此时，线程 A 执行完毕，释放许可证，将 head 节点状态【-1 => 0】，调用 `unparkSuccessor()` 方法唤醒同步队列中的节点，对应线程 C。如果 head 节点没有改变，则线程 A 跳出当前循环；如果改变，则继续执行 `for(;;)`。与此同时，线程 C 在 `doAcquireSharedInterruptibly()` 方法里面开始执行。【跳转到 `setHeadAndPropagate()` 方法】

###### 2.4.1.2.1、`unparkSuccessor()` 方法 —— AQS ######

private void unparkSuccessor(Node node) {
        
        int ws = node.waitStatus;
        if (ws < 0) {
        
             compareAndSetWaitStatus(node, ws, 0);
        }
        // 以下操作为获取队列第一个非取消状态的结点，并将其唤醒
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
        
            // s 为空，或者其为取消状态，说明 s 是无效节点，此时需要执行 for 里的逻辑
            s = null;
            // 以下操作为从尾向前获取最后一个非取消状态的结点
            for (Node t = tail; t != null && t != node; t = t.prev) {
        
                if (t.waitStatus <= 0) {
        
                    s = t;                                   
                }
            }
        }
        if (s != null) {
        
        	// 唤醒同步队列中当前节点对应的线程
            LockSupport.unpark(s.thread);
        }
    }

【参考资料】  
[AQS深入理解 setHeadAndPropagate源码分析 JDK8][AQS_ setHeadAndPropagate_ JDK8]

[282be794a9b04abb8318bc4b97719966.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/53f446eb275540349ab95d2b0ac58856.png
[c8fce3a0393d4f23a776d469adc2e8b2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/87a4dd811ea54861b9d1b75d6e9a21ef.png
[62338291a8634ca29ee8ea13efccd814.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/8cb5b0eb1e0b403e9dde08048975642f.png
[126b670bd65a4ab1b69e395576364347.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/1333d5564b2a425fbce468f0364bb5c7.png
[0c019e2b78994ecbb539d75d8806d362.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/1ad4884286a64b1883d55034c37c478e.png
[AQS_ setHeadAndPropagate_ JDK8]: https://blog.csdn.net/anlian523/article/details/106319294?spm=1001.2101.3001.6650.4&utm_medium=distribute.pc_relevant.none-task-blog-2~default~BlogCommendFromBaidu~Rate-4-106319294-blog-129042579.235%5Ev43%5Epc_blog_bottom_relevance_base5&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2~default~BlogCommendFromBaidu~Rate-4-106319294-blog-129042579.235%5Ev43%5Epc_blog_bottom_relevance_base5&utm_relevant_index=9