线程条件队列ConditionObject源码解读

小记
好久没更博,窗外光芒万丈,冬日的晚晨,多么美好,就不浪费了,循着键盘上的点点星辰,开工!

啥子是条件队列?

我们都知道,在万类之祖Object里面定义了几个监视器方法:wait(),notify
(),notifyAll(),配合synchronized语义来控制线程的一些状态,在JDK1.5之后,由Lock替代了synchronized,而这几个监视器由条件队列Condition来实现,以便在某个状态条件现在可能为 true 的另一个线程通知它之前,一直挂起该线程(即让其“等待”),以原子的方式释放锁,并挂起当前线程,所以,也可以叫它为线程的条件队列(自创的)。

来看一个应用示例

在以前刚学习Java的时候写过一个题,题干大概是这样的:开启三个线程依次轮流打印出75个数,且次序不能乱。下面是代码:

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class CountTo_75 {

    public static void main(String[] args) {
        final Box box = new Box();

        new Thread(new Runnable() { // 线程1
            @Override
            public void run() {
                for (int i = 1; i <= 5; i++) {
                    box.main_1();
                }
            }
        },"Thread-1").start();
        new Thread(new Runnable() { // 线程2
            @Override
            public void run() {
                for (int i = 1; i <= 5; i++) {
                    box.main_2();
                }
            }
        },"Thread-2").start();
        new Thread(new Runnable() { // 线程3
            @Override
            public void run() {
                for (int i = 1; i <= 5; i++) {
                    box.main_3();
                }
            }
        },"Thread-3").start();
    }

    static class Box {
        Lock lock = new ReentrantLock();
        Condition condition_1 = lock.newCondition();
        Condition condition_2 = lock.newCondition();
        Condition condition_3 = lock.newCondition();
        private volatile int flag = 1;
        public static int count = 0;   //用于计数的变量

        public void main_1() {
            lock.lock();
            try {
                while(flag != 1){
                    try {
                        condition_1.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                for (int j = 0; j < 5; j++) {
                    count++;
                    System.out.println(Thread.currentThread().getName() +" "+ count);
                }
                flag = 2;
                condition_2.signal();
            } finally {
                lock.unlock();
            }
        }

        public void main_2() {
            lock.lock();
            try {
                while(flag != 2){
                    try {
                        condition_2.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                for (int j = 0; j < 5; j++) {
                    count++;
                    System.out.println(Thread.currentThread().getName() +" " + count);
                }
                flag = 3;
                condition_3.signal();
            } finally {
                lock.unlock();
            }
        }

        public void main_3() {
            lock.lock();
            try {
                while(flag != 3){
                    try {
                        condition_3.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                for (int j = 0; j < 5; j++) {
                    count++;
                    System.out.println(Thread.currentThread().getName() + " " + count);
                }
                flag = 1;
                condition_1.signal();
            } finally {
                lock.unlock();
            }
        }
    }
}

抛开当时混乱的逻辑和性能考虑不足不谈,这段丑陋的代码,终归是实现了功能,而且是运用了Lock和Condition来实现的,用在这里来说明Condition的语义再好不过了。代码中初始化了3个条件队列,分别来控制3个线程的挂起状态,flag变量则控制它们之间的关系。

与Lock之间的实现关系

Condition是一个接口,其实现类只有两个:AQS和AQLS,都以内部类的形式存在,内部类叫做ConditionObject,这里有点纳闷,既然这个类是为Lock专属定制的,为什么不在ReentrantLock里面来实现呢?放在AQS不会太臃肿吗?不知道Doug Lea上神当时是怎么考虑的。
由于在AQS中已经实现,因此在ReentrantLock里面对其操作也是很简单的,创建一个条件队列:

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

sync中的实现:

    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;
        final ConditionObject newCondition() {
            return new ConditionObject();
        }
    }

很简单吧?呵呵,下面来看ConditionObject

ConditionObject实现

首先定义了两个核心成员变量,条件队列的头节点和尾节点:

/** First node of condition queue. */
private transient Node firstWaiter;

/** Last node of condition queue. */
private transient Node lastWaiter;
1、核心方法:await() 不可中断的条件等待实现
public final void await() throws InterruptedException {
     //当前线程被中断则抛异常
     if (Thread.interrupted()) throw new InterruptedException();
     //添加进条件队列
     Node node = addConditionWaiter();
     //释放当前线程持有的锁
     int savedState = fullyRelease(node);
     int interruptMode = 0;
     //在这里一直查找创建的Node节点在不在Sync队列,不在就一直禁用当前线程
     while (!isOnSyncQueue(node)) {
         //park当前线程,直到唤醒
         LockSupport.park(this);
         //如被中断也跳出循环
         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 
         break;
     }
     //此时已被唤醒,获取之前被释放的锁,
     if (acquireQueued(node, savedState) && interruptMode != THROW_IE)   
       interruptMode = REINTERRUPT;
     //移除节点,释放内存
     if (node.nextWaiter != null) // clean up if cancelled
       unlinkCancelledWaiters();
     //被中断后的操作
     if (interruptMode != 0)
       reportInterruptAfterWait(interruptMode);
}

简而言之,await()方法其实就是为当前线程创建一个Node节点,加入到Condition队列并释放锁,之后就一直查看这个节点是否在Sync队列中了(signal()方法将它移到Sync队列),如果在的话就唤醒此线程,重新获取锁。此外,awaitNanos(long nanosTimeout) 方法和await(long time, TimeUnit unit) 方法的实现大同小异,只是在何时跳出while循环的时候加了一个超时罢了。
另外还有几个相关的方法也看一下:

/**
 *  添加一个Node节点到Condition队列中
 */
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;
       }
   }
    /**
     *  检查当前节点在不在Sync队列
     */
   final boolean isOnSyncQueue(Node node) {
       //如果当前节点状态为CONDITION,一定还在Condition队列
       //如果Sync队列的前置节点为null,则表明当前节点一定还在Condition队列
       if (node.waitStatus == Node.CONDITION || node.prev == null)
           return false;
       //有后继节点,也有前置节点,那么一定在Sync队列
       if (node.next != null) // If has successor, it must be on queue
           return true;

       //倒查Node节点,前置节点不能为null,第一个if已经做了判断,其前置节点为non-null,但是当前节点也不在Sync也是可能的,因为CAS操作将其加入队列也可能失败,所以我们需要从尾部开始遍历确保其在队列
       return findNodeFromTail(node);
   }
2、核心方法:signal() 将等待时间最长的线程(如果存在)从Condition队列中移动到拥有锁的Sync队列
   public final void signal() {
       //当前线程非独占线程,报非法监视器状态异常
       if (!isHeldExclusively())
           throw new IllegalMonitorStateException();
       //头节点是等待时间最长的节点
       Node first = firstWaiter;
       if (first != null)
           doSignal(first);
   }
   private void doSignal(Node first) {
         do {
             //如果头节点的下一节点为null,则将Condition的lastWaiter置为null
             if ( (firstWaiter = first.nextWaiter) == null)
                 lastWaiter = null;
             //将头结点的下一个节点设为null
             first.nextWaiter = null;
             //被唤醒并且头节点不为null则结束循环
         } while (!transferForSignal(first) &&
                  (first = firstWaiter) != null);
     }
   final boolean transferForSignal(Node node) {
       //如果无法改变节点状态,说明节点已经被唤醒
       if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
           return false;
       //将当前节点添加到Sync队列尾部,并设置前置节点的waitStatus为SIGNAL,表明后继有节点(可能)将被唤醒,如果取消或者设置waitStatus失败,会唤醒重新同步操作,这时候waitStatus是瞬时的,出现错误也是无妨的
       Node p = enq(node);
       int ws = p.waitStatus;
       if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
           LockSupport.unpark(node.thread);
       return true;
   }
3、核心方法:signalAll() 将所有线程从Condition队列移动到拥有锁的Sync队列中。
   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);
   }

猜你喜欢

转载自blog.csdn.net/rickiyeat/article/details/78637452