java Semaphore信号量从入门到源码精通
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的使用结果.png
看Semaphore源码建议有AQS的基础
「并发编程」并发包中工具类的基础：AQS
「并发编程」通俗易懂的来学ReentrantLock锁源码
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();
// 尝试获取锁，这里不满足直接抛出异常
if (tryAcquireShared(arg) < 0)
// 尝试获取共享锁或者中断
doAcquireSharedInterruptibly(arg);
}
Semaphore的尝试获取锁逻辑：tryAcquireShared(arg)
/**
* 公平锁的tryAcquireShared实现方式。
* 相对于非公平锁多了一个校验：hasQueuedPredecessors()
*/
protected int tryAcquireShared(int acquires) {
for (;;) {
// 如果有链表中有排队的对象！返回-1
if (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 order
Node 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);
// 定义失败标记为true
boolean 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
// 将失败标记改为false
failed = 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，前驱节点设置为null
setHead(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 changed
break;
}
}
/**
* 唤醒锁的流程
*/
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，尝试将状态变为0
if (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) {
// 设置下个节点为null
s = 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;
// 设置当前节点的线程为null
node.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) // overflow
throw new Error("Maximum permit count exceeded");
// CAS将持有的状态（剩余资源数）设置为新的值
if (compareAndSetState(current, next))
return true;
}
}
原文作者：绽放苍穹的程序员
扫码领红包
微信赞赏
支付宝扫码领红包
声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。侵权投诉：375170667@qq.com
