juc - Condition源码解读

之前写过ReentrantLock的源码的博客,但是还有个重要的东西没有介绍——Condition,他的用法我就不介绍了(这种介绍性的博客我不喜欢写),我今天就写一下Condition的实现原理,还是从Conditon的常用方法入手。

1、await():这个方法由当前获得锁的线程调用,意思是释放锁,挂起自己并且唤醒等待当前线程持有的锁的其他线程(在aqs的等待队列中的其他节点),类似于synchronized同步代码块的wait方法的意思。

 public final void await() throws InterruptedException {
           if (Thread.interrupted())
               throw new InterruptedException();
           Node node = addConditionWaiter();//这个方法是将当前线程放到当前condition的队列中
           int savedState = fullyRelease(node);//释放当前线程站有的锁
           int interruptMode = 0;
           while (!isOnSyncQueue(node)) {//isOnSyncQueue的意思是判断当前当前的线程是否在sync的队列(sync的队列就是aqs的队列),while循环的意思是只要封装当前线程的node没有在aqs的队列中,就一直循环。
               LockSupport.park(this);//将当前的线程挂起,也就是说等当前的线程获得锁之后,还是从这里开始运行
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//这里是判断在线程挂起的时间内,有没有别的线程调用这个线程的iterrupt方法,如果没有的话返回0
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//挂起结束,也就是重新运行之后,调用acquireQueued方法,即进入aqs的队列中,acquireQueued方法在 http://suichangkele.iteye.com/blog/2368173  博客中已经讲了,就是获取锁的代码,如果获取失败则被挂起。
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

 下面把上面设计的方法一个一个的拆开看:

/**
* Adds a new waiter to wait queue. 添加一个新的等待者到wait queue中,注意这里是叫做wait queue,并不是上面说的sync,wait queue是condition自己维护的一个队列。
* @return its new wait node
*/
private Node addConditionWaiter() {
            Node t = lastWaiter;//这里的lastWaiter是wait queue的尾
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();//将取消的节点从队列中删除
                t = lastWaiter;
            }
            Node node = new Node(Thread.currentThread(), Node.CONDITION);//将当前的线程放入到wait queue中,也就是condition自己的队列中,节点的状态是condition.
            if (t == null)//
                firstWaiter = node;//如果当前的队列没有头,则置为头
            else
                t.nextWaiter = node;//有头的话,则一定有尾,将当前的节点放到尾的后面,
            lastWaiter = node;//更新队列的尾。
            return node;
        }

很简单就 看完了上面的代码,他就是将当前的线程放入到condition自己维护的队列中。全部操作都不是cas的,因为await方法只有当前获得锁的线程才可以调用,所以不会发生并发操作。

 

  下一个方法fullRelease方法,顾名思义,全部释放,他的真实意思是:将标记(标记的概念在 http://suichangkele.iteye.com/blog/2368173 这篇博客中)置为0,释放锁,

 

 final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();
            if (release(savedState)) {//将标记变为0,释放锁,并唤醒在aqs的等待队列中的线程。
                failed = false;
                return savedState;
            } 。。。。//下面还有很多,没贴
    }

 经过上面的两步,已经释放了锁,并且当前的线程已经加入了condition的wait queue中。

 

 再看是不是在aqs的队列中的代码:

final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)//如果node的状态是condition,则为false,因为condition的话说明是刚刚加入到condition的wait queue中,也就是刚才持有锁的线程,即使他在aqs的队列中的话他也一定是head,所以要重新放入到aqs的队列中;同理第二个判断node.prev==null,说明如果刚才持有
锁的线程的状态不是condition的话,如果是aqs的队列中的head,也不可以,因为head是持有锁的,就要将其挂起,也就是返回false。理解这一点很重要,也有点困难,他的最根本的逻辑是:如果当前的线程是持有锁的线程,则要将其挂起,否则不挂起。
            return false;
     。。。。。。//还有很多,删了
    }

上面的代码说明了一件事,如果当前线程持有锁,在调用了await之后就要放弃锁,被挂起,唤醒在aqs中等待的其他线程,然后加入到conditon的一个队列中。在唤醒正在aqs队列中的其他线程后,其他线程就会获得锁(这段代码在acquireQueued方法中)。但是需要注意的是,被加入到condition的wait queue的线程此时并不在aqs的等待队列中,也就是说他不会获得锁,如果要让他获得锁,就要看下面的signal方法。

 

 

2、signal(),他就是用来实现上面的问题的,他将在condition的wait queue中等待的线程转移到aqs的队列中:

public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;//firstwaiter就是condition的队列中的head
            if (first != null)
                doSignal(first);
        }
 private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&//transferForSignal的意思就是转移first节点到aqs的队列中。
                     (first = firstWaiter) != null);//直到转移成功first或者队列中没有节点
        }

 

 final boolean transferForSignal(Node node) {
        /* If cannot change waitStatus, the node has been cancelled. */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//将node的waitStatus设置为0,如果设置失败,返回false,继续进行操作,设置成功向下走。
            return false;

        Node p = enq(node);//enq之前说过了,是将其加入到aqs的队列中
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//将node的waitStatus设置为signal,这样在aqs的队列中就和普通的等待着一样了。
            LockSupport.unpark(node.thread);
        return true;
    }

 通过signla方法,之前在condition的等待队列中的线程的head已经转移到aqs的队列中去等待获得锁了,这样就回到了之前ReentrantLock的博客中。

 

看完了源码,再举个例子,就能更好的理解了,我们举多个生产者和多个消费者的例子,生产者生产商品,消费者消费商品,生产出的商品放到容器中,容器中只能放一个商品。刚上来容器中为空,假设消费者获得锁,则所有的消费者在叫做empty的condition上等待,同时释放锁,这时,可能有一个或者多个消费者在empty上等待。此时生产者获得锁(因为只要是消费者获得锁就会进入empty的condition的队列中),将生产的商品放入到容器中,此时容器中已经有商品,其他的生产者发现不能放了,则在full这个condition上等待,也就是进入full这个condition的队列中,生产者放入后,会调用empty.signal,意思是将在empty中等待的线程转移到aqs的队列中,准备获得锁,然后生产者释放所,   消费者获得锁(因为现在容器中有商品了,虽然生产者有机会获得锁,但是只要是生产者持有锁,就会调用full.await,也就是现在条件不满足,虽然我获得了锁,但是必须要等待消费者消费,于是进入full的队列中,所以最终是消费者获得锁)。消费者获得锁后将商品消费,然后调用full.signal,也就是通知生产者生产,这样就回到了最开始的时候。。。。。

 

3、awaitNanos(long nanosTimeout) 带有时间的等待(这个是我后来补充的)

public final long awaitNanos(long nanosTimeout) throws InterruptedException {
       if (Thread.interrupted())//如果当前线程被其他的线程调用interrupt方法,则抛出异常
          throw new InterruptedException();
      Node node = addConditionWaiter();//将当前线程加入到condtion的等待队列中
      int savedState = fullyRelease(node);//释放当前占有的锁
      long lastTime = System.nanoTime();
      int interruptMode = 0;
      while (!isOnSyncQueue(node)) {//循环,直到添加到aqs的队列中
          if (nanosTimeout <= 0L) {//刚上来nanosTimeout不是<=0的,不过后面的parkNanos之后就会符合这个条件。
              transferAfterCancelledWait(node);//这个方法的意思是如果等待的时间足够长了,则转移到aqs的队列中,他是有时间的condition的关键的方法
                  break;
          }
          LockSupport.parkNanos(this, nanosTimeout);//将当前的线程挂起指定的时间,
          if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//在挂起期间内如果调用iterrupt方法则跳出循环。
              break;
          long now = System.nanoTime();
          nanosTimeout -= now - lastTime;//挂起指定时间后继续执行,减小nanosTimeout的值
          lastTime = now;
      }
      if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//在
          interruptMode = REINTERRUPT;
      if (node.nextWaiter != null)
          unlinkCancelledWaiters();
      if (interruptMode != 0)
          reportInterruptAfterWait(interruptMode);
      return nanosTimeout - (System.nanoTime() - lastTime);
 } 

 

transferAfterCancelledWait方法很重要,看看他的实现:

final boolean transferAfterCancelledWait(Node node) {//参数是等待了足够长时间的节点(也就是线程)
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {//如果成功的将节点的属性设置为condition,则将其添加到aqs的等待的队列中
            enq(node);//进入到aqs的队列中
            return true;
        }
        //上面的代码唯一失败的可能是在此期间,condition被调用了signal方法,导致上面的compareAndSetWaitStatus返回false,这时更要将当前线程放入到aqs的队列了,所以调用了当前线程的yield方法,
        while (!isOnSyncQueue(node))
            Thread.yield();//当前线程让步,让另一个线程执行。等待另一个线程(也就是调用condition的signal的方法的线程)执行enq完成。
        return false;
    }

从这个方法可以看出,如果 在有时间的condition中,在时间消耗完成之后并不是会退出,而是直接进入到aqs的队列中等待获得锁。

 

condition就这么简单,就是内部维持了一个队列,在调用await的时候将线程阻塞,释放锁,将其加入到队列中等待,在调用signal的时候将自己队列中的head转移到aqs的队列中,等待获得锁。

 

    原文作者:JUC
    原文地址: https://blog.csdn.net/iteye_14612/article/details/82681126
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞