精华内容
下载资源
问答
  • AQS源码分析

    2020-05-10 17:00:50
    AQS源码分析AQS的成员变量ReentrantLook抢占锁 AQS的成员变量 /** * FIFO队列头部的指针 */ private transient volatile Node head; /** * FIFO队列屁股的指针 */ private transient volatile Node tail; ...

    AQS的成员变量

    	/**
         *	FIFO队列头部的指针
         */
        private transient volatile Node head;
    
        /**
         * FIFO队列屁股的指针
         */
        private transient volatile Node tail;
    
        /**
         * 同步状态值,AQS的核心
         */
        private volatile int state;
    

    ReentrantLook非公平抢占锁

    java.util.concurrent.locks.ReentrantLock.NonfairSync

        /**
         * Sync object for non-fair locks
         */
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = 7316153563782823691L;
    
            /**
             * Performs lock.  Try immediate barge, backing up to normal
             * acquire on failure.
             */
            final void lock() {
            	//直接调用AQS的方法通过CAS的方式获取锁
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                	//获取锁失败
                    acquire(1);
            }
    		//子类实现的AQS的tryAcquire方法
            protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);
            }
        }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#compareAndSetState

    	//可以看到直接使用unsafe类操控内存
    	protected final boolean compareAndSetState(int expect, int update) {
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire

        public final void acquire(int arg) {
        	//MethodTemplate模式,调用子类的实现方法tryAcquire(arg),获取锁失败返回false取反
            if (!tryAcquire(arg) &&
            	//addWaiter方法 加入FIFO先进先出队列,参数 独占锁,1
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    

    子类的tryAcquire(arg)又调用nonfairTryAcquire(acquires)方法

    java.util.concurrent.locks.ReentrantLock.Sync#nonfairTryAcquire

            /**
             * 会再次判断satrt的值,尝试抢占,并且判断重入锁,都失败返回false
             */
            final boolean nonfairTryAcquire(int acquires) {
            	//保存当前线程到current变量
                final Thread current = Thread.currentThread();
                //调用AQS的final getState()方法 获取start值
                int c = getState();
                //如果start==0,代表可以抢占锁
                if (c == 0) {
                	//用cas的方式去获取锁,尝试把start值改成1
                    if (compareAndSetState(0, acquires)) {
                    	//设置当前拥有独占访问权的线程
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                //判断占有锁的线程是否是当前线程,如果是当前线程,就把satrt+acquires(就是参数值1)
                //重入锁的具体实现,重入锁释放锁时也要一次减1
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#addWaiter

        /**
         * 创建新节点,添加到队列末尾
         *
         * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
         * @return the new node
         */
        private Node addWaiter(Node mode) {
        	//创建新线程节点
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            //获取到FIFO屁股的线程节点
            Node pred = tail;
            if (pred != null) {
            	//如果队列里还有节点,把新创建的node的per指针指向它
                node.prev = pred;
                //cas方式把两个node关联
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            //直接调用enq(node)方法将当前线程加入等待队列尾部
            enq(node);
            return node;
        }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#enq

       /**
    	* 把新节点添加到队列结尾
    	*/
        private Node enq(final Node node) {
        	//这里是循环如果第一遍没有走else,一定会执行第二遍到else,队列里面就会有两个节点
            for (;;) {
                Node t = tail;
                //FIFO队列尾巴节点为null
                if (t == null) { // Must initialize
                	//新创建一个头节点,并且把头尾设置为同一个值
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                	//设置新建节点的pre节点
                    node.prev = t;
                    //把新节点添加到FIFO队列尾巴
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireQueued

        /**
         * 会先判断当前元素的前一个元素是不是head节点,如果是头节点就尝试获得锁,
         * 获得锁成功就把当前节点设置为head节点,并且返回false终止点用着if判断。
         * 如果不是head节点,就直接LockSuport.park挂起线程在队列中等待
         * 
         * @param node the node
         * @param arg the acquire argument
         * @return {@code true} if interrupted while waiting
         */
        final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                //又是一个死循环
                for (;;) {
                	//获取当前节点的前一个节点
                    final Node p = node.predecessor();
                    //如果前一个节点是head节点 && 尝试获得锁成功
                    if (p == head && tryAcquire(arg)) {
                    	//就把当传递进来的node设置为head结点
                        setHead(node);
                        //把之前的节点设置为null方便GC
                        p.next = null; // help GC
                        failed = false;
                        //返回一个false
                        return interrupted;
                    }
                    //如果不是头节点,调用LockSupport.park()挂起当前线程
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
        
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
                return true;
            if (ws > 0) {
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
                do {
                    node.prev = pred = pred.prev;
                } 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.
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }    
    
    	private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }
    

    ReentrantLock释放锁

    java.util.concurrent.locks.ReentrantLock#unlock

        public void unlock() {
        	//调用AQS的release方法释放锁
            sync.release(1);
        }
    
        /**
         * Releases in exclusive mode.  Implemented by unblocking one or
         * more threads if {@link #tryRelease} returns true.
         * This method can be used to implement method {@link Lock#unlock}.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryRelease} but is otherwise uninterpreted and
         *        can represent anything you like.
         * @return the value returned from {@link #tryRelease}
         */
        public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                //head节点不等于null,并且它的等待状态值不等于0
                if (h != null && h.waitStatus != 0)
                	//释放 LockSuport.unpark
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    

    java.util.concurrent.locks.ReentrantLock.Sync#tryRelease

            protected final boolean tryRelease(int releases) {
            	//start值-1
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
                    free = true;
                    //解除当前线程占用
                    setExclusiveOwnerThread(null);
                }
                //设置satrt值=0
                setState(c);
                return free;
            }
    

    java.util.concurrent.locks.AbstractQueuedSynchronizer#unparkSuccessor

       /**
         * Wakes up node's successor, if one exists.
         *
         * @param node the node
         */
        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;
            //等待状态值小于0
            if (ws < 0)
            	
                compareAndSetWaitStatus(node, ws, 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)
            	//唤醒下一个节点的线程,之前没有抢到锁,这里unpark唤醒next继续抢占锁
                LockSupport.unpark(s.thread);
        }
    

    ReentrantLook公平抢占锁

    不同之处在于构造方法

        /**
         * 非公平是默认的构造方法,会把内部类sync指针指向NonfairSync
         */
        public ReentrantLock() {
            sync = new NonfairSync();
        }
    
        /**
         * 公平锁会把内部类sync指针指向FairSync
         */
        public ReentrantLock(boolean fair) {
            sync = fair ? new FairSync() : new NonfairSync();
        }
    

    java.util.concurrent.locks.ReentrantLock.FairSync

        static final class FairSync extends Sync {
            private static final long serialVersionUID = -3000897897090466540L;
    
            final void lock() {
                acquire(1);
            }
    
            /**
             * Fair version of tryAcquire.  Don't grant access unless
             * recursive call or no waiters or is first.
             */
            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;
            }
        }
    

    公平锁的源代码就不细看了,不同之处在于:

    非公平锁:有新的线程抢占锁的时候不会管FIFO队列里是否有其他线程排队,会直接CAS的方式抢占锁,如果抢占成功,之前队列中排队的线程依旧会被park,抢占失败了才会进入FIFO队列中排队。

    公平锁:新线程加入的时候,会先判断队列中是否存在节点,如果存在节点,新线程节点会直接进入队列中排队。

    非公平锁性能高于公平锁性能。非公平锁可以减少CPU唤醒线程的开销,整体的吞吐效率会高点,CPU也不必取唤醒所有线程,会减少唤起线程的数量
    非公平锁性能虽然优于公平锁,但是会存在导致线程饥饿的情况。在最坏的情况下,可能存在某个线程一直获取不到锁。不过相比性能而言,饥饿问题可以暂时忽略,这可能就是ReentrantLock默认创建非公平锁的原因之一了。

    AQS继承图

    在这里插入图片描述看图会发现 ReentrantLock , ReentrantReadWriteLock , ConutDownLatch , Semaphore 都有一个Sync的子类,每个Sync类都继承了AQS,每个Sync的子类又有FairSync(公平)和NonFairSync(非公平)两个子类。

    AQS的核心就是volatile state状态值 和 内部维护的FIFO队列,根据子类不同的实现 state所代表的意义也不一样。

    戏入人生
    有理解不正确的地方,请留言指正,谢谢。

    展开全文
  • 并发-AQS源码分析

    万次阅读 2019-11-04 17:55:37
    一、概述 ... 类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch...。 二、框架 ...

    微信搜索:“二十同学” 公众号,欢迎关注一条不一样的成长之路

    一、概述

      谈到并发,不得不谈ReentrantLock;而谈到ReentrantLock,不得不谈AbstractQueuedSynchronizer(AQS)!

      类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch...。

     

    二、框架

      它维护了一个volatile int state(代表共享资源)和一个FIFO线程等待队列(多线程争用资源被阻塞时会进入此队列)。这里volatile是核心关键词,具体volatile的语义,在此不述。state的访问方式有三种:

    • getState()
    • setState()
    • compareAndSetState()

      AQS定义两种资源共享方式:Exclusive(独占,只有一个线程能执行,如ReentrantLock)和Share(共享,多个线程可同时执行,如Semaphore/CountDownLatch)。

      不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:

    • isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
    • tryAcquire(int):独占方式。尝试获取资源,成功则返回true,失败则返回false。
    • tryRelease(int):独占方式。尝试释放资源,成功则返回true,失败则返回false。
    • tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
    • tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回true,否则返回false。

      以ReentrantLock为例,state初始化为0,表示未锁定状态。A线程lock()时,会调用tryAcquire()独占该锁并将state+1。此后,其他线程再tryAcquire()时就会失败,直到A线程unlock()到state=0(即释放锁)为止,其它线程才有机会获取该锁。当然,释放锁之前,A线程自己是可以重复获取此锁的(state会累加),这就是可重入的概念。但要注意,获取多少次就要释放多么次,这样才能保证state是能回到零态的。

      再以CountDownLatch以例,任务分为N个子线程去执行,state也初始化为N(注意N要与线程个数一致)。这N个子线程是并行执行的,每个子线程执行完后countDown()一次,state会CAS减1。等到所有子线程都执行完后(即state=0),会unpark()主调用线程,然后主调用线程就会从await()函数返回,继续后余动作。

      一般来说,自定义同步器要么是独占方法,要么是共享方式,他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式,如ReentrantReadWriteLock。

    三、源码详解

      本节开始讲解AQS的源码实现。依照acquire-release、acquireShared-releaseShared的次序来。

    3.1 acquire(int)

      此方法是独占模式下线程获取共享资源的顶层入口。如果获取到资源,线程直接返回,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。这也正是lock()的语义,当然不仅仅只限于lock()。获取到资源后,线程就可以去执行其临界区代码了。下面是acquire()的源码:

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

    函数流程如下:

    1. tryAcquire()尝试直接去获取资源,如果成功则直接返回;
    2. addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
    3. acquireQueued()使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
    4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

      这时单凭这4个抽象的函数来看流程还有点朦胧,不要紧,看完接下来的分析后,你就会明白了。就像《大话西游》里唐僧说的:等你明白了舍生取义的道理,你自然会回来和我唱这首歌的。

    3.1.1 tryAcquire(int)

      此方法尝试去获取独占资源。如果获取成功,则直接返回true,否则直接返回false。这也正是tryLock()的语义,还是那句话,当然不仅仅只限于tryLock()。如下是tryAcquire()的源码:

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

      什么?直接throw异常?说好的功能呢?好吧,还记得概述里讲的AQS只是一个框架,具体资源的获取/释放方式交由自定义同步器去实现吗?就是这里了!!!AQS这里只定义了一个接口,具体资源的获取交由自定义同步器去实现了(通过state的get/set/CAS)!!!至于能不能重入,能不能加塞,那就看具体的自定义同步器怎么去设计了!!!当然,自定义同步器在进行资源访问时要考虑线程安全的影响。

      这里之所以没有定义成abstract,是因为独占模式下只用实现tryAcquire-tryRelease,而共享模式下只用实现tryAcquireShared-tryReleaseShared。如果都定义成abstract,那么每个模式也要去实现另一模式下的接口。说到底,Doug Lea还是站在咱们开发者的角度,尽量减少不必要的工作量。

    3.1.2 addWaiter(Node)

      此方法用于将当前线程加入到等待队列的队尾,并返回当前线程所在的结点。还是上源码吧:

     private Node addWaiter(Node mode) {
          //以给定模式构造结点。mode有两种:EXCLUSIVE(独占)和SHARED(共享)
          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入队。
         enq(node);
         return node;
     }

     不用再说了,直接看注释吧。

    3.1.2.1 enq(Node)

       此方法用于将node加入队尾。源码如下:

    private Node enq(final Node node) {
          //CAS"自旋",直到成功加入队尾
          for (;;) {
              Node t = tail;
              if (t == null) { // 队列为空,创建一个空的标志结点作为head结点,并将tail也指向它。
                  if (compareAndSetHead(new Node()))
                      tail = head;
              } else {//正常流程,放入队尾
                  node.prev = t;
                 if (compareAndSetTail(t, node)) {
                     t.next = node;
                     return t;
                 }
             }
         }
     }

     

    如果你看过AtomicInteger.getAndIncrement()函数源码,那么相信你一眼便看出这段代码的精华。CAS自旋volatile变量,是一种很经典的用法。还不太了解的,自己去百度一下吧。

    3.1.3 acquireQueued(Node, int)

      OK,通过tryAcquire()和addWaiter(),该线程获取资源失败,已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一部该干什么了吧:进入等待状态休息,直到其他线程彻底释放资源后唤醒自己,自己再拿到资源,然后就可以去干自己想干的事了。没错,就是这样!是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事:在等待队列中排队拿号(中间没其它事干可以休息),直到拿到号后再返回。这个函数非常关键,还是上源码吧:

    final boolean acquireQueued(final Node node, int arg) {
          boolean failed = true;//标记是否成功拿到资源
          try {
              boolean interrupted = false;//标记等待过程中是否被中断过
              
              //又是一个“自旋”!
              for (;;) {
                  final Node p = node.predecessor();//拿到前驱
                  //如果前驱是head,即该结点已成老二,那么便有资格去尝试获取资源(可能是老大释放完资源唤醒自己的,当然也可能被interrupt了)。
                 if (p == head && tryAcquire(arg)) {
                     setHead(node);//拿到资源后,将head指向该结点。所以head所指的标杆结点,就是当前获取到资源的那个结点或null。
                     p.next = null; // setHead中node.prev已置为null,此处再将head.next置为null,就是为了方便GC回收以前的head结点。也就意味着之前拿完资源的结点出队了!
                     failed = false;
                     return interrupted;//返回等待过程中是否被中断过
                 }
                 
                 //如果自己可以休息了,就进入waiting状态,直到被unpark()
                 if (shouldParkAfterFailedAcquire(p, node) &&
                     parkAndCheckInterrupt())
                     interrupted = true;//如果等待过程中被中断过,哪怕只有那么一次,就将interrupted标记为true
             }
         } finally {
             if (failed)
                 cancelAcquire(node);
         }
     }

     

    到这里了,我们先不急着总结acquireQueued()的函数流程,先看看shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()具体干些什么。

    3.1.3.1 shouldParkAfterFailedAcquire(Node, Node)

      此方法主要用于检查状态,看看自己是否真的可以去休息了,万一队列前边的线程都放弃了只是瞎站着,那也说不定,对吧!

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
          int ws = pred.waitStatus;//拿到前驱的状态
          if (ws == Node.SIGNAL)
              //如果已经告诉前驱拿完号后通知自己一下,那就可以安心休息了
              return true;
          if (ws > 0) {
              /*
               * 如果前驱放弃了,那就一直往前找,直到找到最近一个正常等待的状态,并排在它的后边。
               * 注意:那些放弃的结点,由于被自己“加塞”到它们前边,它们相当于形成一个无引用链,稍后就会被保安大叔赶走了(GC回收)!
              */
             do {
                 node.prev = pred = pred.prev;
             } while (pred.waitStatus > 0);
             pred.next = node;
         } else {
              //如果前驱正常,那就把前驱的状态设置成SIGNAL,告诉它拿完号后通知自己一下。有可能失败,人家说不定刚刚释放完呢!
             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
         }
         return false;
     }

     

    整个流程中,如果前驱结点的状态不是SIGNAL,那么自己就不能安心去休息,需要去找个安心的休息点,同时可以再尝试下看有没有机会轮到自己拿号。

    3.1.3.2 parkAndCheckInterrupt()

      如果线程找好安全休息点后,那就可以安心去休息了。此方法就是让线程去休息,真正进入等待状态。

    private final boolean parkAndCheckInterrupt() {
         LockSupport.park(this);//调用park()使线程进入waiting状态
         return Thread.interrupted();//如果被唤醒,查看自己是不是被中断的。
     }

       park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:1)被unpark();2)被interrupt()。需要注意的是,Thread.interrupted()会清除当前线程的中断标记位。 

    3.1.3.3 小结

      OK,看了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt(),现在让我们再回到acquireQueued(),总结下该函数的具体流程:

    1. 结点进入队尾后,检查状态,找到安全休息点;
    2. 调用park()进入waiting状态,等待unpark()或interrupt()唤醒自己;
    3. 被唤醒后,看自己是不是有资格能拿到号。如果拿到,head指向当前结点,并返回从入队到拿到号的整个过程中是否被中断过;如果没拿到,继续流程1。

     

    3.1.4 小结

      OKOK,acquireQueued()分析完之后,我们接下来再回到acquire()!再贴上它的源码吧:

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

    再来总结下它的流程吧:

    1. 调用自定义同步器的tryAcquire()尝试直接去获取资源,如果成功则直接返回;
    2. 没成功,则addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
    3. acquireQueued()使线程在等待队列中休息,有机会时(轮到自己,会被unpark())会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
    4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

    由于此函数是重中之重,我再用流程图总结一下:

    至此,acquire()的流程终于算是告一段落了。这也就是ReentrantLock.lock()的流程,不信你去看其lock()源码吧,整个函数就是一条acquire(1)!!!

    3.2 release(int)

       上一小节已经把acquire()说完了,这一小节就来讲讲它的反操作release()吧。此方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。这也正是unlock()的语义,当然不仅仅只限于unlock()。下面是release()的源码:

    public final boolean release(int arg) {
         if (tryRelease(arg)) {
             Node h = head;//找到头结点
             if (h != null && h.waitStatus != 0)
                 unparkSuccessor(h);//唤醒等待队列里的下一个线程
             return true;
         }
         return false;
     }

      逻辑并不复杂。它调用tryRelease()来释放资源。有一点需要注意的是,它是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自定义同步器在设计tryRelease()的时候要明确这一点!!

    3.2.1 tryRelease(int)

      此方法尝试去释放指定量的资源。下面是tryRelease()的源码:

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

      跟tryAcquire()一样,这个方法是需要独占模式的自定义同步器去实现的。正常来说,tryRelease()都会成功的,因为这是独占模式,该线程来释放资源,那么它肯定已经拿到独占资源了,直接减掉相应量的资源即可(state-=arg),也不需要考虑线程安全的问题。但要注意它的返回值,上面已经提到了,release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自义定同步器在实现时,如果已经彻底释放资源(state=0),要返回true,否则返回false。

    3.2.2 unparkSuccessor(Node)

      此方法用于唤醒等待队列中下一个线程。下面是源码:

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

     

      这个函数并不复杂。一句话概括:用unpark()唤醒等待队列中最前边的那个未放弃线程,这里我们也用s来表示吧。此时,再和acquireQueued()联系起来,s被唤醒后,进入if (p == head && tryAcquire(arg))的判断(即使p!=head也没关系,它会再进入shouldParkAfterFailedAcquire()寻找一个安全点。这里既然s已经是等待队列中最前边的那个未放弃线程了,那么通过shouldParkAfterFailedAcquire()的调整,s也必然会跑到head的next结点,下一次自旋p==head就成立啦),然后s把自己设置成head标杆结点,表示自己已经获取到资源了,acquire()也返回了!!And then, DO what you WANT!

    3.2.3 小结

      release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。

    3.3 acquireShared(int)

      此方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源,获取成功则直接返回,获取失败则进入等待队列,直到获取到资源为止,整个过程忽略中断。下面是acquireShared()的源码:

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

           这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了:负值代表获取失败;0代表获取成功,但没有剩余资源;正数表示获取成功,还有剩余资源,其他线程还可以去获取。所以这里acquireShared()的流程就是:

    1. tryAcquireShared()尝试获取资源,成功则直接返回;
    2. 失败则通过doAcquireShared()进入等待队列,直到获取到资源为止才返回。

    3.3.1 doAcquireShared(int)

      此方法用于将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回。下面是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) {//如果到head的下一个,因为head是拿到资源的线程,此时node被唤醒,很可能是head用完资源来唤醒自己的
                      int r = tryAcquireShared(arg);//尝试获取资源
                     if (r >= 0) {//成功
                         setHeadAndPropagate(node, r);//将head指向自己,还有剩余资源可以再唤醒之后的线程
                         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);
         }
     }

      有木有觉得跟acquireQueued()很相似?对,其实流程并没有太大区别。只不过这里将补中断的selfInterrupt()放到doAcquireShared()里了,而独占模式是放到acquireQueued()之外,其实都一样,不知道Doug Lea是怎么想的。

      跟独占模式比,还有一点需要注意的是,这里只有线程是head.next时(“老二”),才会去尝试获取资源,有剩余的话还会唤醒之后的队友。那么问题就来了,假如老大用完后释放了5个资源,而老二需要6个,老三需要1个,老四需要2个。老大先唤醒老二,老二一看资源不够,他是把资源让给老三呢,还是不让?答案是否定的!老二会继续park()等待其他线程释放资源,也更不会去唤醒老三和老四了。独占模式,同一时刻只有一个线程去执行,这样做未尝不可;但共享模式下,多个线程是可以同时执行的,现在因为老二的资源需求量大,而把后面量小的老三和老四也都卡住了。当然,这并不是问题,只是AQS保证严格按照入队顺序唤醒罢了(保证公平,但降低了并发)。

     

    3.3.1.1 setHeadAndPropagate(Node, int)

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

      此方法在setHead()的基础上多了一步,就是自己苏醒的同时,如果条件符合(比如还有剩余资源),还会去唤醒后继结点,毕竟是共享模式!

      doReleaseShared()我们留着下一小节的releaseShared()里来讲。

     

    3.3.2 小结

      OK,至此,acquireShared()也要告一段落了。让我们再梳理一下它的流程:

      1. tryAcquireShared()尝试获取资源,成功则直接返回;
      2. 失败则通过doAcquireShared()进入等待队列park(),直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

      其实跟acquire()的流程大同小异,只不过多了个自己拿到资源后,还会去唤醒后继队友的操作(这才是共享嘛)

    3.4 releaseShared()

      上一小节已经把acquireShared()说完了,这一小节就来讲讲它的反操作releaseShared()吧。此方法是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果成功释放且允许唤醒等待线程,它会唤醒等待队列里的其他线程来获取资源。下面是releaseShared()的源码:

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

      此方法的流程也比较简单,一句话:释放掉资源后,唤醒后继。跟独占模式下的release()相似,但有一点稍微需要注意:独占模式下的tryRelease()在完全释放掉资源(state=0)后,才会返回true去唤醒其他线程,这主要是基于独占下可重入的考量;而共享模式下的releaseShared()则没有这种要求,共享模式实质就是控制一定量的线程并发执行,那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。例如,资源总量是13,A(5)和B(7)分别获取到资源并发运行,C(4)来时只剩1个资源就需要等待。A在运行过程中释放掉2个资源量,然后tryReleaseShared(2)返回true唤醒C,C一看只有3个仍不够继续等待;随后B又释放2个,tryReleaseShared(2)返回true唤醒C,C一看有5个够自己用了,然后C就可以跟A和B一起运行。而ReentrantReadWriteLock读锁的tryReleaseShared()只有在完全释放掉资源(state=0)才返回true,所以自定义同步器可以根据需要决定tryReleaseShared()的返回值。

    3.4.1 doReleaseShared()

      此方法主要用于唤醒后继。下面是它的源码:

     private void doReleaseShared() {
          for (;;) {
              Node h = head;
              if (h != null && h != tail) {
                  int ws = h.waitStatus;
                  if (ws == Node.SIGNAL) {
                      if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                          continue;
                      unparkSuccessor(h);//唤醒后继
                 }
                 else if (ws == 0 &&
                          !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                     continue;
             }
             if (h == head)// head发生变化
                 break;
         }
     }

    3.5 小结

      本节我们详解了独占和共享两种模式下获取-释放资源(acquire-release、acquireShared-releaseShared)的源码,相信大家都有一定认识了。值得注意的是,acquire()和acquireSahred()两种方法下,线程在等待队列中都是忽略中断的。AQS也支持响应中断的,acquireInterruptibly()/acquireSharedInterruptibly()即是,这里相应的源码跟acquire()和acquireSahred()差不多,这里就不再详解了。

     

    四、简单应用

      通过前边几个章节的学习,相信大家已经基本理解AQS的原理了。这里再将“框架”一节中的一段话复制过来:

      不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:

    • isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
    • tryAcquire(int):独占方式。尝试获取资源,成功则返回true,失败则返回false。
    • tryRelease(int):独占方式。尝试释放资源,成功则返回true,失败则返回false。
    • tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
    • tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回true,否则返回false。

      OK,下面我们就以AQS源码里的Mutex为例,讲一下AQS的简单应用。

    4.1 Mutex(互斥锁)

      Mutex是一个不可重入的互斥锁实现。锁资源(AQS里的state)只有两种状态:0表示未锁定,1表示锁定。下边是Mutex的核心源码:

      class Mutex implements Lock, java.io.Serializable {
          // 自定义同步器
          private static class Sync extends AbstractQueuedSynchronizer {
              // 判断是否锁定状态
              protected boolean isHeldExclusively() {
                  return getState() == 1;
              }
      
              // 尝试获取资源,立即返回。成功则返回true,否则false。
             public boolean tryAcquire(int acquires) {
                 assert acquires == 1; // 这里限定只能为1个量
                 if (compareAndSetState(0, 1)) {//state为0才设置为1,不可重入!
                     setExclusiveOwnerThread(Thread.currentThread());//设置为当前线程独占资源
                     return true;
                 }
                return false;
             }
     
             // 尝试释放资源,立即返回。成功则为true,否则false。
             protected boolean tryRelease(int releases) {
                 assert releases == 1; // 限定为1个量
                 if (getState() == 0)//既然来释放,那肯定就是已占有状态了。只是为了保险,多层判断!
                     throw new IllegalMonitorStateException();
                 setExclusiveOwnerThread(null);
                 setState(0);//释放资源,放弃占有状态
                 return true;
             }
         }
     
         // 真正同步类的实现都依赖继承于AQS的自定义同步器!
         private final Sync sync = new Sync();
     
         //lock<-->acquire。两者语义一样:获取资源,即便等待,直到成功才返回。
         public void lock() {
             sync.acquire(1);
         }
     
         //tryLock<-->tryAcquire。两者语义一样:尝试获取资源,要求立即返回。成功则为true,失败则为false。
        public boolean tryLock() {
             return sync.tryAcquire(1);
         }
     
         //unlock<-->release。两者语文一样:释放资源。
         public void unlock() {
             sync.release(1);
         }
     
         //锁是否占有状态
         public boolean isLocked() {
             return sync.isHeldExclusively();
         }
     }

     

      同步类在实现时一般都将自定义同步器(sync)定义为内部类,供自己使用;而同步类自己(Mutex)则实现某个接口,对外服务。当然,接口的实现要直接依赖sync,它们在语义上也存在某种对应关系!!而sync只用实现资源state的获取-释放方式tryAcquire-tryRelelase,至于线程的排队、等待、唤醒等,上层的AQS都已经实现好了,我们不用关心。

      除了Mutex,ReentrantLock/CountDownLatch/Semphore这些同步类的实现方式都差不多,不同的地方就在获取-释放资源的方式tryAcquire-tryRelelase。掌握了这点,AQS的核心便被攻破了!

      

           下面将从实现角度分析AQS是如何完成线程同步,主要包括:同步队列、独占式同步状态获取与释放、共享式同步状态获取与释放、超时获取同步状态等AQS的核心数据结构模板方法。

    同步队列

    AQS依赖同步队列(一个FIFO双向队列)来完成同步状态的管理。当前线程获取同步状态失败时,AQS会将当前线程以及等待状态等信息构造成一个节点(Node)并且将其加入到同步队列中,同时会阻塞当前线程,当同步状态释放时,会把首节点中的线程唤醒,使其再次尝试获取同步状态。

    同步队列中的Node节点用来保存获取同步状态失败的线程引用。等待状态以及前驱和后继节点。

     

        static final class Node {
            /** 表示节点正处在共享模式下等待的标记 **/
            static final Node SHARED = new Node();
            /**表示节点正在以独占模式等待的标记*/
            static final Node EXCLUSIVE = null;
            /**waitStatus值,表示线程已取消 */
            static final int CANCELLED =  1;
            /** waitStatus值,表示后继线程需要取消挂起 */
            static final int SIGNAL    = -1;
            /** waitStatus值,表示线程正在等待条件 */
            static final int CONDITION = -2;
            /**waitStatus值指示下一个acquireShared应无条件传播*/
            static final int PROPAGATE = -3;
    
            /**
             * 状态字段,仅接受值:
             * 
             * SIGNAL:值为-1 ,后继节点的线程处于等待状态,
             * 而当前节点的线程如果释放了同步状态或者被取消,
             * 将会通知后继节点,使后继节点的线程得以运行。
             *     
             * CANCELLED:值为1,由于在同步队列中等待的
             * 线程等待超时或者被中断,需要从同步队列中取消等待,
             * 节点进入该状态将不会变化
             *
             * CONDITION: 值为-2,节点在等待队列中,
             * 节点线程等待在Condition上,当其他线程
             * 对Condition调用了singal方法后,该节点
             * 将会从等待队列中转移到同步队列中,加入到
             * 对同步状态的获取中
             *
             * PROPAGATE: 值为-3,表示下一次共享模式同步
             * 状态获取将会无条件地传播下去
             * 
             * INITIAL: 初始状态值为0 
             */
            volatile int waitStatus;
    
            /**
            * 链接到前驱节点,当前节点/线程依赖它来检查waitStatus。
            * 在入同步队列时被设置,并且仅在移除同步队列时才归零
            * (为了GC的目的)。 此外,在取消前驱节点时,我们在找到
            * 未取消的一个时进行短路,这将始终存在,因为头节点从未
            * 被取消:节点仅作为成功获取的结果而变为头。
            * 被取消的线程永远不会成功获取,并且线程只取消自身,
            * 而不是任何其他节点。
            */
            volatile Node prev;
    
            /**
            * 链接到后续节点,当前节点/线程释放时释放。
            * 在入同步队列期间分配,在绕过取消的前驱节
            * 点时调整,并在出同步队列时取消(为了GC的目的)。
            * enq操作不会分配前驱节点的next字段,直到附加之后,
            * 因此看到一个为null的next字段不一定意味着该节点在
            * 队列的末尾。 但是,如果next字段显示为null,我们
            * 可以从尾部扫描prev,仔细检查。 被取消的节点的next字段
            * 被设置为指向节点本身而不是null,以使isOnSyncQueue更
            * 方便操作。调用isOnSyncQueue时,如果节点(始终
            * 是放置在条件队列上的节点)正等待在同步队列上重新获取,则返回true。
            **/
            volatile Node next;
            /**
            * 将此节点入列的线程。在构造方法里初始化,使用后清零。
            * 链接到下一个节点等待条件,或特殊值SHARED。
            * 因为条件队列只有在保持在独占模式时才被访问,
            * 所以我们只需要一个简单的链接队列来保存节点,
            * 同时等待条件。 然后将它们转移到队列中以重新获取。
            * 并且因为条件只能是排它的,我们通过使用特殊的
            * 值来指示共享模式来保存一个字段。
            */
    
            Node nextWaiter;
    
            /**
             *如果节点在共享模式下等待,则返回true
             */
            final boolean isShared() {
                return nextWaiter == SHARED;
            }
    
          /**
          * 返回上一个节点,如果为null,
          * 则抛出NullPointerException。当前驱节点不为null时使用。
          **/
            final Node predecessor() throws NullPointerException {
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
            /**
            *用于建立初始化head节点或SHARED标记
            **/
            Node() {  
    
            }
            /**
            *由addWaiter使用
            **/
            Node(Thread thread, Node mode) {
                this.nextWaiter = mode;
                this.thread = thread;
            }
            /**
            * 供Condition使用
            */
            Node(Thread thread, int waitStatus) {
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }

     

    Node是构成同步队列的基础,AQS拥有首节点(head)和尾节点(tail),没有成功获取同步状态的线程将会放入到队列的尾部

    同步队列的基本结构:

     

    这里写图片描述

     

    同步器中包含了两个节点类型的引用,一个指向头节点(head),一个指向尾节点(tail),没有获取到锁的线程,加入到队列的过程必须保证线程安全,因此同步器提供了一个基于CAS的设置尾节点的方法CompareAndSetTail(Node expect,Node update),它需要传递当前线程认为的尾节点和当前节点,只有设置成功后,当前节点才能正式与之前的尾节点建立关联。

     

    同步器将节点加入到同步队列的过程如图所示: 
    这里写图片描述

     

    同步队列遵循FIFO,首节点是获取锁成功的节点,首节点的线程在释放锁时,将会唤醒后继节点,而后继节点将会在获取到锁时,将自己设置位首节点,过程如下所示: 
    这里写图片描述

     

    设置首节点是由成功获取锁的线程来完成的,由于只有一个线程能够成功获取锁,因此设置首节点不需要CAS操作。

    AQS同步状态获取与释放

    1. 独占式

    2. 共享式

    3. 独占式超时获取

    独占式同步状态的获取与释放

    通过调用AQS的acquire(int arg)方法可以获取同步状态,该方法对中断不起作用,即由于线程获取同步状态失败后进入同步队列中,后续对线程进行中断操作时,线程不会从同步队列中移出。

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

     

    该方法主要的逻辑是:首先调用自定义同步器实现的tryAcquire(int arg)方法,该方法保证线程安全的获取同步状态,如果同步状态获取失败,则构造同步节点(独占式Node.EXCLUSIVE,同一时刻只能有一个线程成功获取同步状态)并通过addWaiter(Node node)方法将节点加入到同步队列的尾部,最后调用acquireQueued(Node node,int arg)方法,使得节点以“死循环”的方式获取同步状态。如果获取不到则阻塞节点中的线程,而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

        private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
            //快速尝试在尾部添加节点
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;//如果尾部节点存在,那么新加入到队列尾部节点的前驱节点为pred
                if (compareAndSetTail(pred, node)) {//通过CAS操作设置尾部节点
                    pred.next = node;//pred的后续节点为先加入的尾部节点
                    return node;
                }
            }
            enq(node);
            return node;
        }
    
        private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                if (t == null) { // 说明队列为null,必须初始化
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }

     

    以上代码,通过compareAndSetTail(Node expect,Node update)方法来确保节点能够被线程安全的添加到同步队列的尾部。在enq(final Node node)方法中,同步器通过“死循环”来保证节点的正确添加,在死循环中,只有通过CAS将节点设置为尾节点后,当前线程才能从该方法返回,否则,当前线程不断得尝试设置。enq(final Node node)方法将并发添加节点的请求通过CAS变得串行化了。

    节点进入到同步队列后,进入了一个自旋的过程,每个节点都在自省的观察,当条件满足,获取到了同步状态时,就可以从这个自旋中退出,否则继续自旋并且阻塞节点的线程

        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;
                    }
                    //检查和更新未能获取的节点的状态。
          //如果线程应该阻塞,返回true。 这是主要信号
          //控制所有获取循环。 需要pred == node.prev
                    if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()
                        interrupted = true;
                }
            } finally {
                if (failed)//如果需要中断,则把节点从队列中移除
                    cancelAcquire(node);
            }
        }

     

    从以上代码可以知道,只有当线程的前驱节点是头节点才能继续获取同步状态,原因如下:

    1. 头节点是成功获取到同步状态的节点,而头节点的线程释放了同步状态后,将会唤醒后继节点,后继节点的线程被唤醒后需要检查自己的前驱节点是否是头节点。

    2. 维护同步队列的FIFO原则。

    独占式同步状态获取流程如下:

     

    这里写图片描述

     

    当前线程获取同步状态并且执行完了对应的逻辑后,需要释放同步状态,使得后续节点继续获取同步状态,通过调用release(int arg)方法来实现。该方法不但释放同步状态,还可以唤醒后继节点重新获取同步状态

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

     

    总结一下:在获取同步状态时,AQS维护这一个同步队列,获取状态失败的线程,会先创建一个独占式节点,并且加入到同步队列的尾部,同时在同步队列中进行自旋;移除队列的条件是前驱节点为头结点并且成功获取了同步状态。在释放同步状态时,同步器调用tryRelease(int arg)方法释放同步状态,然后唤醒头节点的后继节点。

    共享式获取同步状态

    共享式获取与独占式获取的最主要区别在于同一时刻能否有多个线程同时获取到同步状态。通过调用acquireShared(int arg)方法可以共享式得获取同步状态。

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

     

    在acquireShared(int arg)方法中,同步器调用tryAcquireShared(int arg)方法尝试获取同步状态,其返回值为int类型,当返回值大于0时,表示能够获取同步状态。因此,在共享式获取的自旋过程中,成功获取同步状态并且退出自旋的条件就是tryAcquireShared(int arg)方法返回值大于等于0。共享式释放同步状态状态是通过调用releaseShared(int arg)方法

        public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }

     

    该方法与独占式主要区别在于tryReleaseShared(int arg)方法必须确保同步状态线程安全释放。一般是通过循环和CAS来保证的。因为释放同步状态的操作会同时来自多个线程。

    这里写图片描述

     

    通过截图我们可以知道,CountDownLatch、ReentrantReadWriteLock、Semaphore等都是共享式获取同步状态的。

    独占式超时获取同步状态

    通过调用同步器的doAcquireNanos(int arg,long nanosTimeOut)方法可以超时获取同步状态,即在指定的时间段内获取同步状态,如果获取到同步状态则返回true,否则,返回false。

    doAcquireNanos(int arg,long nanosTimeOut)方法针对超时获取,主要需要计算出需要睡眠的时间间隔nanosTimeout,为了防止过早同时,nanosTimeout计算公式为:nanosTimeout -=now-lastTime,其中now为当前唤醒时间,lastTime为上次唤醒时间,如果nanosTimeout大于0则表示超时时间未到,反之,则表示超时。

     private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
            long lastTime = System.nanoTime();
            final Node node = addWaiter(Node.EXCLUSIVE);
            boolean failed = true;
            try {
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return true;
                    }
                    if (nanosTimeout <= 0)//超时
                        return false;
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        nanosTimeout > spinForTimeoutThreshold)//spinForTimeoutThreshold=1000L
                        LockSupport.parkNanos(this, nanosTimeout);//
                    long now = System.nanoTime();
                    nanosTimeout -= now - lastTime;
                    lastTime = now;
                    if (Thread.interrupted())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

     

    该方法在自旋时,当节点的前驱节点为头节点时尝试获取同步状态,如果获取成功则从方法返回,这个过程和独占式获取过程相似,但是在获取失败时,有所不同。如果当前线程获取同步状态失败时,则判断是否超时(nanosTimeout小于等于0表示已经超时),如果没有超时,重新计算超时间隔nanosTimeout,然后是当前线程等待nanosTimeout纳秒,当已到设置的超时时间,该线程会从LockSupport.parkNanos(Object blocker,long nanos)方法返回。

    如果nanosTimeout小于等于spinForTimeoutThreshold(1000纳秒)时,将不会使该线程进行超时等待,而是进入快速的自旋过程。因为,非常短的超时等待无法做到十分精确,如果这时在进行超时等待,相反会让nanosTimeout的超时从整体上表现的反而 不精确。因此,在超市分长短的场景下,同步器会进入无条件的快速自旋。

    独占式超时获取同步状态的流程 è¿éåå¾çæè¿°

     

     

    展开全文
  • AQS源码分析 (1).pdf

    2021-01-22 10:44:25
    java锁底层实现,AQS源码分析。我在公司内部分享写的,如果想进一步了解,可以私聊
  • JUC(一)-AQS源码分析

    2021-01-21 16:42:15
    AQS源码分析一、锁的介绍1.1 乐观锁/悲观锁1.2 共享锁/独占锁1.3 公平锁/非公平锁1.4 小结二、AQS框架结构介绍2.1 类图2.2 AQS数据结构三、源码详解3.1 acquire源码详解3.2 release源码详解四、从ReentranLock看公平...
  • 2_1AQS源码分析

    2020-05-04 18:48:17
    AQS源码分析 AQS是Concurrent包下的核心类,很多的类包括ReentrantLock, CountDownLatch, CyclicBarrier, ReadWriteLock等包都是通过AQS这个类实现的。 这篇文章将会和大家一起看一下AQS的源码分析,从而了解AQS的...

    AQS源码分析

    AQS是Concurrent包下的核心类,很多的类包括ReentrantLock, CountDownLatch, CyclicBarrier, ReadWriteLock等包都是通过AQS这个类实现的。

    这篇文章将会和大家一起看一下AQS的源码分析,从而了解AQS的内部运行情况。

    AQS的核心内容包括三大块(先了解一下):

    ​ state状态:0表示锁处于释放状态,1-n表示有线程取得锁,可以为n表示可重入锁;

    ​ queue双向链表:没有拿到锁的线程进入queue中等待;

    ​ CAS保证原子操作:因为可能多个线程同时去争抢锁,也就是修改state的值,还有多个线程加入queue队列,所以,通过CAS保证原子性(它的高效性体现在如果不使用CAS,而是对整个queue加synchronized锁,效率会很低)。

    先看一下总体流程图,看不懂没关系,下面一步步分析;

    流程简介:
    当一个线程通过reentrantlock.lock()获取锁的时候,

    会先尝试CAS修改state的状态从而获取锁,成功,直接修改state的状态;

    如果失败,再次尝试修改状态,此次允许可重入;

    如果失败,进入等待队列,然后自旋等待状态被释放;

    这个过程中,只有队列的第二个节点会进行tryAcquire尝试再次获取,

    其他节点,尝试阻塞;通过lockSupport.lock();

    AQS的实现的锁的种类:

    1. 公平和非公平:一个线程通过lock()进来的时候是否直接.acquire()抢锁还是先去队列判断有没有等待的线程;

    2. 可重入:state只有0/1 还是可重入

    3. 共享和独占:共享锁相比与独占锁,可以多个线程同时拿到锁,原理:propogate,每一个线程拿到锁后,循环让下一个线程来拿锁;会先判断是否被独占。 没有详细看,只是看了一下流程?

    4. ReentrantReadWriteLock, 基于共享锁的实现;

    5. Semaphore, 可以控制并发量。基于共享锁的实现;每次.await()将Semaphore-1; 当semaphor信号量减到0的时候,所有的线程propogate获取锁;

    6. CountDownLatch 其他线程.countdown(); latch–;主线程.await(), 当latch为0的时候,其他线程.await()释放

      也是共享模式,多个线程可以CAS操作state–; 当state为0的时候,所有.await()线程开启;

    7. CyclicBarrier .await(). state–。 state为0后,所有的线程开启。

    下面,我们以ReentrantLock加锁解锁的过程为例来分析这个过程。

    1.这是一个demo

    我们获得lock锁,然后对其lock加锁和unlock解锁。接下来看一下lock的源码;
    在这里插入图片描述

    2.ReentrantLock调用Sync类中的lock方法。

    在这里插入图片描述

    Sync什么类呢?它是ReenntrantLock的内部类,

    在这里插入图片描述

    通过看它的继承树可以直到,它是AQS的子类。 NonSync–> Sync --> AQS
    在这里插入图片描述

    3. 再来看sycn中的lock方法

    在这里插入图片描述

    首先这里调用了CAS去抢锁。如果成功了,就设置为exclusive独占锁。如果失败了,调用acquire(1)方法。

    1. 再来看一下acquire方法。

    !](https://img-blog.csdnimg.cn/20200504183845892.png)

    acquire方法,分成了3步,

    第一步:再次尝试获得锁,获取成功,就获得锁,而且这里会进行重入锁的判断,看4.1

    第二步:addWaiter,通过CAS操作将当前Thread构建node加到队列尾中;看4.2

    第三步:acquireQueued,再队列中不断尝试获取锁,或者挂断;看4.3

    4.三步

    4.1 tryAcquire调用的是nonfairTryAcquire(), 默认的是非公平锁,所以这里,直接放这个方法。在这里会进行可重入锁的判断。

    在这里插入图片描述

    4.2 如果没有获得锁,加入队列。这里的2.1 CAS加入队列会尝试加锁,如果失败后,if之前的node.pre = pred还是执行,可能会出现多个node的前node指向了同一个node。 因此,失败后,在enq方法中再次尝试CAS操作。

    在这里插入图片描述

    在这里插入图片描述

    4.3 加锁队列后,acquireQueue会循环尝试获得锁。

    这里分成两块,

    如果当前node的前置节点是个head,也就是正在拿到锁的运行线程,那么它可能即将获得锁,因此,tryAcquire尝试获得锁,如果还是失败,在进入3.3 挂起的条件,如果满足,就挂起,等待前置节点完成后,唤醒它。

    第二块,非第二个node,也就是当前节点的前置节点不是head,那么它可能等待的时间较长,会直接进入3.3尝试挂起.等待前置节点完成后,唤醒它。

    接下来,看一下unlock。释放锁的代码

    在这里插入图片描述

    5. 释放锁

    在这里插入图片描述

    6. tryRelease

    在这里插入图片描述

    7. unPark

    在这里插入图片描述

    8. 公平锁和非公平锁

    public ReentrantLock() {
        // 默认非公平锁
        sync = new NonfairSync();
    }
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
     
    static final class FairSync extends Sync {
        final void lock() {
            acquire(1);
        }
        // AbstractQueuedSynchronizer.acquire(int arg)
        public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                // 1. 和非公平锁相比,这里多了一个判断:是否有线程在等待
                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;
        }
    }
     
     
    static final class NonfairSync extends Sync {
        final void lock() {
            // 2. 和公平锁相比,这里会直接先进行一次CAS,成功就返回了
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
        // AbstractQueuedSynchronizer.acquire(int arg)
        public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }
    /**
     * Performs non-fair tryLock.  tryAcquire is implemented in
     * subclasses, but both need nonfair try for trylock method.
     */
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 这里没有对阻塞队列进行判断
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
    
    

    总结:公平锁和非公平锁只有两处不同:

    1. 非公平锁在调用 lock 后,会调用 CAS 进行一次抢锁
    2. 在 tryAcquire 方法中,如果(state == 0),非公平锁会直接 CAS 抢锁,但是公平锁会判断等待队列是否有线程处于等待状态,

    相对来说,非公平锁会有更好的性能,因为它的吞吐量比较大。当然,非公平锁让获取锁的时间变得更加不确定,可能会导致在阻塞队列中的线程长期处于饥饿状态。

    总结

    在并发环境下,加锁和解锁需要以下三个部件的协调:

    锁状态。 state 的作用,它为 0 的时候代表没有线程占有锁,可以去争抢这个锁,用 CAS 将 state 设为 1,如果 CAS 成功,说明抢到了锁,这样其他线程就抢不到了,如果锁重入的话,state进行 +1 就可以,解锁就是减 1,直到 state 又变为 0,代表释放锁,所以 lock() 和 unlock() 必须要配对啊。然后唤醒等待队列中的第一个线程,让其来占有锁。
    线程的阻塞和解除阻塞。AQS 中采用了 LockSupport.park(thread) 来挂起线程,用 unpark 来唤醒线程。
    阻塞队列。

    争抢这个锁,用 CAS 将 state 设为 1,如果 CAS 成功,说明抢到了锁,这样其他线程就抢不到了,如果锁重入的话,state进行 +1 就可以,解锁就是减 1,直到 state 又变为 0,代表释放锁,所以 lock() 和 unlock() 必须要配对啊。然后唤醒等待队列中的第一个线程,让其来占有锁。
    线程的阻塞和解除阻塞。AQS 中采用了 LockSupport.park(thread) 来挂起线程,用 unpark 来唤醒线程。
    阻塞队列。

    展开全文
  • AQS源码分析之独占式获取锁和释放锁 队列同步器AbstractQueuedSynchronizer(以下简称AQS),是用来构建锁或者其他同步组件的基础框架,ReentrantLock、ReentrantReadWriteLock和CountDownLatch等并发工具类底层都是...

    AQS源码分析之独占式获取锁和释放锁

    队列同步器AbstractQueuedSynchronizer(以下简称AQS),是用来构建锁或者其他同步组件的基础框架,ReentrantLock、ReentrantReadWriteLock和CountDownLatch等并发工具类底层都是通过AQS实现的。

    AQS的数据结构

    同步状态

    AQS内部使用一个被volatile关键字修饰的state属性来表示同步状态,并提供了下面三个方法来对这个同步状态进行操作:

    private volatile int state; // 同步状态
    
    protected final int getState() { // 获取当前同步状态
        return state;
    }
    
    protected final void setState(int newState) { // 设置当前同步状态
        state = newState;
    }
    
    protected final boolean compareAndSetState(int expect, int update) { // 使用cas来修改state,保证高并发情况下操作的原子性
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }
    

    一般来说,state=0表示没有线程来获取锁,state=1表示有一个线程获取锁,state大于表示锁的重入次数。

    疑问:在高并发下,不应该都是使用cas来修改state吗,为什么还要提供一个setState()方法?这里卖个关子,后面解答。

    同步队列

    AQS依赖内部的同步队列(一个先进先出的双向链表)来完成同步状态的管理,当前线程获取同步状态失败时,同步器会将当前线程以及等待状态等信息构造成为一个节点(Node)并将其加入同步队列,同时会阻塞当前线程,当同步状态释放时,会把首节点中的线程唤醒,使其再次尝试获取同步状态。

    同步队列中的节点Node的数据结构:

    static final class Node {
    
        // 线程的等待状态
        // SIGNAL:当前节点的后继节点处于等待状态时,如果当前节点的同步状态被释放或者取消,必须唤起它的后继节点
        // CANCELLED: 一个节点由于超时或者中断需要在CLH队列中取消等待状态,被取消的节点不会再次等待
        // CONDITION:  当前节点在等待队列中,只有当节点的状态设为0的时候该节点才会被转移到同步队列
        // PROPAGATE:  下一次的共享模式同步状态的获取将会无条件的传播
        // waitStatus的初始值时0,使用CAS来修改节点的状态
        volatile int waitStatus;
    
        volatile Node prev; // 前驱节点
    
        volatile Node next; // 后继节点
    
        volatile Thread thread; // 当前节点对应的线程
    }
    

    AQS有头节点和尾节点,这样就构成了一个双向链表:

    private transient volatile Node head; // 头节点
    
    private transient volatile Node tail; // 尾节点
    

    没有资源竞争,就一个线程获取锁

    先来看一个例子,然后根据例子追踪源码:

    package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 没有资源竞争,就一个线程获取锁
     */
    public class NoCompetitionDemo {
    
        public static void main(String[] args) throws InterruptedException {
    
            Thread t1 = new Thread(() -> {
                ReentrantLock reentrantLock = new ReentrantLock();
                reentrantLock.lock();
                try {
                    System.out.println("do something...");
                } finally {
                    reentrantLock.unlock();
                }
            }, "t1");
            t1.start();
            t1.join();
        }
    }
    

    下面的源码分析会根据上面的例子来,抓要点和主干,也就是没有进入的代码暂时不分析,屏蔽一些细节,后面通过其他的场景实例来分析,不然一开始就会很复杂,容易晕。

    reentrantLock.lock()方法实现如下:

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

    sync.lock()方法的实现有两个,一个是公平锁FairSync,一个是非公平锁NonfairSync,默认是非公平锁,这里先看非公平锁lock方法的实现:

    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread()); // 获取锁成功,标记一下哪个线程获得了锁,便于后面的重入
        else
            acquire(1); // 获取锁失败
    }
    

    在上面的例子中,没有资源竞争,就一个线程获取锁,这样只需要使用cas将state设置为1,获取锁成功,然后就直接返回了。

    再来看一下释放锁的过程,ReentrantLock#unlock:

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

    AbstractQueuedSynchronizer#release:

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h); // 同步队列中暂时没有节点,不需要唤醒,t1不会进入
            return true;
        }
        return false;
    }
    

    ReentrantLock.Sync#tryRelease:

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases; // state-1,如果没有重入,此时c=0
        if (Thread.currentThread() != getExclusiveOwnerThread()) // 不是持有锁的线程不能释放锁
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null); // exclusiveOwnerThread=null
        }
        setState(c); // state=0,释放锁的时候没有并发,不需要cas,疑问的答案在这。
        return free;
    }
    

    AQS中锁的变化过程如下:

    在这里插入图片描述

    上面的例子很简单,下面来看一个复杂的。

    有资源竞争,多个线程抢锁

    package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 有资源竞争,多个线程抢锁
     */
    public class MultiThreadDemo {
    
        public static void main(String[] args) throws InterruptedException {
    
            ReentrantLock reentrantLock = new ReentrantLock();
            new Thread(() -> {
                reentrantLock.lock();
                try {
                    System.out.println("t1 do something...");
                    TimeUnit.SECONDS.sleep(100); // 休眠,便于调试与观察
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    reentrantLock.unlock();
                }
            }, "t1").start();
    
            TimeUnit.SECONDS.sleep(1); // 让t1跑起来获得锁
    
            new Thread(() -> {
                reentrantLock.lock();
                try {
                    System.out.println("t2 do something...");
                } finally {
                    reentrantLock.unlock();
                }
            }, "t2").start();
    
            TimeUnit.SECONDS.sleep(1); // 让t2跑起来获得锁
    
            new Thread(() -> {
                reentrantLock.lock();
                try {
                    System.out.println("t3 do something...");
                } finally {
                    reentrantLock.unlock();
                }
            }, "t3").start();
        }
    
    }
    

    上面的例子中,t1会一直持有锁,t2、t3只能进入同步队列等待t1释放锁后被唤醒,t1的加锁流程与之前的例子中的加锁流程一样,这里不再赘述。

    t1加锁后,t2来加锁前,AQS中各个数据结构的值如下:

    在这里插入图片描述

    此时t2来加锁,他会向t1一样先尝试cas设置的值为1,结果失败了,因为锁被t1持有了,然后就会进入AbstractQueuedSynchronizer#acquire方法:

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

    acquire()采用了模板方法模式,tryAcquire()的具体实现在子类NonfairSync#tryAcquire中:

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
    

    ReentrantLock.Sync#nonfairTryAcquire,这里的条件都不满足,所以返回false:

    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) { // t2进来时state=1,这里不会进入
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires; // 这里是锁的重入
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
    

    所以上面的NonfairSync#tryAcquire会返回false,然后就会执行AbstractQueuedSynchronizer#addWaiter:

    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode); // 创建一个Node节点
    
        Node pred = tail;
        if (pred != null) { // t2进来的时候,同步队列还没有被初始化,head=tail=null这里不会进入
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
    
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // t2第一遍循环进来这个if,初始化队列
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) { // t2第二遍循环进来这个else,将node添加到双向链表的尾部
                    t.next = node;
                    return t;
                }
            }
        }
    }
    

    此时,队列大概长这样:

    在这里插入图片描述

    然后程序接着往下执行,会进入到AbstractQueuedSynchronizer#acquireQueued:

    final boolean acquireQueued(final Node node, int arg) { // node为t2
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor(); // 拿到t2的前驱节点,也就是h
                if (p == head && tryAcquire(arg)) { // h是头节点,所以又会调用tryAcquire再次尝试获取锁,这里还是失败,不会进if
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    

    AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { // pred为h,node为t2
        int ws = pred.waitStatus; 
        if (ws == Node.SIGNAL) 
            return true; // t2第二次进入,直接返回true
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL); // t1第一次进入,将h的waitStatus改为Node.SIGNAL(-1),也就是告诉h,你释放锁后要唤醒我
        }
        return false;
    }
    

    shouldParkAfterFailedAcquire这个方法t2第一次进来会把h的waitStatus改为-1,以便h释放锁后,可以通知t2,返回false,然后又会再次进入AbstractQueuedSynchronizer#acquireQueued中的死循环,开始第二次自旋,此时t1还是没有释放锁,所以前面的执行结果还是一样,然后第二次进入shouldParkAfterFailedAcquire这个方法,直接返回true。

    在这里插入图片描述

    最后t2会进入AbstractQueuedSynchronizer#parkAndCheckInterrupt,开始阻塞休眠,直接t1释放锁后,将它唤醒:

    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); // 睡了
        return Thread.interrupted();
    }
    

    下面来分析t3的加锁过程:

    t3和t2的逻辑基本一样,有两个稍微不同的地方:

    1. t3加入队列的时候,队列已经初始化完成了,不需要再去初始化了。
    2. t3的前驱节点不是头结点,在AbstractQueuedSynchronizer#acquireQueued方法不会再去尝试加锁了。

    t3最后也会被进入到同步队列中,然后阻塞休眠,直接t2释放锁后,将它唤醒,此时同步队列中的结构如下:

    在这里插入图片描述

    下面来看一下t1解锁后是怎么唤醒t2的:

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h); // t1在释放锁时,进入此
            return true;
        }
        return false;
    }
    

    AbstractQueuedSynchronizer#unparkSuccessor:

    private void unparkSuccessor(Node node) { // 这个node是h,也就是头节点
    
        int ws = node.waitStatus; // -1
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0); // 先把-1改为0,表示不需要再通知了
    
        Node s = node.next; // s=t1
        if (s == null || s.waitStatus > 0) { // s不会为null,不会进入
            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); // 唤醒t1
    }
    

    在这里插入图片描述

    t2之前是在AbstractQueuedSynchronizer#parkAndCheckInterrupt里面睡的,被t1唤醒后,就会接着往下执行:

    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) { // t2醒来又会进入循环中
                final Node p = node.predecessor(); // p是头结点
                if (p == head && tryAcquire(arg)) { // 这里获取锁会成功,因为t1已经释放了
                    setHead(node); // t2变成头结点
                    p.next = null; // 之前的头结点没有引用指向它,就会被GC回收了
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }
    

    t2获得了锁,同步队列的结构如下:

    在这里插入图片描述

    好了,后面的流程都类似。

    公平锁的加锁流程

    ReentrantLock.FairSync#lock,直接调用acquire方法。不像非公平锁,一上来就直接加锁,不成功才调用acquire:

    final void lock() {
        acquire(1);
    }
    

    acquire方法中会尝试加锁,ReentrantLock.FairSync#tryAcquire:

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

    在加锁之前得先判断同步队列中是否有线程在等待,如果有就不会去加锁,然后把自己加入到等待队列中等待:

    public final boolean hasQueuedPredecessors() {
        Node t = tail;
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }
    
    展开全文
  • AQS源码分析(上)

    2020-06-16 23:32:03
    AQS源码分析–前置知识准备 AQS(AbstractQueuedSynchronizer):抽象式队列同步器 AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,这里通过ReentrantLock来介绍AQS的底层源码时如何实现的 ...
  • ReentrantLock 与 AQS 源码分析 1. 基本结构    重入锁 ReetrantLock,JDK 1.5新增的类,作用与synchronized关键字相当,但比synchronized更加灵活。ReetrantLock本身也是一种支持重进入的锁,即该锁可以支持...
  • AQS源码分析之中断与超时获取锁

    万次阅读 2020-09-23 10:54:56
    AQS源码分析之中断与超时获取锁 中断基础 Thread.interrupt():对象方法,置中断标记为true。 Thread.currentThread().isInterrupted():对象方法,返回线程当前的中断标记状态,不会清除中断标志位。 Thread....
  • AQS源码分析--jdk1.8

    2019-09-18 05:55:59
    ArrayList源码分析--jdk1.8LinkedList源码分析--jdk1.8HashMap源码分析--jdk1.8AQS源码分析--jdk1.8ReentrantLock源码分析--jdk1.8 AbstractQueuedSynchronizer概述   1. AQS是一个基于FIFO队列,可以用于构建锁...
  • AQS源码分析总结(1)

    2021-07-20 21:44:35
    AQS源码分析总结(1) @author:Jingdai @date:2021.07.20 最近研究了一下AQS源码,记录一下,水平有限,不免理解有错,欢迎讨论指正。 整体思路 AQS是一个提供了简化同步类设计的框架,利用AQS可以比较容易的...
  • AQS源码分析 AQS(AbstractQueueedSynchronizer)使用一个int成员变量表示同步状态,通过内置的FIFO队列完成资源获取的排队工作 voaltile state 是为了保证state变量线程的可见性, AQS改变state的方法主要有以下几个...
  • Java并发之AQS源码分析

    2019-05-08 20:47:00
    我在Java并发之AQS源码分析(一)这篇文章中,从源码的角度深度剖析了 AQS 独占锁模式下的获取锁与释放锁的逻辑,如果你把这部分搞明白了,再看共享锁的实现原理,思路就会清晰很多。下面我们继续从源码中窥探共享锁...
  • 1、队列同步器AQS源码分析之概要分析 2、队列同步器AQS源码分析之独占模式 3、队列同步器AQS源码分析之共享模式 4、队列同步器AQS源码分析之Condition接口、等待队列 本文主要讲解队列同步器AQS的独占模式:主要...
  • 1、队列同步器AQS源码分析之概要分析 2、队列同步器AQS源码分析之独占模式 3、队列同步器AQS源码分析之共享模式 4、队列同步器AQS源码分析之Condition接口、等待队列 通过上一篇文章的的分析,我们知道独占模式...
  • 1、队列同步器AQS源码分析之概要分析 2、队列同步器AQS源码分析之独占模式 3、队列同步器AQS源码分析之共享模式 4、队列同步器AQS源码分析之Condition接口、等待队列 通过前面三篇关于AQS文章的学习,我们深入...
  • 1、队列同步器AQS源码分析之概要分析 2、队列同步器AQS源码分析之独占模式 3、队列同步器AQS源码分析之共享模式 4、队列同步器AQS源码分析之Condition接口、等待队列 先推荐两篇不错的博文: 1、一行一行源码...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 13,844
精华内容 5,537
关键字:

aqs源码分析