JUC并发编程之Semaphore源码详解

public class SemaphoreTest {    public static void main(String[] args) {        // 声明3个窗口        Semaphore windows = new Semaphore(3, true);        for (int i = 0; i < 10; i++) {            new Thread(() -> {                try {                    //占用窗口                    windows.acquire();                    System.out.println(Thread.currentThread().getName() + ":开始买票");                    //模拟买票流程                    Thread.sleep(3000);                    System.out.println(Thread.currentThread().getName() + ":购买成功");                } catch (InterruptedException e) {                    e.printStackTrace();                } finally {                    windows.release();                }            }).start();        }    }}

// 默认构造方法，走非公平锁流程，permits:支持令牌数量public Semaphore(int permits) {    sync = new NonfairSync(permits);}// 根据fair属性决定是否走公平还是非公平锁流程，permits:支持令牌数量public Semaphore(int permits, boolean fair) {    sync = fair ? new FairSync(permits) : new NonfairSync(permits);}

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

public final void acquireSharedInterruptibly(int arg)        throws InterruptedException {    if (Thread.interrupted())        throw new InterruptedException();    if (tryAcquireShared(arg) < 0)        doAcquireSharedInterruptibly(arg);}

protected int tryAcquireShared(int arg) {    throw new UnsupportedOperationException();}

protected int tryAcquireShared(int acquires) {    return nonfairTryAcquireShared(acquires);}

final int nonfairTryAcquireShared(int acquires) {    for (;;) { // 自旋        int available = getState();  //获得当前令牌数量        int remaining = available - acquires; // 计算令牌数量        if (remaining < 0 ||              compareAndSetState(available, remaining))  // 还有令牌可用的话,则直接进行CAS获取令牌            return remaining;    }}

public final void acquireSharedInterruptibly(int arg)        throws InterruptedException {    if (Thread.interrupted())        throw new InterruptedException();    if (tryAcquireShared(arg) < 0)        doAcquireSharedInterruptibly(arg);}

private void doAcquireSharedInterruptibly(int arg)    throws InterruptedException {    final Node node = addWaiter(Node.SHARED);  // 构建双向链表 或 入队操作    boolean failed = true;    try {        for (;;) { // 自旋            final Node p = node.predecessor();  //获取当前节点的前驱节点            if (p == head) {                int r = tryAcquireShared(arg);  // 尝试获取令牌                if (r >= 0) {  // 获取令牌成功                    setHeadAndPropagate(node, r);  //传播链表                    p.next = null; // help GC    将前驱节点的引用指向为NULL,待垃圾回收器回收                    failed = false;                    return;  // 获取令牌成功,退出自旋                }            }            if (shouldParkAfterFailedAcquire(p, node) &&                parkAndCheckInterrupt())   // 阻塞当前线程                throw new InterruptedException();        }    } finally {        // 如果某个线程被中断，非正常流程退出则将当前线程的节点设置为cancel状态        if (failed)            cancelAcquire(node);    }}

private Node addWaiter(Node mode) {    Node node = new Node(Thread.currentThread(), mode);   // 封装节点    // Try the fast path of enq; backup to full enq on failure    Node pred = tail;  // 获取末尾节点    if (pred != null) {        node.prev = pred;   // 当前节点的前驱引用指向为pred        if (compareAndSetTail(pred, node)) {  // 将当前节点设置为链表末尾节点            pred.next = node;  // 原末尾节点后驱引用指向为当前节点            return node;         }    }    enq(node);    return node;}

private Node enq(final Node node) {    for (;;) {        Node t = tail; // 获取末尾节点        if (t == null) { // Must initialize   // 构建双向链表            if (compareAndSetHead(new Node()))                tail = head;        } else {            node.prev = t;            if (compareAndSetTail(t, node)) {                t.next = node;                return t;            }        }    }}

private void doAcquireSharedInterruptibly(int arg)    throws InterruptedException {    final Node node = addWaiter(Node.SHARED);  // 构建双向链表 或 入队操作    boolean failed = true;    try {        for (;;) { // 自旋            final Node p = node.predecessor();  //获取当前节点的前驱节点            if (p == head) {                int r = tryAcquireShared(arg);  // 尝试获取令牌                if (r >= 0) {  // 获取令牌成功                    setHeadAndPropagate(node, r);  //传播链表                    p.next = null; // help GC    将前驱节点的引用指向为NULL,待垃圾回收器回收                    failed = false;                    return;  // 获取令牌成功,退出自旋                }            }            if (shouldParkAfterFailedAcquire(p, node) && //判断线程是否需要被阻塞                parkAndCheckInterrupt())   // 阻塞当前线程                throw new InterruptedException();        }    } finally {        // 如果某个线程被中断，非正常流程退出则将当前线程的节点设置为cancel状态        if (failed)            cancelAcquire(node);    }}

private void setHeadAndPropagate(Node node, int propagate) {    Node h = head; // Record old head for check below    setHead(node);    /*     * Try to signal next queued node if:     *   Propagation was indicated by caller,     *     or was recorded (as h.waitStatus either before     *     or after setHead) by a previous operation     *     (note: this uses sign-check of waitStatus because     *      PROPAGATE status may transition to SIGNAL.)     * and     *   The next node is waiting in shared mode,     *     or we don't know, because it appears null     *     * The conservatism in both of these checks may cause     * unnecessary wake-ups, but only when there are multiple     * racing acquires/releases, so most need signals now or soon     * anyway.     */    if (propagate > 0 || h == null || h.waitStatus < 0 ||   // 还有令牌可获取 || 头节点状态处于等待状态        (h = head) == null || h.waitStatus < 0) {        Node s = node.next;  // 获取当前下一节点        if (s == null || s.isShared())  // 判断下节点是否为共享节点            doReleaseShared();  // 传播~~ 具体传播什么呢？？？    }}

private void setHead(Node node) {    head = node;    node.thread = null;    node.prev = null;}

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) {  // 头节点不为null 其 头非等于尾节点 则证明当前链表还有多个节点            int ws = h.waitStatus;   // 获取head的节点状态            if (ws == Node.SIGNAL) {  // 如果当前节点状态为SIGNAL，就代表后驱节点正在被阻塞着                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))  // 通过cas将状态从等待更换为非等待,然后取反的话,将下一个节点唤醒                    continue;            // loop to recheck cases                unparkSuccessor(h);  // 唤醒线程 去获取令牌            }            else if (ws == 0 &&                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))  // 如果节点状态已经为0，则会将节点的状态更新为PROPAGATE   PROPAGATE：表示下一次共享式同步状态获取将会被无条件地传播下去                continue;                // loop on failed CAS        }        if (h == head)                   // loop if head changed            break;   // 跳出当前循环    }}

sync.releaseShared(1);

public final boolean releaseShared(int arg) {    if (tryReleaseShared(arg)) {  //通用释放令牌        doReleaseShared();  //唤醒后驱节点        return true;    }    return false;}

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

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) {  // 头节点不为null 其 头非等于尾节点 则证明当前链表还有多个节点            int ws = h.waitStatus;   // 获取head的节点状态            if (ws == Node.SIGNAL) {  // 如果当前节点状态为SIGNAL，就代表后驱节点正在被阻塞着                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))  // 通过cas将状态从等待更换为非等待,然后取反的话,将下一个节点唤醒                    continue;            // loop to recheck cases                unparkSuccessor(h);  // 唤醒线程 去获取令牌            }            else if (ws == 0 &&                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))  // 如果节点状态已经为0，则会将节点的状态更新为PROPAGATE   PROPAGATE：表示下一次共享式同步状态获取将会被无条件地传播下去                continue;                // loop on failed CAS        }        if (h == head)                   // loop if head changed            break;   // 跳出当前循环    }}

private void unparkSuccessor(Node node) {    // 先获取head节点的状态，应该是等于-1，原因在shouldParkAfterFailedAcquire方法中有体现    int ws = node.waitStatus;     // 由于-1会小于0，所以更新改为0    if (ws < 0)        compareAndSetWaitStatus(node, ws, 0);    // 获取第一个正常排队的节点    Node s = node.next;     //正常解锁流程不会走该if判断    if (s == null || s.waitStatus > 0) {        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);}

private final boolean parkAndCheckInterrupt() {    LockSupport.park(this);    return Thread.interrupted();}

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

public final void acquireSharedInterruptibly(int arg)        throws InterruptedException {    if (Thread.interrupted())        throw new InterruptedException();    if (tryAcquireShared(arg) < 0)        doAcquireSharedInterruptibly(arg);}

protected int tryAcquireShared(int arg) {    throw new UnsupportedOperationException();}

protected int tryAcquireShared(int acquires) {    for (;;) {        if (hasQueuedPredecessors())            return -1;        int available = getState();        int remaining = available - acquires;        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;    return h != t &&        ((s = h.next) == null || s.thread != Thread.currentThread());}