Semaphore是什么？

• Semaphore，俗称信号量，它是操作系统中PV操作的原语在java的实现。
• 基于AbstractQueuedSynchronizer实现！
• Semaphore的功能非常强大，大小为1的信号量就类似于互斥锁，通过同时只能有一个线程获取信号量实现。
• 大小为n（n>0）的信号量可以实现限流的功能，它可以实现只能有n个线程同时获取信号量。

PV操作是什么？

• PV操作是操作系统一种实现进程互斥与同步的有效方法。
• PV操作与信号量（S）的处理相关，P表示通过的意思，V表示释放的意思。
• 用PV操作来管理共享资源时，首先要确保PV操作自身执行的正确性。

P操作的主要动作：

• S减1；
• 若S减1后仍大于或等于0，则进程继续执行；
• 若S减1后小于0，则该进程被阻塞后放入等待该信号量的等待队列中，然后转进程调度。

V操作的主要动作：

• S加1；
• 若相加后结果大于0，则进程继续执行；
• 若相加后结果小于或等于0，则从该信号的等待队列中释放一个等待进程，然后再返回原进程继续执行或转进程调度。

Semaphore的使用场景

• 可以用于做流量控制，特别是公用资源有限的应用场景！

Semaphore的使用方式

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class SemaphoneTest1 {/*** 实现一个同时只能处理3个请求的限流器*/private static Semaphore semaphore = new Semaphore(3);/*** 定义一个线程池*/private static ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 50, 60, TimeUnit.SECONDS,new LinkedBlockingDeque<>(200));/*** 模拟执行方法*/public static void exec() {try {semaphore.acquire(1);// 模拟真实方法执行System.out.println("执行exec方法" + System.currentTimeMillis());Thread.sleep(2000);} catch (Exception e) {e.printStackTrace();} finally {semaphore.release(1);}}public static void main(String[] args) throws InterruptedException {{for (;;) {Thread.sleep(100);// 模拟请求以10个/s的速度executor.execute(() -> exec());}}}
}

• 执行结果

Semaphore的构造方法

/*** 默认使用非公平的方式*/
public Semaphore(int permits) {sync = new NonfairSync(permits);
}/*** permits 表示许可证的数量（资源数）* fair 表示公平性，如果这个设为 true 的话，下次执行的线程会是等待最久的线程*/
public Semaphore(int permits, boolean fair) {sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

Semaphore构造方法源码分析

/*** 非公平锁：直接调用了父类（AQS）的实现*/
NonfairSync(int permits) {super(permits);
}/*** 公平锁：直接调用了父类（AQS）的实现*/
FairSync(int permits) {super(permits);
}/*** 调用setState(permits);方法去设置资源数*/
Sync(int permits) {setState(permits);
}/*** 给AQS的state赋值 */
protected final void setState(int newState) {state = newState;
}

Semaphore的获取锁方法：acquire(int permits);源码分析

/*** 获取锁的源码*/
public void acquire(int permits) throws InterruptedException {// 当传入的许可证数量小于0，抛异常if (permits < 0) throw new IllegalArgumentException();// 调用AQS中的获取可中断的共享资源方法sync.acquireSharedInterruptibly(permits);
}/*** 获取可中断的共享资源*/
public final void acquireSharedInterruptibly(int arg)throws InterruptedException {// 如果线程被中断，抛出异常if (Thread.interrupted())throw new InterruptedException();// 尝试获取锁，小于0说明没有锁了if (tryAcquireShared(arg) < 0)// 尝试获取共享锁或者中断doAcquireSharedInterruptibly(arg);
}

Semaphore的尝试获取锁逻辑：tryAcquireShared(arg)

/*** 公平锁的tryAcquireShared实现方式。* 相对于非公平锁多了一个校验：hasQueuedPredecessors()*/
protected int tryAcquireShared(int acquires) {for (;;) {// 如果有链表中有排队的对象！返回-1if (hasQueuedPredecessors())return -1;// 获取当前的资源数int available = getState();// 当前的资源数减去需要消耗的资源数，计算深入的资源数int remaining = available - acquires;// 只有剩余的资源数小于0（没有竞争了）或者能通过CAS将当前的资源数变为减少后的值（这些线程可以去获取锁或者等待锁），才会返回结果，否则一直循环。if (remaining < 0 ||compareAndSetState(available, remaining))// 返回剩余的资源数return remaining;}
}/*** 非公平锁的tryAcquireShared实现方式：调用nonfairTryAcquireShared(acquires)*/
protected int tryAcquireShared(int acquires) {return nonfairTryAcquireShared(acquires);
}/*** 非公平锁的尝试获取共享资源*/
final int nonfairTryAcquireShared(int acquires) {for (;;) {// 获取当前的资源数int available = getState();// 当前的资源数减去需要消耗的资源数，计算深入的资源数int remaining = available - acquires;// 如果剩余的资源数小于0（没有竞争了）或者能通过CAS将当前的资源数变为减少后的值（这些线程可以去获取锁或者等待锁）if (remaining < 0 ||compareAndSetState(available, remaining))// 返回剩余的资源数return remaining;}
}/*** 判断队列中是否有排队的*/
public final boolean hasQueuedPredecessors() {// The correctness of this depends on head being initialized// before tail and on head.next being accurate if the current// thread is first in queue.Node t = tail; // Read fields in reverse initialization orderNode h = head;Node s;// 未初始化的时候，队列头部尾部的值为null，相等----不满足// 只有一个的时候，头尾相等----不满足// 头部的下一个为null，说明只有一个----满足// 下一个的线程是当前线程，重入了----满足// 简单说：链表中只有一个或者链表是空的，返回false。链表中有多个，不满足重入的机制，返回false。只有链表中有多个数据并且持有线程是当前线程的时候才会返回true！return h != t &&((s = h.next) == null || s.thread != Thread.currentThread());
}

尝试获取共享锁或者中断：doAcquireSharedInterruptibly

/*** 尝试获取共享锁或者中断*/
private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {// 添加到等待队列final Node node = addWaiter(Node.SHARED);// 定义失败标记为trueboolean failed = true;try {for (;;) {// 得到当前节点的前驱节点final Node p = node.predecessor();// 如果前置节点是头结点if (p == head) {// 尝试去获取锁：公平锁与非公平锁实现不一致！int r = tryAcquireShared(arg);// 获取的资源数大于0（获取到了资源，Semaphore）if (r >= 0) {// 设置头结点和链表，准备唤醒后继节点setHeadAndPropagate(node, r);// 取消节点的引用，方便GC去回收p.next = null; // help GC// 将失败标记改为falsefailed = false;// 跳出方法！return;}}// 代码执行到这里，说明尝试获取锁，但是获取锁失败了。// 阻塞前的准备工作操作成功（状态是-1的时候成功）// 将线程阻塞，等待他去唤醒。唤醒后返回线程的中断状态！if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {// 上面代码抛出异常的时候，会执行这里的逻辑if (failed)// 取消获取锁的逻辑cancelAcquire(node);}
}/*** 添加线程到同步队列*/
private Node addWaiter(Node mode) {// 创建一个当前线程的Node节点Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failure：尝试enq的快速路径；故障时备份到完整enq// 得到之前的尾节点Node pred = tail;// 尾节点不为空，说明队列存在if (pred != null) {// 设置当前线程节点的上一个节点是之前的尾节点node.prev = pred;// cas尝试将尾节点设置为当前线程的节点if (compareAndSetTail(pred, node)) {// 设置之前尾节点，现在的倒数第二个节点的下一个节点是当前线程节点pred.next = node;// 返回当前节点return node;}}// CAS失败或者节点没有创建,会执行这入队的操作。详细请看下面的代码enq(node);// 入队成功后，返回当前节点return node;
}/*** 设计精髓：100%创建队列或者100%入队*/
private Node enq(final Node node) {for (;;) {// 得到之前的尾节点Node t = tail;// 之前的尾节点为空，需要进行初始化队列if (t == null) { // Must initialize// 通过CAS的方式将头节点设置为当前节点if (compareAndSetHead(new Node()))// 头结点设置成功后，复制给尾节点。只有一个节点的状态tail = head;} else {// 设置当前线程节点的上一个节点是之前的尾节点node.prev = t;// cas尝试将尾节点设置为当前线程的节点if (compareAndSetTail(t, node)) {// 设置之前尾节点，现在的倒数第二个节点的下一个节点是当前线程节点t.next = node;// 返回当前节点！注意：这里是这个方法唯一返回的地方！也就是说初始化后还会继续循环一次来设置上一个下一个节点，然后进行返回。return t;}}}
}/*** 设置头结点和链表*/
private void setHeadAndPropagate(Node node, int propagate) {// 得到旧的头Node h = head; // Record old head for check below// 将当前节点设置为头节点，所属线程设置为null，前驱节点设置为nullsetHead(node);// propagate > 0：说明还有剩余共享锁可以获取，那么短路后面条件。// h == null与(h = head) == null是一个防止控制住的标准写法，因为经过了addWaiter方法，肯定会有一个节点！// h.waitStatus < 0：状态小于0。在调用doReleaseShared的时候回cas状态到0！说明有其他线程调用了doReleaseShared()。第一个是旧的头结点，第二个是新的头结点（当前节点）！// (h = head) == null：除了防止空指针，这里会将当前节点复制到h变量上if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {// 得到当前节点的下一个节点Node s = node.next;// 下一个节点为空（node是队尾）或者下一个节点是共享模式if (s == null || s.isShared())// 唤醒后继的节点并且保证继续传播doReleaseShared();}
}/*** 唤醒后继的节点并且保证继续传播*/
private void doReleaseShared() {/** Ensure that a release propagates, even if there are other* in-progress acquires/releases.  This proceeds in the usual* way of trying to unparkSuccessor of head if it needs* signal. But if it does not, status is set to PROPAGATE to* ensure that upon release, propagation continues.* Additionally, we must loop in case a new node is added* while we are doing this. Also, unlike other uses of* unparkSuccessor, we need to know if CAS to reset status* fails, if so rechecking.*/for (;;) {// 得到当前的头结点Node h = head;// 当头结点不为空并且头结点不是尾节点的时候if (h != null && h != tail) {// 获取头结点的状态int ws = h.waitStatus;// 头结点状态为-1：当前节点的后继节点包含的线程需要运行if (ws == Node.SIGNAL) {// cas将当前头结点状态设置为0（当前节点在sync队列中，等待着获取锁）失败。进入下次循环！CAS成功继续执行！if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue;            // loop to recheck cases// 执行唤醒锁的流程unparkSuccessor(h);}// 头结点的状态为0并且不能变为cas状态到-3（后续的acquireShared能够得以执行），继续执行！否则进入下次循环else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}// 如果头部改变，则继续循环。没有改变就跳出循环！if (h == head)                   // loop if head changedbreak;}
}/*** 唤醒锁的流程*/
private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling.  It is OK if this* fails or if status is changed by waiting thread.*/// 获取当前节点的状态int ws = node.waitStatus;// 如果状态小于0，尝试将状态变为0if (ws < 0)compareAndSetWaitStatus(node, ws, 0);/** Thread to unpark is held in successor, which is normally* just the next node.  But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*/// 获取当前节点（头节点）的下一个节点Node s = node.next;// 头节点的下一个节点是空的或者下一个节点的状态是1（当前的线程被取消）if (s == null || s.waitStatus > 0) {// 设置下个节点为nulls = null;// 从尾部开始向前遍历，找到最前的一个处于正常阻塞状态的结点，直到节点重合// 从尾部遍历的原因是为了防止在高并发场景下漏掉线程for (Node t = tail; t != null && t != node; t = t.prev)if (t.waitStatus <= 0)s = t;}// 过滤后的下一个节点不为null，唤醒他if (s != null)// 唤醒下一个节点LockSupport.unpark(s.thread);
}/*** 获取锁失败后的准备逻辑，阻塞前的准备逻辑*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// 获取前驱当前节点的等待状态int ws = pred.waitStatus;// 状态为-1的时候：当前节点的后继节点包含的线程需要运行，也就是unpark；if (ws == Node.SIGNAL)/** This node has already set status asking a release* to signal it, so it can safely park.*/// 前驱当前节点状态已经设置为SIGNAL，可以进行安全的阻塞return true;// 大于0（CANCELLED状态）：表示当前的线程被取消；if (ws > 0) {/** Predecessor was cancelled. Skip over predecessors and* indicate retry.*/// 前驱节点已经因为超时或响应了中断，需要跳过这些状态大于0的节点，直到找到一个状态不是大于0的。do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);// 跳过中断的线程后，设置前驱节点的下一个节点为当前节点。pred.next = node;} else {/** waitStatus must be 0 or PROPAGATE.  Indicate that we* need a signal, but don't park yet.  Caller will need to* retry to make sure it cannot acquire before parking.*/// 针对于ReentrantLock，到这里的状态只能为0或者PROPAGATE（-3）// 通过CAS将前置节点的状态设置为SIGNAL（-1）compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;
}/*** 状态为SIGNAL（-1）成功。他需要排队，所以直接调用park方法进行阻塞*/
private final boolean parkAndCheckInterrupt() {// 阻塞LockSupport.park(this);// unpark之后，返回当前的中断状态，并清除中断标志位return Thread.interrupted();
}/*** 取消获取锁的逻辑*/
private void cancelAcquire(Node node) {// Ignore if node doesn't exist// 忽略节点不存在的时候if (node == null)return;// 设置当前节点的线程为nullnode.thread = null;// Skip cancelled predecessors：有前驱节点被取消，跳过所有被取消的// 得到前驱节点Node pred = node.prev;// 前驱节点的状态大于0，被取消while (pred.waitStatus > 0)// 将前驱结点的前驱结点设置为当前节点的前驱结点。简单理解就是将当前节点的前驱节点设置为第一个找到的正常状态（<=0）的前驱节点node.prev = pred = pred.prev;// predNext is the apparent node to unsplice. CASes below will// fail if not, in which case, we lost race vs another cancel// or signal, so no further action is necessary.// 获取当前节点的下一个节点Node predNext = pred.next;// Can use unconditional write instead of CAS here.// After this atomic step, other Nodes can skip past us.// Before, we are free of interference from other threads.// 将当前节点状态设置为1（取消状态）。这里不用CAS的原因是这个执行完其他线程会跳过取消状态，这个执行前无其他线程在执行！node.waitStatus = Node.CANCELLED;// If we are the tail, remove ourselves.// 如果当前节点是尾节点，将尾节点设置为上一个节点。简单理解就是移除当前节点。if (node == tail && compareAndSetTail(node, pred)) {compareAndSetNext(pred, predNext, null);} else {// 进入else说明node不是队尾（或者是队尾但是cas队尾失败（其实结果也不是队尾，因为被别的线程抢先了））// If successor needs signal, try to set pred's next-link// so it will get one. Otherwise wake it up to propagate.// 定义一个状态标识int ws;// 筛选后的前驱节点不是头结点// 并且当前节点状态为-1（等待唤醒）或者（当前节点不在运行或者不被取消（<= 0）并且可以将当前节点CAS到-1状态（等待唤醒））// 并且前驱节点有线程持有！if (pred != head &&((ws = pred.waitStatus) == Node.SIGNAL ||(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&pred.thread != null) {// 得到当前节点的下一个节点Node next = node.next;// 下一个节点不为空，并且下一个节点没有被取消if (next != null && next.waitStatus <= 0)// CAS将前一个节点与下一个节点连接。简单理解就是跳过（取消）当前节点！compareAndSetNext(pred, predNext, next);} else {// 唤醒下一个不被取消的节点！unparkSuccessor(node);}// 当前节点的下一个节点取消指向node.next = node; // help GC}
}

Semaphore的释放锁方法：release(int permits);源码分析

/*** 释放锁的逻辑*/
public void release(int permits) {// 传入的数量小于0，直接抛出异常if (permits < 0) throw new IllegalArgumentException();// 调用释放共享说的方法sync.releaseShared(permits);
}/*** 释放共享锁的逻辑*/
public final boolean releaseShared(int arg) {// 尝试去释放共享锁if (tryReleaseShared(arg)) {// 唤醒后继的节点并且保证继续传播doReleaseShared();// 整体返回true，表示释放共享锁成功return true;}// java规范的写法：必须有个返回值，不会执行到这里！return false;
}/*** 尝试去释放共享锁*/
protected final boolean tryReleaseShared(int releases) {for (;;) {// 获取状态：锁的数量，构造方法传入的数量int current = getState();// 得到如果释放够的总数量int next = current + releases;// 大于int的最大值，next变为负数！溢出了，抛异常！if (next < current) // overflowthrow new Error("Maximum permit count exceeded");// CAS将持有的状态（剩余资源数）设置为新的值if (compareAndSetState(current, next))return true;}
}

结束语

• 获取更多有价值的文章，让我们一起成为架构师！
• 关注公众号，可以让你对MySQL有非常深入的了解
• 关注公众号，每天持续高效的了解并发编程！
• 关注公众号，后续持续高效的了解spring源码！
• 这个公众号，无广告！！！每日更新！！！