【Java并发】AQS抽象队列同步器深入解析

public abstract class AbstractQueuedSynchronizer {
    // 同步状态，不同同步器有不同含义
    private volatile int state;

    protected final int getState() {
        return state;
    }

    protected final void setState(int newState) {
        state = newState;
    }

    // CAS修改state
    protected final boolean compareAndSetState(int expect, int update) {
        return STATE.compareAndSet(this, expect, update);
    }
}

static final class Node {
    // 节点状态
    volatile int waitStatus;

    // 前驱节点
    volatile Node prev;

    // 后继节点
    volatile Node next;

    // 节点关联的线程
    volatile Thread thread;

    // 条件队列中的下一个节点（Condition使用）
    Node nextWaiter;

    // 共享模式标记
    static final Node SHARED = new Node();

    // 独占模式标记
    static final Node EXCLUSIVE = null;

    // waitStatus常量
    static final int CANCELLED =  1;  // 线程已取消
    static final int SIGNAL    = -1;  // 后继节点需要被唤醒
    static final int CONDITION = -2;  // 在条件队列中等待
    static final int PROPAGATE = -3;  // 共享模式下传播唤醒
}

// 独占获取
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

// 独占释放
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

// 共享获取
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

// 共享释放
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

public final void acquire(int arg) {
    // 1. tryAcquire尝试获取锁（子类实现）
    // 2. 失败则addWaiter加入队列
    // 3. acquireQueued自旋获取锁
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

private Node addWaiter(Node mode) {
    Node node = new Node(mode);

    for (;;) {
        Node oldTail = tail;
        if (oldTail != null) {
            // 设置前驱
            node.setPrevRelaxed(oldTail);
            // CAS设置尾节点
            if (compareAndSetTail(oldTail, node)) {
                oldTail.next = node;
                return node;
            }
        } else {
            // 队列为空，初始化
            initializeSyncQueue();
        }
    }
}

final boolean acquireQueued(final Node node, int arg) {
    boolean interrupted = false;
    try {
        for (;;) {
            // 获取前驱节点
            final Node p = node.predecessor();

            // 如果前驱是head，尝试获取锁
            if (p == head && tryAcquire(arg)) {
                setHead(node);  // 成功，设为新head
                p.next = null;  // help GC
                return interrupted;
            }

            // 检查是否需要阻塞
            if (shouldParkAfterFailedAcquire(p, node))
                interrupted |= parkAndCheckInterrupt();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;

    if (ws == Node.SIGNAL)
        // 前驱状态是SIGNAL，可以安全阻塞
        return true;

    if (ws > 0) {
        // 前驱已取消，跳过取消的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 将前驱状态设为SIGNAL
        pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
    }
    return false;
}

public final boolean release(int arg) {
    // tryRelease由子类实现
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);  // 唤醒后继
        return true;
    }
    return false;
}

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);

    // 找到需要唤醒的后继节点
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 从尾部向前找有效节点
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null)
        LockSupport.unpark(s.thread);  // 唤醒
}

public final void acquireShared(int arg) {
    // tryAcquireShared返回值：
    // < 0: 获取失败
    // = 0: 获取成功，但后续共享获取不会成功
    // > 0: 获取成功，后续共享获取也可能成功
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean interrupted = false;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 成功获取，设置head并传播
                    setHeadAndPropagate(node, r);
                    p.next = null;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node))
                interrupted |= parkAndCheckInterrupt();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    } finally {
        if (interrupted)
            selfInterrupt();
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    setHead(node);

    // 如果还有剩余资源或需要传播，唤醒后继共享节点
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == 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;
            if (ws == Node.SIGNAL) {
                if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
                    continue;
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}

public class ReentrantLock {
    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        // 非公平尝试获取
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                // 无锁，CAS获取
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                // 重入
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

        // 释放
        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    }

    // 非公平锁
    static final class NonfairSync extends Sync {
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

    // 公平锁
    static final class FairSync extends Sync {
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                // 检查队列中是否有等待者
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                setState(nextc);
                return true;
            }
            return false;
        }
    }
}

public class Semaphore {
    private final Sync sync;

    abstract static class Sync extends AbstractQueuedSynchronizer {
        Sync(int permits) {
            setState(permits);  // state = 许可数量
        }

        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

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

    // 使用
    // Semaphore sem = new Semaphore(3);
    // sem.acquire();  // state: 3 -> 2
    // sem.acquire();  // state: 2 -> 1
    // sem.release();  // state: 1 -> 2
}

public class CountDownLatch {
    private final Sync sync;

    private static final class Sync extends AbstractQueuedSynchronizer {
        Sync(int count) {
            setState(count);  // state = 计数
        }

        int getCount() {
            return getState();
        }

        // 共享获取：count为0时成功
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        // 共享释放：count减1
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c - 1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;  // 减到0时返回true，触发唤醒
            }
        }
    }

    // 使用
    // CountDownLatch latch = new CountDownLatch(3);
    // latch.countDown();  // state: 3 -> 2
    // latch.countDown();  // state: 2 -> 1
    // latch.countDown();  // state: 1 -> 0，唤醒所有await的线程
    // latch.await();      // state != 0 时阻塞
}

public class ReentrantReadWriteLock {
    // state的高16位表示读锁数量，低16位表示写锁数量
    static final int SHARED_SHIFT   = 16;
    static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
    static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
    static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

    // 读锁数量
    static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
    // 写锁数量
    static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

    abstract static class Sync extends AbstractQueuedSynchronizer {
        // 写锁获取（独占）
        protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            int c = getState();
            int w = exclusiveCount(c);
            if (c != 0) {
                // 有锁
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;  // 有读锁，或写锁不是自己的
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                setState(c + acquires);  // 写锁重入
                return true;
            }
            // 无锁，CAS获取写锁
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }

        // 读锁获取（共享）
        protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            // 有写锁且不是自己持有
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            int r = sharedCount(c);
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                // 成功获取读锁
                return 1;
            }
            return fullTryAcquireShared(current);
        }
    }
}

public class SimpleMutex {
    private final Sync sync = new Sync();

    private static class Sync extends AbstractQueuedSynchronizer {
        // 是否被独占
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        // 尝试获取锁
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        // 尝试释放锁
        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0)
                throw new IllegalMonitorStateException();
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    public void lock() {
        sync.acquire(1);
    }

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

    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    public Condition newCondition() {
        return sync.newCondition();
    }
}

public class SharedLock {
    private final Sync sync;

    public SharedLock(int count) {
        sync = new Sync(count);
    }

    private static class Sync extends AbstractQueuedSynchronizer {
        Sync(int count) {
            setState(count);
        }

        @Override
        protected int tryAcquireShared(int arg) {
            for (;;) {
                int current = getState();
                int newCount = current - arg;
                if (newCount < 0 || compareAndSetState(current, newCount)) {
                    return newCount;
                }
            }
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            for (;;) {
                int current = getState();
                int newCount = current + arg;
                if (compareAndSetState(current, newCount)) {
                    return true;
                }
            }
        }
    }

    public void lock() {
        sync.acquireShared(1);
    }

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

public class OneShotLatch {
    private final Sync sync = new Sync();

    private static class Sync extends AbstractQueuedSynchronizer {
        // state: 0=关闭，1=打开

        @Override
        protected int tryAcquireShared(int arg) {
            // 打开状态返回1（成功），关闭状态返回-1（失败）
            return getState() == 1 ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            setState(1);  // 打开闭锁
            return true;
        }
    }

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

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

// 使用
OneShotLatch latch = new OneShotLatch();

// 等待线程
new Thread(() -> {
    latch.await();  // 阻塞直到signal
    System.out.println("继续执行");
}).start();

// 触发线程
Thread.sleep(1000);
latch.signal();  // 打开闭锁，唤醒所有等待线程

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();

    // 1. 加入条件队列
    Node node = addConditionWaiter();

    // 2. 完全释放锁
    int savedState = fullyRelease(node);

    int interruptMode = 0;
    // 3. 循环检查是否在同步队列中
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);  // 阻塞
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }

    // 4. 被signal后，重新获取锁
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;

    // 5. 清理取消的节点
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();

    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();

    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

private void doSignal(Node first) {
    do {
        // 移除条件队列头节点
        if ((firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&  // 转移到同步队列
             (first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
    // CAS修改状态：CONDITION -> 0
    if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
        return false;

    // 加入同步队列尾部
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);  // 唤醒
    return true;
}