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

　　

(相关资料图)

JUC同步锁原理源码解析四----SemaphoreSemaphore1.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写代码的思路。

关键词：