多线程进阶——JUC并发编程之抽象同步队列AQS框架设计理念一探究竟🔥

  /**
     * <p>To enqueue into a CLH lock, you atomically splice it in as new
     * tail. To dequeue, you just set the head field.
     * <pre>
     *      +------+  prev +-----+       +-----+
     * head |      | <---- |     | <---- |     |  tail
     *      +------+       +-----+       +-----+
     * </pre>
     *
     * <p>Insertion into a CLH queue requires only a single atomic
     * operation on "tail", so there is a simple atomic point of
     * demarcation from unqueued to queued. Similarly, dequeuing
     * involves only updating the "head". However, it takes a bit
     * more work for nodes to determine who their successors are,
     * in part to deal with possible cancellation due to timeouts
     * and interrupts.
     *
     */

static final class Node {
        /** 共享 */
        static final Node SHARED = new Node();
        /** 独占 */
        static final Node EXCLUSIVE = null;
        /** 因为超时或者中断，节点会被设置为取消状态，被取消的节点是不会参与竞争的，它会一直保持取消状态不会转变为其它状态 */
        static final int CANCELLED =  1; 
        /** 前节点的线程如果释放了同步状态或者被取消会通知后继处于等待状态的节点，使得后继节点得以运行 */
        static final int SIGNAL    = -1;
        /** 节点在Condition等待队列中，当其他线程对Condition调用Signal()后，该节点将会从等待队列转移到同步队列，加入同步状态的获取 */
        static final int CONDITION = -2;
        /** 表示下一次共享模式将会无条件传递下去 */
        static final int PROPAGATE = -3;
        /** 等待状态 */
        volatile int waitStatus;
        /** 前驱节点*/
        volatile Node prev;
        /** 后继节点 */
        volatile Node next;
        /** 获取同步状态的线程 */
        volatile Thread thread;
        /** 用来连接Condition等待队列中的节点 */
        Node nextWaiter;
       ....
   }

public final void acquire(int arg) {
    /**
     * 1 tryAcquire    尝试获取同步状态; 
     * 2 addWaiter     如果尝试获取到同步状态失败，则加入到同步队列中; 
     * 3 acquireQueued 在队列中尝试获取同步状态.
     * 4 selfInterrupt 如果线程在等待过程中被中断过，它是不响应的，只是获取资源后才再将自我中断补上
     */
     if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
         selfInterrupt();
    }

protected boolean tryAcquire(int arg) {
      throw new UnsupportedOperationException();
}

        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;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

//没有获取到同步状态的线程添加到AQS等待队列队尾
private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // 快速入队操作
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            //将当前线程设置到队尾
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //快速入队失败兜底入队的方法
        enq(node);
        return node;
    }
//AQS同步队列中的节点，尝试获取同步状态
  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)) {
                    //当前节点设置为头结点，相当于头结点出队
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                } 
                //如果自己可以休息了，就通过park进入waiting状态，直到被unpark，如果不可中断的情况下被中断了，那么会从park中醒来，发现拿不到资源，继续进入park等待
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())//阻塞挂起当前线程直到被唤醒
                    interrupted = true;//如果等待过程被中断，就将状态标记为true
            }
        } finally {
            if (failed)//如果等待过程中没有成功获取到资源（如timeout，或者可中断的情况下被中断）那么取消节点在队列中的等待
                cancelAcquire(node);
        }
    }

    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;
            //释放成功后，执行unpark,即唤醒操作（暂时可忽略waitStatus涉及到条件队列）
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
//尝试释放同步状态，同步状态-需要释放资源的值
//如果state=0，表示当前线程 获取次数=释放次数，既然释放成功，此时持有同步状态线程标志位null
protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            //状态码=0，表示释放成功了
            if (c == 0) {
                free = true;
                //独占标志设置为null
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

//唤醒后继节点
private void unparkSuccessor(Node node) {
        //这里node一般为当前线程所在的节点
        int ws = node.waitStatus;
        if (ws < 0)
            //置为0当前线程所在的节点状态，允许失败
            compareAndSetWaitStatus(node, ws, 0);
        //找到下一个需要被唤醒的节点
        Node s = node.next;
        //如果为空，或者已经取消了
        if (s == null || s.waitStatus > 0) {
            s = null;
            //从后往前找
            for (Node t = tail; t != null && t != node; t = t.prev)
                //从这里可以看出，<=0 的节点还是有效节点
                if (t.waitStatus <= 0)
                    s = t;
        }
        //唤醒后继节点
        if (s != null)
            LockSupport.unpark(s.thread);
    }

//在AQS等待队列阻塞的节点尝试获取同步状态
//获取成功后，将当前节点设置为头结点，头结点设置为null，即头结点出队
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)) {
                    //a.操作，当前节点设置为头结点，当前节点的前驱节点设置为null
                    setHead(node);
                    //b.原始的head的next设置为null，此时原始的head已经被移出队列
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
//补充a操作代码，设置当前节点的前驱节点为null
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

/** 
  * 共享模式获取同步状态
  * 1、首先至少要调用一次tryAcquireShared方法，获取同步状态
  * 2、加入到同步队列中，循环阻塞与唤醒，直到获取同步状态为止，获取成功会唤醒后面还在等待的共享节点
  * 并把该唤醒事件传递下去，也即依次唤醒在该节点后面的所有共享节点
  */    
public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
/** 自旋模式获取同步状态 */
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) {
                    // r>0 表示获取同步状态成功，并且还有共享类型节点在同步队列中
                    // r == 0 表示获取同步状态成功，但是同步队列中没有其它共享模式节点
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        //获取同步状态成功后，将当前node设置为头结点，并向后传播，唤醒共享模式等待的节点
                        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; // Record old head for check below
        setHead(node);
       //propagate>0表示后面需要唤醒的共享模式节点
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            //如果当前节点的后继节点是共享类型或者没有后继节点，则进行唤醒
            //这里可以理解为除非明确指明不需要唤醒（后继等待节点是独占类型），否则都要唤醒
            if (s == null || s.isShared())
                doReleaseShared();
        }

/** 唤醒所有共享模式节点 */
 private void doReleaseShared() {
       
        for (;;) {
            //唤醒操作由头结点开始，注意这里的头结点已经是上面新设置好的头结点
            //其实就是唤醒上面新获取到的共享模式节点的后继节点
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                //表示后继节点需要被唤醒
                if (ws == Node.SIGNAL) {
                    //这里需要控制并发，因为入口有setHeadAndPropagate和release两个，避免两次UNpark
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);//执行唤醒操作（唤醒后续共享模式节点）
                }
                //如果头结点没有发生变化，表示设置完成，退出循环
                //如果头结点发生变化，比如其它线程获取了锁，为了使自己的唤醒动作可以被传递，必须进行重试
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

    public final boolean releaseShared(int arg) {
        //尝试释放同步状态
        if (tryReleaseShared(arg)) {
            //释放成功后，唤醒后续等待共享节点
            doReleaseShared();
            return true;
        }
        return false;
    }

/**
 * 自定义互斥锁
 */
public class MutexLock {

    private static final Sync STATE_HOLDER = new Sync();

    /**
     * 通过 Sync 内部类来持有同步状态，当状态为1表示锁被持有，0表示锁处于空闲状态
     */
    private static class Sync extends AbstractQueuedSynchronizer {
        /**
         * 是否被独占，有两种表示方式
         * 1、可以根据状态，state=1 表示锁被占用，0 表示空闲
         * 2、可以根据当前独占锁的线程来判断，即 getExclusiveOwnerThread()!=null 表示被独占
         */
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        /**
         * 尝试获取锁，从状态0修改为1，表示锁被当前线程独占了
         * @param acquires
         * @return
         */
        @Override
        protected boolean tryAcquire(int acquires) {
            assert acquires ==1;//这里限定只能为1个
            if (compareAndSetState(0,1)){//state为0才设置为1，不可重入
                setExclusiveOwnerThread(Thread.currentThread());//设置为当前线程独占资源
                return true;
            }
            return false;
        }

        /**
         * 尝试释放资源
         * @param releases
         * @return
         */
        @Override
        protected boolean tryRelease(int releases) {
            assert releases ==1;
            //既然要释放资源，那肯定有资源啦
            if (getState() == 0){
                throw new UnsupportedOperationException();
            }
            setExclusiveOwnerThread(null);
            //释放资源，放弃占有状态
            setState(0);
            return true;
        }
    }
    /**
     * 下面的实现Lock接口需要重写的方法，基本就是调用内部类Sync的方法
     */
    public void lock(){
        STATE_HOLDER.acquire(1);
    }
    public void unlock(){
        STATE_HOLDER.release(1);
    }
}

package com.lock.ext;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义锁 测试
 */
public class MyLockTest {
    public static void main(String[] args) throws InterruptedException {
        int threadNum = 10;
        int countPerThread = 10000;
        //线程池创建的正确姿势
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(threadNum,
                threadNum, 1000, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.AbortPolicy());
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        Counter counter = new Counter();
        Counter counterUnsafe = new Counter();
        for (int i = 0; i < threadNum; i++) {
            poolExecutor.submit(()->{
                for (int j = 0; j < countPerThread; j++) {
                    counter.getAndIncrement();
                    counterUnsafe.getAndIncrementUnSfae();
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        System.out.printf("%s 个线程，每个线程累加了 %s 次，执行结果：safeCounter = %s, unsafeCounter = %s ", threadNum, countPerThread, counter.get(), counterUnsafe.get());
        poolExecutor.shutdownNow();
    }
}


package com.lock.ext;

class Counter {
    private MutexLock mutexLock;
    private volatile int count;

    Counter() {
        this.mutexLock = new MutexLock();
    }

    int get() {
        return count;
    }

    int getAndIncrement() {
        mutexLock.lock();
        count++;
        mutexLock.unlock();
        return count;
    }

    int getAndIncrementUnSfae() {
        count++;
        return count;
    }
}