ConditionObject源码阅读

摘要：前言本来准备做源码阅读的几千行看着太累了看了几篇大神的文章后才基本搞懂附在这里阅读本文前请先看懂的介绍和原理分析并发包源码学习之框架四源码分析接口实现接口一般看一个类实现的接口可以看出它的目的其实也是熟悉的目的主要是替代的方法的它是基于实现

本来准备做AbstractQueuedSynchronizer源码阅读的,几千行看着太累了,看了几篇大神的文章后才基本搞懂,附在这里,阅读本文前请先看懂AQS

AbstractQueuedSynchronizer的介绍和原理分析

Java并发包源码学习之AQS框架（四）AbstractQueuedSynchronizer源码分析

ConditionObeject实现Condition接口,一般看一个类实现的接口可以看出它的目的,其实也是熟悉API.
Condition的目的主要是替代Object的wait,notify,notifyAll方法的,它是基于Lock实现的.(而Lock是来替代synchronized方法).

public interface Condition {

     /** 
      * 暂停此线程直至一下四种情况发生
      * 1.此Condition被signal()
      * 2.此Condition被signalAll()
      * 3.Thread.interrupt()
      * 4.伪wakeup
      * 以上情况.在能恢复方法执行时,当前线程必须要能获得锁
      */
    void await() throws InterruptedException;

 
    //跟上面类似,不过不响应中断
    void awaitUninterruptibly();

    //带超时时间的await()
    long awaitNanos(long nanosTimeout) throws InterruptedException;

    //带超时时间的await()
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    
    //带deadline的await()
    boolean awaitUntil(Date deadline) throws InterruptedException;

    //唤醒某个等待在此condition的线程
    void signal();
  
    //唤醒所有等待在此condition的所有线程
    void signalAll();
}

此类的构造方法没有参数,所以不用讲
整体讲下这个ConditionObject的实现,其实其维护两个队列,

Condition队列,表示等待的队列,其waitStatus=Node.Condition,由firstWaiter和lastWaiter两个属性操控.

Sync队列,表示可以竞争锁的队列,这个跟AQS一致,waitStatus=0;

await()方法呢就是把当前线程创建一个Node加入Condition队列,接着就一致循环查其在不在Sync队列,如果当前节点在Sync队列里了,就可以竞争锁,恢复运行了.

signal()方法就是把某个节点的nextWaiter设为null,再把其从Condition队列转到Sync队列.
具体细节看下面源码及注释:

  /**
   * 暂停此线程,直至许可满足
   * 只看没有中断和超时的await方法,其它处理中断和超时的逻辑不care   
   */
  public final void awaitUninterruptibly() {
    
            Node node = addConditionWaiter();
            //释放当前线程的锁
            int savedState = fullyRelease(node);
            boolean interrupted = false;
            //看自己在不在等待队列,在
            while (!isOnSyncQueue(node)) {
                //为了线程调度,禁用当前的线程,直到许可可用
                LockSupport.park(this);
                if (Thread.interrupted())
                    interrupted = true;
            }
            //获取刚才释放的锁,参考AQS中acquire的讲解吧
            if (acquireQueued(node, savedState) || interrupted)
                selfInterrupt();
        }
        
  /**
   * 添加一个等待Node到队列中   
   */
  private Node addConditionWaiter() {
            Node t = lastWaiter;
            // 如果尾节点被取消了,清理掉
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            // 新建一个Condition状态的节点,并将其加在尾部
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }
        
  /**
   * 释放当前的state,最终还是调用tryRelease方法
   */
  final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }
    
  /**
   * 是否还需要等待
   */
  final boolean isOnSyncQueue(Node node) {
        //如果状态为Node.CONDITION,即还在Condition队列中,还得再循环中等待
        //如果其waitStatus不为Node.CONDITION,其前置节点,即Sync的前置节点为null,为啥也继续等着呢???谁来解答我的疑问
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        //如果状态不为Node.CONDITION,即不在Condition队列了,有前置节点也有后置节点,那么其一定在Sync队列
        if (node.next != null) 
            return true;
        /*
         * 其前置节点为非null,但是也不在Sync也是可能的,因为CAS将其加入队列失败.所以我们需要从尾部开始遍历确保其在队列
         */
        return findNodeFromTail(node);
    }
    /
    
    /**
     * 从尾部查找node节点
     */
    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
    }

/**
 * 释放信号
 */
 public final void signal() {
            //如果不是排它模式，则抛出IllegalMonitorStateException异常
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            //将等待队列的第一个节点出队列，并将其加入AQS的锁队列
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
 }
 
/**
 * 真正的释放信号
 */
private void doSignal(Node first) {
            do {
                //讲first的nextWaiter设为null
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            }
            // 如果转换队列不成功且等待队列不为null,继续do
            while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
}

/**
 * 将一个节点从condition队列转换到Sync队列
 */
final boolean transferForSignal(Node node) {
        //所谓的Condition队列和Sync队列就在于waitStatus值
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
}

/**
 * 唤醒所有等待在此condition的所有线程
 */
public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
}

/**
 * 遍历所有节点,使其加入到Sync队列
 */
private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
    }

在此输入正文