AQS源码解析(2)——共享模式

/**
 * @Author: zhouguanya20@163.com
 * @Date: 2019-09-08 10:14
 * @Description: 信号量——共享锁实现
 *
 * Semaphore还用于Hystrix限流框架中，控制系统并发在可控的范围内，保证系统高可用
 */
public class SemaphoreDemo {
    /**
     * 限定线程数量
     */
    private static final Semaphore SEMAPHORE = new Semaphore(3);
    /**
     * 每次获取的许可数
     */
    private static final int PERMITS = 1;

    /**
     * 线程
     */
    static class TestThread implements Runnable {
        @Override
        public void run() {
            try {
                // 获取许可
                SEMAPHORE.acquire(PERMITS);
                System.out.println(printCurrent() +
                        " : " + Thread.currentThread().getName() + " 进来了");
                // 持有许可，不会释放许可
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // ！！！！非常重要！！！！
                SEMAPHORE.release(PERMITS);
            }
        }

        /**
         * 打印时间
         *
         * @return  当前时间字符串
         */
        private String printCurrent() {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
            return sdf.format(new Date());
        }
    }

    public static void main(String[] args) {
        // 注意：本例只做演示使用。不要显示创建线程池，详情参考：Alibaba Java Coding Guidelines
        // 不要显式创建线程，请使用线程池。
        // 线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。
        // 说明使用线程池的好处是减少在创建和销毁线程上所花的时间以及系统资源的开销，解决资源不足的问题。
        // 如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者“过度切换”的问题。
        Thread t1 = new Thread(new TestThread(), "TestThread1");
        Thread t2 = new Thread(new TestThread(), "TestThread2");
        Thread t3 = new Thread(new TestThread(), "TestThread3");
        Thread t4 = new Thread(new TestThread(), "TestThread4");
        t1.start();
        t2.start();
        t3.start();
        t4.start();
    }
}

2019-09-22 10:35:30 : TestThread1 进来了
2019-09-22 10:35:30 : TestThread3 进来了
2019-09-22 10:35:30 : TestThread2 进来了
2019-09-22 10:35:33 : TestThread4 进来了

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(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;
    }
}

/**
 * Acquires in shared uninterruptible mode.
 * @param arg the acquire argument
 */
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) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

/**
 * Sets head of queue, and checks if successor may be waiting
 * in shared mode, if so propagating if either propagate > 0 or
 * PROPAGATE status was set.
 *
 * @param node the node
 * @param propagate the return value from a tryAcquireShared
 */
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();
    }
}

/**
 * Releases in shared mode.  Implemented by unblocking one or more
 * threads if {@link #tryReleaseShared} returns true.
 *
 * @param arg the release argument.  This value is conveyed to
 *        {@link #tryReleaseShared} but is otherwise uninterpreted
 *        and can represent anything you like.
 * @return the value returned from {@link #tryReleaseShared}
 */
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

/**
 * Attempts to set the state to reflect a release in shared mode.
 *
 * <p>This method is always invoked by the thread performing release.
 *
 * <p>The default implementation throws
 * {@link UnsupportedOperationException}.
 *
 * @param arg the release argument. This value is always the one
 *        passed to a release method, or the current state value upon
 *        entry to a condition wait.  The value is otherwise
 *        uninterpreted and can represent anything you like.
 * @return {@code true} if this release of shared mode may permit a
 *         waiting acquire (shared or exclusive) to succeed; and
 *         {@code false} otherwise
 * @throws IllegalMonitorStateException if releasing would place this
 *         synchronizer in an illegal state. This exception must be
 *         thrown in a consistent fashion for synchronization to work
 *         correctly.
 * @throws UnsupportedOperationException if shared mode is not supported
 */
protected boolean tryReleaseShared(int arg) {
    throw new UnsupportedOperationException();
}

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;
    }
}