一文搞懂Java并发AQS的共享锁模式

@Slf4j
public class ShareLock {

    /**
     * 共享锁帮助类
     */
    private static class ShareSync extends AbstractQueuedSynchronizer {

        private int lockCount;

        /**
         * 创建共享锁帮助类，最多有count把共享锁，超过了则阻塞
         *
         * @param count 共享锁数量
         */
        public ShareSync(int count) {
           this.lockCount = count;
        }

        /**
         * 尝试获取共享锁
         *
         * @param arg 每次获取锁的数量
         * @return 返回正数，表示后续其他线程获取共享锁可能成功； 返回0，表示后续其他线程无法获取共享锁；返回负数，表示当前线程获取共享锁失败
         */
        @Override
        protected int tryAcquireShared(int arg) {
            // 自旋
            for (;;) {
                int c = getState();
                // 如果持有锁的数量大于指定数量，返回-1，线程进入阻塞
                if(c >= lockCount) {
                    return -1;
                }
                int nextc = c + 1;
                // cas设置成功，返回1，获取到共享锁
                if (compareAndSetState(c, nextc)) {
                    return 1;
                }
            }
        }

        /**
         * 尝试释放共享锁
         *
         * @param arg 释放锁的数量
         * @return 如果释放后允许唤醒后续等待结点返回true，否则返回false
         */
        @Override
        protected boolean tryReleaseShared(int arg) {
            // 自旋操作
            for (; ; ) {
                int c = getState();
                // 如果没有锁了
                if (c == 0) {
                    return false;
                }
                // 否则锁量-1
                int nextc = c - 1;
                // cas修改状态
                if (compareAndSetState(c, nextc)) {
                    return true;
                }
            }
        }
    }

    private final ShareSync sync;

    public ShareLock(int count) {
        this.sync = new ShareSync(count);
    }

    /**
     * 加共享锁
     */
    public void lockShare() {
        sync.acquireShared(1);
    }

    /**
     * 释放共享锁
     */
    public void releaseShare() {
        sync.releaseShared(1);
    }
}

public static void main(String[] args) throws InterruptedException {
        ShareLock shareLock = new ShareLock(3);
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                shareLock.lockShare();
                try {
                    log.info("lock success");
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    shareLock.releaseShare();
                    log.info("release success");
                }

            }, "thread-" + i).start();
        }
        Thread.sleep(10000);
    }

private transient volatile Node head;
private transient volatile Node tail;

static final class Node {
    //共享模式下的等待标记
    static final Node SHARED = new Node();
    //独占模式下的等待标记
    static final Node EXCLUSIVE = null;
    //表示当前结点已取消调度。当timeout或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的结点将不会再变化。
    static final int CANCELLED =  1;
    //表示后继结点在等待当前结点唤醒。后继结点入队时，会将前继结点的状态更新为SIGNAL。
    static final int SIGNAL    = -1;
    //表示结点等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
    static final int CONDITION = -2;
    //共享模式下，前继结点不仅会唤醒其后继结点，同时也可能会唤醒后继的后继结点。
    static final int PROPAGATE = -3;
    //状态，包括上面的四种状态值，初始值为0，一般是节点的初始状态
    volatile int waitStatus;
    //上一个节点的引用
    volatile Node prev;
    //下一个节点的引用
    volatile Node next;
    //保存在当前节点的线程引用
    volatile Thread thread;
    //condition队列的后续节点
    Node nextWaiter;
}

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
    //封装线程为共享Node 加入队列尾部
    final Node node = addWaiter(Node.SHARED);
    //是否成功标志
    boolean failed = true;
    try {
        //等待过程中是否被中断过的标志
        boolean interrupted = false;
        // 自旋操作
        for (;;) {
            // 获取前驱节点
            final Node p = node.predecessor();
            //如果到head的下一个，因为head是拿到资源的线程，此时node被唤醒，很可能是head用完资源来唤醒自己的
            if (p == head) {
                //尝试获取资源
                int r = tryAcquireShared(arg);
                //成功
                if (r >= 0) {
                    //将head指向自己，还有剩余资源可以再唤醒之后的线程
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    //如果等待过程中被打断过，此时将中断补上。
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }

            //判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    //head指向自己
    setHead(node);
     //如果还有剩余量，继续唤醒下一个邻居线程
    if (propagate > 0 || h == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            // 唤醒操作
            doReleaseShared();
    }
}

public final boolean releaseShared(int arg) {
    //尝试释放资源
    if (tryReleaseShared(arg)) {
        //唤醒后继结点
        doReleaseShared();
        return true;
    }
    return false;
}

private void doReleaseShared() {
    // 自旋操作
    for (;;) {
        // 获取头节点
        Node h = head;
        if (h != null && h != tail) {
            // 获取节点的等待状态
            int ws = h.waitStatus;
            // 如果节点等待状态是-1， -1表示有责任唤醒后续节点的状态
            if (ws == Node.SIGNAL) {
                // cas修改当前节点的等待状态为0
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                //唤醒后续节点
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)// head发生变化
            break;
    }
}