JUC同步锁原理源码解析四----Semaphore

Semaphore

1.Semaphore的来源

                        
                          A counting semaphore.  Conceptually, a semaphore maintains a set of permits.  Each {@link #acquire} blocks if necessary until a permit isavailable, and then takes it.  Each {@link #release} adds a permit,potentially releasing a blocking acquirer.

                        
                      

​ 一组数量的信号，只有获取到信号的线程才允许执行。通过acquire进行获取，如果获取不到则需要阻塞等待直到一个信号可用。release会释放一个信号量。通过这种方式可以实现限流.

2.Semaphore的底层实现

​ Semaphore的底层实现依旧依赖于AQS的共享锁机制.

2.AQS源码

Node节点

                        
                           static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
 
        static final int PROPAGATE = -3;

        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;
       
        volatile Thread thread;

        Node nextWaiter;
}

                        
                      

AbstractQueuedSynchronizer类

                        
                          public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
 	private transient volatile Node head;

    /**
     * Tail of the wait queue, lazily initialized.  Modified only via
     * method enq to add new wait node.
     */
    private transient volatile Node tail;

    /**
     * The synchronization state.
     */
    private volatile int state;//最重要的一个变量
       
}

                        
                      

ConditionObject类

                        
                          public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;
}

                        
                      

accquire方法

                        
                          public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试获取锁
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果获取锁失败，添加到队列中，由于ReentrantLock是独占锁所以节点必须是EXCLUSIVE类型
        selfInterrupt();//添加中断标识位
}

                        
                      

addWaiter方法

                        
                          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;
         if (compareAndSetTail(pred, node)) {//采用尾插法，充分利用时间局部性和空间局部性。尾插的节点一般不容易被取消。
             pred.next = node;
             return node;
         }
     }
     enq(node);//cas失败后执行入队操作，继续尝试
     return node;
 }


                        
                      

enq方法

                        
                          private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//获取尾指针
        if (t == null) { //代表当前队列没有节点
            if (compareAndSetHead(new Node()))//将当前节点置为头结点
                tail = head;
        } else {//当前队列有节点
            node.prev = t;//
            if (compareAndSetTail(t, node)) {//将当前节点置为尾结点
                t.next = node;
                return t;
            }
        }
    }
}

                        
                      

acquireQueued方法

                        
                          final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//找到当前节点的前驱节点
            if (p == head && tryAcquire(arg)) {//前驱节点等于头节点尝试cas抢锁。
                setHead(node);//抢锁成功将当前节点设置为头节点
                p.next = null; // help GC  当头结点置空
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//当队列中有节点在等待，判断是否应该阻塞
                parkAndCheckInterrupt())//阻塞等待，检查中断标识位
                interrupted = true;//将中断标识位置为true
        }
    } finally {
        if (failed)//
            cancelAcquire(node);//取消当前节点
    }
}


 private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

 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;//获取当前节点状态
     if (ws < 0)//如果节点为负数也即不是取消节点
         compareAndSetWaitStatus(node, ws, 0);//cas将当前节点置为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;//获取到下一个节点
     if (s == null || s.waitStatus > 0) {//下一个节点等于空或者下一个节点是取消节点
         s = null;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

                        
                      

shouldParkAfterFailedAcquire方法

                        
                          
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//获取上一个节点的等待状态
    if (ws == Node.SIGNAL)//如果状态为SIGNAL,代表后续节点有节点可以唤醒，可以安心阻塞去
        /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
        return true;
    if (ws > 0) {//如果当前状态大于0，代表节点为CANCELLED状态
        /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
        do {
            node.prev = pred = pred.prev;//从尾节点开始遍历，找到下一个状态不是CANCELLED的节点。将取消节点断链移除
        } 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.
             */
        //这里需要注意ws>0时，已经找到了一个不是取消状态的前驱节点。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//将找到的不是CANCELLED节点的前驱节点，将其等待状态置为SIGNAL
    }
    return false;
}

                        
                      

cancelAcquire方法

                        
                           private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

                        
                      

unparkSuccessor方法

                        
                           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;//获取当前节点状态
     if (ws < 0)//如果节点为负数也即不是取消节点
         compareAndSetWaitStatus(node, ws, 0);//cas将当前节点置为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;//获取到下一个节点
     if (s == null || s.waitStatus > 0) {//下一个节点等于空或者下一个节点是取消节点
         s = null;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

                        
                      

release方法

                        
                          public final boolean release(int arg) {
    if (tryRelease(arg)) {//子类实现如何释放锁
        Node h = head;//获取到头结点
        if (h != null && h.waitStatus != 0)//获取到头结点，如果头结点不为空，等待状态不为0,唤醒后续节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

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;//获取节点的等待状态
    if (ws < 0)//如果等待状态小于0,标识节点属于有效节点
        compareAndSetWaitStatus(node, ws, 0);//将当前节点的等待状态置为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;//获取到下一个节点
    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);//唤醒当前等待的有效节点S
}

                        
                      

acquireShared方法

                        
                          public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)//由子类实现
        doAcquireShared(arg);
}


                        
                      

doAcquireShared方法

                        
                          private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);//将共享节点也即读线程入队并返回
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//找到节点的前驱节点
            if (p == head) {//如果前驱节点等于头结点
                int r = tryAcquireShared(arg);//尝试获取共享锁数量
                if (r >= 0) {//如果锁的数量大于0，表示还有多余的共享锁。这里等于0也需要进一步判断。由于如果当执行到这里时，有另外的线程释放了共享锁，如果不进行判断，将会导致释放锁的线程没办法唤醒其他线程。所以这里会伪唤醒一个节点，唤醒的节点后续如果没有锁释放，依旧阻塞在当前parkAndCheckInterrupt方法中
                    setHeadAndPropagate(node, r);//将当前节点的等待状态设置为Propagate。
                    p.next = null; // help GC
                    if (interrupted)//判断是会否中断过
                        selfInterrupt();//设置中断标识位
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//判断是否应该阻塞等待
                parkAndCheckInterrupt方法中())//阻塞并检查中断标识
                interrupted = true;//重置中断标识位
        }
    } finally {
        if (failed)//如果失败
            cancelAcquire(node);//取消节点
    }
}

                        
                      

setHeadAndPropagate方法

                        
                          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 //可获取的共享锁也即读锁的数量，对于ReentrantReadWriteLock而言，永远都是1，所以会继续唤醒下一个读线程
            || h == null //如果旧的头结点为空
            || h.waitStatus < 0 ||//头结点的等待状态不为0
            (h = head) == null || h.waitStatus < 0) {//旧头节点不为空并且等待状态小于0也即是有效节点
            Node s = node.next;//获取到node的下一个节点
            if (s == null || s.isShared())//如果node的下一个节点为空或者是共享节点
                doReleaseShared();//唤醒下一个线程
        }
    }

                        
                      

releaseShared方法

                        
                          public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//子类实现释放锁
        doReleaseShared();//唤醒后续线程
        return true;//释放成功
    }
    return false;//释放是吧
}

                        
                      

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;//获取当前节点的等待状态
            if (ws == Node.SIGNAL) {//如果状态为SIGNAL
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//cas将SIGNAL状态置为0。SIGNAL标识后续有线程需要唤醒
                    continue;            // loop to recheck cases
                unparkSuccessor(h);//唤醒后续线程
            }
            else if (ws == 0 &&//如果当前状态为0。表示有线程将其置为0
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))//cas将0状态置为PROPAGATE。在多个共享锁同时释放时，方便继续进行读传播，唤醒后续节点
                continue;                // loop on failed CAS
        }
        if (h == head)//如果头结点没有改变，证明没有必要继续循环等待了，直接退出吧，如果头结点放生变化，可能有其他线程释放了锁。
            break;
    }
}

                        
                      

await()

                        
                          public final void await() throws InterruptedException {
    if (Thread.interrupted())//线程是否发生中断，是，就抛出中断异常
        throw new InterruptedException();
    Node node = addConditionWaiter();//加入条件等待队列
    int savedState = fullyRelease(node);//释放锁，并返回。因为当前线程需要等待
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {//判断是否在竞争队列中。AQS分为两个队列一个是竞争队列，等待调度执行，一个是等待队列等待在ConditionObject上。
        LockSupport.park(this);//阻塞等待
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//重新去获取锁并判断当前中断模式不是THROW_IE
        interruptMode = REINTERRUPT;//将中断模式置为REINTERRUPT
    if (node.nextWaiter != null) // clean up if cancelled如果当前节点的下一个节点不为空
        unlinkCancelledWaiters();//清除等待队列中已经取消的节点
    if (interruptMode != 0)//如果当前中断模式不等于0
        reportInterruptAfterWait(interruptMode);
}

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)//如果是THROW_IE直接抛出异常
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)//如果是REINTERRUPT
        selfInterrupt();//重置中断标识位
}

                        
                      

addConditionWaiter方法

                        
                          private Node addConditionWaiter() {
    Node t = lastWaiter;//获取到最后一个节点
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {//最后一个节点不等于空，并且等待状态不等于CONDITION
        unlinkCancelledWaiters();//将取消节点断链,标准的链表操作
        t = lastWaiter;//获取到最后一个有效的节点
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);//将当前节点封装成node
    if (t == null)//如果最后一个节点为空，表示当前节点是第一个入队的节点
        firstWaiter = node;
    else
        t.nextWaiter = node;//否则将当前node挂在链表末尾
    lastWaiter = node;//设置最后节点的指针指向当前node
    return node;
}

                        
                      

fullyRelease方法

                        
                          final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();//获取当前state状态
        if (release(savedState)) {//释放锁尝试
            failed = false;
            return savedState;//返回
        } else {
            throw new IllegalMonitorStateException();//抛出释放锁异常
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;//如果失败将节点置为取消状态
    }
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {//尝试释放锁，在CyclciBarrier中由于线程需要去阻塞，所以需要将锁释放，后续重新拿锁
        Node h = head;
        if (h != null && h.waitStatus != 0)//从头结点开始唤醒
            unparkSuccessor(h);
        return true;
    }
    return false;
}

                        
                      

isOnSyncQueue方法

                        
                          final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)//如果当前节点是Condition或者node.pre节点为空，标识不在竞争队列中，返回faslse
        return false;
    if (node.next != null) // If has successor, it must be on queue  表示在竞争队列中
        return true;
    /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
    return findNodeFromTail(node);//从竞争队列的尾结点开始找当前node，找到就返回true，否则为false
}

private boolean findNodeFromTail(Node node) {
    Node t = tail;//获取到尾结点
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        t = t.prev;
    }
}

                        
                      

findNodeFromTail方法

                        
                          private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?//判断当前是否中断过
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) ://如果移动到竞争队列中并入队成功，返回THROW_IE，否则返回REINTERRUPT
    0;//没有中断过直接返回0
}

//走到这里表示条件队列的条件满足，可以将节点移动到竞争队列中执行
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {//尝试将当前为Condition的节点置为0，并移动到竞争队列中
        enq(node);
        return true;
    }
    /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
    while (!isOnSyncQueue(node))//如果不在竞争队列中返回false
        Thread.yield();
    return false;
}

                        
                      

signalAll方法 。

                        
                          public final void signalAll() {
    if (!isHeldExclusively())//是不是持有独占锁
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;//获取等待队列的第一个节点
    if (first != null)//如果节点不为空
        doSignalAll(first);//唤醒所有线程
}

//从头指针一直遍历等待队列，将其移动到竞争队列中
private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);//
        first = next;
    } while (first != null);
}

                        
                      

transferForSignal方法

                        
                          final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//cas自旋将其等待状态改为0
        return false;

    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    Node p = enq(node);//将其放入竞争队列
    int ws = p.waitStatus;//获取节点的等待状态
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//如果节点是取消状态或者cas将其置为signal失败，唤醒当前线程，让他自己处理，后续在竞争队列中会自动移除取消节点
        LockSupport.unpark(node.thread);
    return true;
}

                        
                      

总结：AQS提供了统一的模板，对于如何入队出队以及线程的唤醒都由AQS提供默认的实现，只需要子类实现自己上锁和解锁的逻辑.

3.Semaphore

基本使用

                        
                          import java.util.concurrent.Semaphore;

public class SemaphoreDemo {
    public static void main(String[] args) {
        //Semaphore s = new Semaphore(2);
        Semaphore s = new Semaphore(2, true);
        //允许一个线程同时执行
        //Semaphore s = new Semaphore(1);
        new Thread(() -> {
            try {
                s.acquire();
                System.out.println("T1 running...");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                s.release();
            }
        }).start();

        new Thread(() -> {
            try {
                s.acquire();
                System.out.println("T2 running...");
                s.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                s.release();
            }
        }).start();
    }
}

                        
                      

Sync类

                        
                          abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 1192457210091910933L;

    Sync(int permits) {
        setState(permits);//设置信号量
    }

    final int getPermits() {
        return getState();//获得信号量
    }

    final int nonfairTryAcquireShared(int acquires) {//非公平锁的抢锁方式
        for (;;) {
            int available = getState();//获取state中的可用信号量
            int remaining = available - acquires;//减1
            if (remaining < 0 ||//信号量小于0，直接返回
                compareAndSetState(available, remaining))//尝试cas抢锁
                return remaining;//返回剩余的信号量
        }
    }

    protected final boolean tryReleaseShared(int releases) {
        for (;;) {
            int current = getState();//获取当前state
            int next = current + releases;//将state+1.也即信号量加1
            if (next < current) // overflow 非法条件判断，超过最大数量
                throw new Error("Maximum permit count exceeded");
            if (compareAndSetState(current, next))//cas尝试释放锁
                return true;//释放成功返回
        }
    }
	
    //减少信号量
    final void reducePermits(int reductions) {
        for (;;) {
            int current = getState();//获取当前state
            int next = current - reductions;
            if (next > current) // underflow
                throw new Error("Permit count underflow");
            if (compareAndSetState(current, next))//cas尝试减少信号量
                return;
        }
    }
	
    //清空信号数量
    final int drainPermits() {
        for (;;) {
            int current = getState();//获取当前state状态
            if (current == 0 || compareAndSetState(current, 0))//当前信号为0 或者将state置为0也即将信号数量置为0
                return current;
        }
    }
}

                        
                      

FairSync与NonfairSync的类实现

                        
                          //公平锁
static final class FairSync extends Sync {
    private static final long serialVersionUID = 2014338818796000944L;

    FairSync(int permits) {
        super(permits);
    }

    protected int tryAcquireShared(int acquires) {
        for (;;) {
            if (hasQueuedPredecessors())//队列中是否有线程在排队
                return -1;//获取失败
            int available = getState();//可用的信号量
            int remaining = available - acquires;//减去当前获取的数量
            if (remaining < 0 ||//可用的信号量小于0
                compareAndSetState(available, remaining))//cas设置state变量.
                return remaining;//返回可用的信号量
        }
    }
}

//非公平锁
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = -2694183684443567898L;

    NonfairSync(int permits) {
        super(permits);
    }

    protected int tryAcquireShared(int acquires) {
        return nonfairTryAcquireShared(acquires);//详情请看父类的实现
    }
}

                        
                      

acquire方法

                        
                          public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);//请查看父类实现，与acquireShared一致，不过加了一场处理
}

                        
                      

release方法:

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

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//Semaphore的类实现锁获取的方法。
        doReleaseShared();//与AQS中一致，不过多赘述
        return true;
    }
    return false;
}

                        
                      

4.留言

​ 到了这里，其实AQS的源码基本已经覆盖了，对于AQS的源码也应该有了清楚的认知。总结就是：一个volatile 的state变量，两个等待队列(竞争队列，条件队列)，通过cas的方式保证单变量的原子性。后续将会对Exchanger以及Phaser进行源码解析，到此基本AQS已经到了一个段落了。后续观看源码时，请注意多考虑一下多线程并发时可能出现的情况，去理解doug lea写代码的思路.

最后此篇关于JUC同步锁原理源码解析四----Semaphore的文章就讲到这里了,如果你想了解更多关于JUC同步锁原理源码解析四----Semaphore的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。