`
春花秋月何时了
  • 浏览: 39546 次
  • 性别: Icon_minigender_1
  • 来自: 成都
社区版块
存档分类
最新评论

Java并发包核心框架AQS之二条件等待与唤醒Condition

 
阅读更多

一、引言

在上一章的AQS同步阻塞与唤醒源码分析中，我们知道作为同步组件的基础核心框架AQS(AbstractQueuedSynchronizer), 其内部层次结构分为三部分，第一部分是AbstractQueuedSynchronizer类本身，第二部分是实现同步队列节点的静态内部类Node，第三部分是内部类ConditionObject。其中AbstractQueuedSynchronizer类自身的大部分 + 静态内部类Node实现了相当于Lock锁的获取与释放操作，Node被实现为AQS内部维护的同步等待队列的节点。关于AQS是如何实现Lock语义的锁的获取与释放操作，以及如何维护同步等待队列在上一章中已经进行了详细的探讨。本章节我们将致力于研究AQS的第三部分内部类ConditionObject所要表达的作用。

 

二、Condition接口基本介绍

通过AQS的源码我们可以知道，其内部类ConditionObject实现了java.util.concurrent.locks.Condition接口，而这个Condition接口同样是Java并发包同步基础组件之一。还记得在我们使用synchronized来控制同步的时候，需要配合Object的wait()、notify()、notifyAll()系列方法实现等待/通知模式的线程通信。那么如果我们使用AQS实现的新的Lock语义的显式锁来控制同步，那么又该如何来实现线程之间的通信呢(因为Object提供的那三个线程之间通信的方法必须包含在synchronized同步块中调用)？其实这就是Condition接口的作用了，Condition接口也提供了类似的Object的监视器的方法，主要包括await()、signal()、signalAll()方法，这些方法与AQS实现的Lock锁配合使用就可以实现等待/通知机制。Condition提供的所有接口方法如下图所示：


  

ASQ的内部类ConditionObject实际上就是实现了对条件等待队列的维护。条件等待的含义是，当某个线程虽然获取到了访问共享资源的锁之后，在运行中发现它需要的进一步的条件还不满足致使其不能继续往下执行，而不得不放弃已经获得的访问共享资源的锁，并让自己陷入沉睡(阻塞)，在将来某个时间点条件满足之后由其他线程来唤醒自己，从而自己可以重新在获取到访问共享资源的锁之后，继续往下执行。这里维护那些当条件不满足的时候，陷入阻塞沉睡状态的线程的队列就是条件队列，不同的条件对应着不同的条件队列，而当条件满足被其他线程唤醒之后，继续尝试获取共享资源锁的过程刚好就是在上一章节中所说的同步等待队列。

 

由此可知，AQS内部其实维护了两种类型的队列：条件等待队列和同步等待队列，其中同步等待队列只有一个，而条件等待队列却可以有多个，每一个都是Condition对象的一个实例，它是通过程序手动绑定到锁对象上的。当执行了某个Condition对象实例的await()方法之后，线程就进入相应的条件等待队列，并将自己阻塞，当其他线程通过调用相同Condition对象实例的signal()/signalAll()方法唤醒自己之后，线程就会由条件等待队列转移到同步等待队列，以尝试竞争同步锁。

 

相比Object实现的监视器方法，Condition接口的监视器方法更加强大，其具有一些Object所没有的特性：

  1. Condition接口可以支持多个条件等待队列，而Object的等待队列只有一个，其与锁对象之间关联在一起。
  2. Condition接口可以支持不响应中断模式的条件等待。我们知道Object的wait()系列方法都是对中断立即响应的，而Condition接口不但提供响应中断的等待，而且也提供不响应中断的等待。
  3. Condition接口可以支持定时等待，虽然Object的wait(timeout)系列方法也提供了超时等待，但却不能设置到精确的将来某一个时间点，而Condition接口提供了这样的功能。

三、AQS内部类ConditionObject中await相关源码分析

知道了Condition接口的重要意义，以及AQS的内部类ConditionObject作为Java并发包基础组件的一部分，它也是Java并发包后面其它重要组成部分的基础，所以我们有必要将其原理进行深入了解，看看AQS是如何来维护条件等待队列的。

关于条件等待队列，AQS在内部也是使用静态内部类Node为节点实现的一个FIFO的队列，所以具体在该Condition对象上等待的线程引用也就自然被包含在Node节点中，Node类已经在上一章节中进行了分析，这里就不在详解，直接来看看ConditionObject的成员属性和构造方法：

public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        
        //条件等待队列的头节点
        private transient Node firstWaiter;
        //条件等待队列的尾节点
        private transient Node lastWaiter;

        //构造方法
        public ConditionObject() { }
    
        //下面两个常量是表示对中断处理的模式，THROW_IE 表示需要立即抛出异常，REINTERRUPT 则表示需要进行自我中断
        private static final int REINTERRUPT =  1;
        private static final int THROW_IE    = -1;
}  

由上面的代码可见，ConditionObject对象的成员属性其实就只有维护条件等待队列的头尾两个节点对象，而构造方法也就只有一个无参的构造方法。接下来我们对其他方法进行详细的分析。

3.1  void await() throws InterruptedException

此方法造成当前线程在接到信号或被中断之前一直处于等待状态。 该方法可以在发生中断的时候立即得到唤醒，但同样需要再次获得同步资源才会返回。

public final void await() throws InterruptedException {
	if (Thread.interrupted()) //在执行之前先进行中断检测
		throw new InterruptedException();
	Node node = addConditionWaiter();//将当前线程添加到条件等待队列队尾
	int savedState = fullyRelease(node);//彻底释放已经被当前线程占用的同步资源，会唤醒后继节点，返回原本占有的同步资源数量
	int interruptMode = 0;
        //如果该节点还不存在与同步等待队列中，就阻塞掉自己
        //这里这样判断的原因是因为signal方法会将条件等待队列中的相应节点转移到同步队列中。
	while (!isOnSyncQueue(node)) {
                //挂起线程直到被唤醒，因为park阻塞存在假唤醒，所以需要在循环中检查是否仍需挂起。
		LockSupport.park(this);
                /**
                  park方法被唤醒之后，发现被中断过(interruptMode将不为0)则跳出循环
                  如果被中断过，这里也会根据情况返回处理中断异常的方式：
                  如果仅仅发生了中断没有发送signal，或者中断发生在signal之前，则返回THROW_IE,
                  如果中断与signal同时发生或者发生在signal之后，则返回REINTERRUPT
                  如果没有中断发生，说明不是被中断唤醒，那么继续判断while条件，是否已经存在于同步等待队列，如果不在同步等待队列则继续阻塞
                **/
		if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
			break;
	}

        //走到这里，表明阻塞被唤醒(不论是发生了中断还是执行了signal)，并且节点已经被转移到同步等待队列
       
        //acquireQueued方法在上一章已经详细分析过，执行独占锁的同步资源获取逻辑，尝试竞争和之前相同数量的同步资源
        //acquireQueued方法只会在成功获取到同步资源之后才会返回，返回就表明成功获取到同步资源，并且返回值为true表明在尝试获取同步资源的过程中发生了中断
        //如果在尝试获取同步资源的过程中发生了中断，但在上面得出的interruptMode不需要立即抛出异常，则设置interruptMode为REINTERRUPT
	if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
		interruptMode = REINTERRUPT;
	if (node.nextWaiter != null) // 如果存在后继条件等待节点，则清理一下条件等待队列
		unlinkCancelledWaiters();//从队列头部至尾部，清除条件队列中所有状态不为Condition的节点

        //根据inerruptMode，集中处理中断异常:
        //仅仅发生了中断，或者中断发生在signal之前立即抛出异常，中断发生在signal之后或同一时刻进行自我中断 
	if (interruptMode != 0)
		reportInterruptAfterWait(interruptMode);
}

//将当前线程封装成条件等待节点，并加入到条件等待队列的队尾
//在加入队尾之前如果发现原来的尾节点不是Condition状态要对队列中所有不是Condition状态的节点进行清理。
private Node addConditionWaiter() {
	Node t = lastWaiter;
	// Node的节点状态如果不为CONDITION，则表示该节点不处于等待状态，需要清除节点
	if (t != null && t.waitStatus != Node.CONDITION) {
		unlinkCancelledWaiters();//从队列头部至尾部，清除条件队列中所有状态不为Condition的节点
		t = lastWaiter;
	}
        //将当前线程封装成新节点，状态CONDITION，并加入到条件队列的最后
	Node node = new Node(Thread.currentThread(), Node.CONDITION);
	if (t == null)
		firstWaiter = node;
	else
		t.nextWaiter = node;
	lastWaiter = node;
	return node;
}

//彻底释放已经占用的同步资源，返回原本占有的同步资源个数，
//注意这里其实最终调用的独占模式下的同步资源释放顶层模板方法，释放成功之后会唤醒队列最前面的那一个节点状态<=0的线程
//释放失败则将节点状态置为CANCELLED，以便被移除条件等待队列，并最终抛出运行时异常IllegalMonitorStateException
//什么时候会失败呢？？要么是获取的同步资源的数量发生了变化，要么是当前线程根本没有占用锁。我觉得失败的主要场景应该还是没有正确调用await()方法导致，例如：await()方法必须在获取到锁之后才能调用。
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) {
        //如果节点状态为CONDITION，那么节点一定就存在于条件队列，因为该状态只会在条件队列中出现
        //如果节点的前驱为空，那么一定不存在于同步等待队列，因为加入同步队列之后一定存在一个前驱(即使空队列也会创建一个空的标识节点即作为头节点也作为尾节点)。
	if (node.waitStatus == Node.CONDITION || node.prev == null)
		return false;
	if (node.next != null) //如果状态不是CONDITION，前驱也存在，后继也存在，那么就可以判定一定存在于同步队列，因为同步队列存在一个空标识的尾节点
		return true;
        //这时候还不能判定是否存在于同步队列的原因是，在并发情况下将节点加入队尾的CAS操作可能会失败数次才成功(因为尾节点可能已经被其他线程改变)
        //所以最终我们只能老老实实的从同步队列的队尾开始向前搜索，看看是否存在一个节点和该节点相等，找到了就说明确实已经存在于同步队列
	return findNodeFromTail(node);
}

//这个方法很简单，检测是否被中断过，如果没有被中断过，则直接返回0.
//如果被中断过，则根据情况要么返回常量THROW_IE，要么返回常量REINTERRUPT，这两个常量都不为0.
//注意，interrupted方法会清空中断状态
private int checkInterruptWhileWaiting(Node node) {
	return Thread.interrupted() ?
		(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
		0;
}

//这个方法用于完成线程被中断之后的后续工作
final boolean transferAfterCancelledWait(Node node) {
        //这里如果CAS操作成功，表示节点状态还是CONDITION，也就说明发生中断时，没有线程调用signal方法，或者中断发生在signal方法执行之前
        //因为signal方式唤醒首先就会将状态置为0，
        //这种情况说明发生了中断，需要将节点加入同步队列队尾(因为await方法必须在获得同步资源之后才能返回)，enq方法在前一章已经分析过。
        //这种情况也就需要立即抛出中断异常给直接调用者
	if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
		enq(node);
		return true;
	}
        //如果上面的CAS操作失败，则说明发生了signal方法的调用，但是我们还需要等待节点被加入同步等待队列
        //这种情况也表明，在执行signal方法之后或者同时发生了中断
        //这种情况，返回false，也就是不需要立即抛出中断异常给直接调用者，而是交由上层调用者来自行处理中断异常
	while (!isOnSyncQueue(node))
		Thread.yield();
	return false;
}

//根据不同的interruptMode进行不同方式的的中断异常处理
//实际上就是，为THROW_IE时要立即抛出中断异常，
//为REINTERRUPT时，就进行自我中断：Thread.currentThread().interrupt()
private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {
	if (interruptMode == THROW_IE) 
		throw new InterruptedException();
	else if (interruptMode == REINTERRUPT)
		selfInterrupt();
}
   根据上面await()方法的源码分析，我们可以简略的总结await()方法执行的流程如下：
  1. 执行addConditionWaiter()将当前线程构造成CONDITION状态的节点，并加入到条件等待队列队尾，有必要时会对当前条件队列中不是CONDITION状态的节点进行清理。 
  2. 执行fullyRelease()方法通过独占式同步资源释放的顶层模板方法release()彻底释放掉当前线程占用的同步资源，如果存在等候的后继节点，就唤醒后继
  3. 通过while循环判断当前节点是否已经被转移到同步等待队列，如果已经被转移到同步等待队列(即signal被调用)或者发生了中断，则退出while循环，否则将当前线程阻塞起来，等待唤醒之后继续进行中断判断和while条件判断。
  4. 退出while循环之后，执行独占式获取同步资源的方法acquireQueued()尝试获取同步资源，直到成功之后才返回。
  5. acquireQueued()获取到和之前第2步释放的等量的同步资源之后，如果有新节点被加入到条件等待队列，则清理一下条件队列中不是CONDITION状态的节点。
  6. 最后对await()阻塞期间发生的中断进行处理，如果中断发生在signal之前要立即抛出中断异常，否则就进行自我中断。

值得注意的是，await()方法确实是立即响应中断的， 但是不论怎样都需要在获取到同步资源之后，await()方法才会返回。另外await()内部是走的独占锁的同步资源获取/释放逻辑，所以，其只能用于独占模式。

3.2  void awaitUninterruptibly()

3.1小节详细分析了await()方法的源码，我们开始分析另外几个await()方法的重载方法，awaitUninterruptibly()方法是忽略中断的条件等待，直到发生了signal唤醒，唤醒之后也一定要成功获取到同步资源之后才会返回。

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;
	}
	if (acquireQueued(node, savedState) || interrupted)
		selfInterrupt();
}

   有了前面对await()基础方法的分析，要理解这个方法就很简单了。其大致流程如下：

  1. 执行addConditionWaiter()将当前线程构造成CONDITION状态的节点，并加入到条件等待队列队尾，有必要时会对当前条件队列中不是CONDITION状态的节点进行清理。 
  2. 执行fullyRelease()方法通过独占式同步资源释放的顶层模板方法release()彻底释放掉当前线程占用的同步资源，如果存在等候的后继节点，就唤醒后继
  3. 通过while循环判断当前节点是否已经被转移到同步等待队列，这里的意思就是必须要有其他线程执行了signal方法才会退出while循环，就算中断发生了也不会响应，只是将中断标记保留。
  4. 其他线程执行了signal方法唤醒当前线程之后，执行独占式获取同步资源的方法acquireQueued()尝试获取同步资源，直到成功之后才返回。
  5. 如果在之前阻塞的过程中或者在尝试获取同步资源的过程中发生了中断，都不直接抛出中断异常，而是重新进行自我中断。

因此，我们可以得出awaitUninterruptibly()和await()方法的区别在于：1) 只有在等到了signal唤醒信号才会退出阻塞，在阻塞期间忽略了中断的影响。2)在成功再次获取到同步资源之后，一定不会抛出中断异常，如果在这之前确实发生了中断，就进行自我中断。

3.3  long awaitNanos(long nanosTimeout)

awaitNanos()方法其实是对await()方法的进一步的增强，它除了响应signal与中断外，还有超时控制 。即如果当前线程没有在指定的时间内收到signal唤醒或者发生中断，则自动唤醒之后将线程节点转移到同步等待队列，并尝试获取同步资源。只有在成功获取到同步资源之后才会走和await()方法相同的中断异常处理逻辑，要么直接抛出异常，要么进行自我中断。

 

其源码就和await()基本一致，主要是使用带有相对超时时间控制的LockSupport.parkNanos()方法来进行阻塞。该方法在没有抛出异常的情况下，其返回值是一个long类型的数值，该值表示当awaitNanos()方法返回时，距离事先传递的超时时间nanosTimeout还有多久，如果返回值 <= 0 ,则可以认定它已经超时了。其单位也是纳秒。

3.4  boolean awaitUntil(Date deadline) throws InterruptedException

awaitUntil()方法其实也是对await()方法的另一种形式的增强，与awaitNanos()增强的方式不同，awaitNanos方法传递的超时时间是一个相对时长，而awaitUntil()方法传递的是一个绝对的时间点，所以awaitUntil()也是支持中断响应的。即如果当前线程没有在指定的时间点到达之前收到signal唤醒或者发生中断，则自动唤醒之后将线程节点转移到同步等待队列，并尝试获取同步资源。只有在成功获取到同步资源之后才会走和await()方法相同的中断异常处理逻辑，要么直接抛出异常，要么进行自我中断。

 

其源码就和awaitNanos()基本一致，区别是使用带有绝对超时时间控制的LockSupport.parkUntil()方法来进行阻塞。该方法在没有抛出异常的情况下，其返回值是一个布尔值，如果在没有到达指定的时间之前线程就从parkUntil()方法的阻塞点被唤醒(可能是发生了signal唤醒，也可能是发送了中断)，返回true，否则返回false.

3.5  boolean await(long time, TimeUnit unit) throws InterruptedException

该方法也是最后一个await()方法的变种，其也可以看做是对awaitNanos()方法的增强，它可以通过第二个参数unit指定第一个超时时间参数time的单位，所以该超时时间可以有很多种单位，从而表示的超时时长也就各不相同。

 

其源码几乎是awaitNanos()和awaitUntil()方法的结合，因为该方法最终还是将参数time结合单位unit转换成纳秒单位的nanosTimeout再使用 LockSupport.parkNanos()方法来实现阻塞的。而其返回值却表达了和awaitUntil()方法相同的意义：如果在没有到达指定的时间之前线程就从parkUntil()方法的阻塞点被唤醒(可能是发生了signal唤醒，也可能是发送了中断)，返回true，否则返回false.

 

四、AQS内部类ConditionObject中signal相关源码分析

在上面的第三节，我们对await()以及其相关的变种方法进行了源码分析，现在我们开始对signal()以及signalAll()进行分析。

4.1  void signal()

唤醒条件等待队列中排在队列头部的一个线程。

public final void signal() {
	if (!isHeldExclusively()) //当前线程必须是持有独占锁的线程
		throw new IllegalMonitorStateException();
	Node first = firstWaiter;
	if (first != null) //取出条件等待队列的头节点开始处理
		doSignal(first);
}

private void doSignal(Node first) {
	do {    //将头节点从条件等待队列中移除
		if ( (firstWaiter = first.nextWaiter) == null)
			lastWaiter = null;
		first.nextWaiter = null;
	} while (!transferForSignal(first) && //如果没有成功将节点转移到同步等待队列则，继续转移新的头节点，直到成功或者条件等待队列为空
			 (first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
        //首先直接将节点状态从CONDITION状态置为0，如果CAS失败表明该节点已经被取消了，则直接返回，继续操作新的头节点
	if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
		return false;
        //调用enq方法将当前节点加入到同步等待队列的队尾，返回的是加入之后的前驱节点
	Node p = enq(node);
	int ws = p.waitStatus;
        //如果前驱节点已经取消了，或者前驱节点状态<=0,但是修改状态失败，则立即唤醒当前头节点代表的线程。
        //注意这里如果前驱节点的状态<=0并且成功修改其状态为SIGNAL，表明后继节点的唤醒可以交给前驱通过独占方式的正常唤醒后继节点的逻辑唤醒，我们就不需要多此一举再唤醒了
	if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
		LockSupport.unpark(node.thread);
	return true;
}

    通过以上对signal()方法的源码分析，我们可以总结其流程如下：

  1. 首先判断当前调用signal()方法的线程是否正在独占共享资源，如果不是立即抛出IllegalMonitorStateException异常。
  2. 取走条件等待队列的头节点，通过CAS将节点的状态由CONDITION重置为0,如果失败表明该节点已经被取消了，则取出新的头节点重复进行状态修改。
  3. 如果第二步状态修改成功，就将节点转移到同步等待队列，如果转移之后的前驱节点被取消了或者企图修改前驱的状态为SIGNAL以期望它通知该节点失败时就立即唤醒该节点对应的线程。
  4. 否则如果前驱节点的状态<=0并且修改前驱的状态为SIGNAL成功，就可以将唤醒当前线程的工作交给前驱节点来完成，这里就不需要多此一举来唤醒节点对应的线程。
  5. 在成功将一个排在最前面的正常等待的节点从条件等待队列转移到同步队列或者条件等待队列为空之后，signal()方法就返回了。

值得注意的是，这里面有一个方法 isHeldExclusively()需要在实现特定同步器时被覆写，这也是在上一章提到的，当用到Condition的时候，必须要覆写isHeldExclusively()方法。另外，signal()也是遵循着先进先出的原则，首先转移的是条件等待队列中排在最前面的节点，所以条件等待队列也是一个FIFO的队列。

4.2  void signalAll()

唤醒条件等待队列中的所有线程。

 

public final void signalAll() {
	if (!isHeldExclusively()) //当前线程必须是持有独占锁的线程
		throw new IllegalMonitorStateException();
	Node first = firstWaiter;
	if (first != null)  //取出条件等待队列的头节点开始处理
		doSignalAll(first);
}

private void doSignalAll(Node first) {
	lastWaiter = firstWaiter = null; //将ConditionObject实例的两个表示头尾节点的成员变量置为空，帮助GC回收
	do {
		Node next = first.nextWaiter;
		first.nextWaiter = null;
		transferForSignal(first);
		first = next;
	} while (first != null); //从条件等待队列的头节点开始依次一个一个的转移到同步等待队列。
}
   通过以上对signalAll方法源码的分析，可以看到，其逻辑非常简单，就是从条件等待队列的头节点开始，依次调用transferForSignal()方法，尝试将其转移到同步等待队列，直到条件等待队列为空。

 

transferForSignal()方法在4.1 节的signal()方法中已经详细分析过了，它会忽略已经取消的节点，并且如果必要时会立即唤醒节点对应的线程，而不是只单单转移到同步队列等待被其前驱节点来唤醒。

 

五、Condition应用

通过上面的源码分析，我们已经对条件等待与唤醒有了了解，现在我们用一个示例来展示我们怎么使用Condition来实现线程之间的通信。

最典型的一个条件等待与唤醒的例子就是生产者/消费者模式。假定有一个有界的缓冲区，它支持put和get方法。如果试图在空的缓冲区上执行get操作，则在缓冲区不为空之前，线程将一直阻塞；如果试图在满的缓冲区上执行put操作，则在有空间变得可用之前，线程将一直阻塞。我们可以创建两个Codition实例对应两个不同的条件等待队列，这样就可以在缓冲区中的项或空间变得可用时利用最佳规划，一次只通知一个线程。 

1. 首先定义一个外部接口Mutex

public interface Mutex {

	public void lock();  
    
    public void unlock(); 
    
    public Condition newCondition();
}
2.  通过私有静态内部类实现同步组件的接口实例MutexImpl 
public class MutexImpl implements Mutex{
	
	private Sync sync = new Sync();  

	@Override
	public void lock() {
		sync.acquire(1);
	}

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

	@Override
	public Condition newCondition() {
		return sync.newCondition();
	}
	
	private static class Sync extends AbstractQueuedSynchronizer{
		
		private static final long serialVersionUID = 1251521894988052053L;

		@Override  
        protected boolean tryAcquire(int arg) {  
            return compareAndSetState(0,1); 
        }  
   
        @Override  
        protected boolean tryRelease(int arg) {  
            return compareAndSetState(1,0);
        }
        
        final ConditionObject newCondition() {
            return new ConditionObject();
        }

		@Override
		protected boolean isHeldExclusively() {
			return getState() == 1;
		}  
		
	}

}
3.  测试类ConditionTest 
public class ConditionTest {

	private LinkedList<String> buffer;

	private int maxSize;

	private Mutex lock;

	private Condition notFull;

	private Condition notEmpty;

	ConditionTest(int maxSize) {
		this.maxSize = maxSize;
		buffer = new LinkedList<String>();
		lock = new MutexImpl();
		notFull = lock.newCondition();
		notEmpty = lock.newCondition();
	}
        //生产
	public void put(String string) throws InterruptedException {
		lock.lock();
		try {
			while (maxSize == buffer.size())
				notFull.await(); //缓冲区满了，等待，直到 not full 才能继续生产
			buffer.add(string);
			notEmpty.signal();  // 生产成功，队列已经 not empty 了，发个通知出去
		} finally {
			lock.unlock();
		}
	}
	//消费
	public String get() throws InterruptedException {
        String string;
        lock.lock();
        try {
            while (buffer.size() == 0){
            	notEmpty.await();  // 队列为空，等待，直到队列 not empty，才能继续消费
            }
            string = buffer.poll();
            notFull.signal(); // 被我消费掉一个，队列 not full 了，发个通知出去
        } finally {
            lock.unlock();
        }
        return string;
    }
	
	public static void main(String[] args) {
		final ConditionTest test = new ConditionTest(5);
		
		new Thread(){
			
			public void run(){
				try {
					while (!Thread.interrupted()) {
						String string = test.get();
						System.out.println("Geted:"+string);
					}
				} catch (Exception e) {
					System.out.println("get Interrupted....");
				}
			}
			
		}.start();
		
		new Thread(){
			
			public void run(){
				for(int i=0;i < 20;i++){
					try {
						test.put(String.valueOf(i));
						System.out.println("Puted:"+String.valueOf(i));
					} catch (InterruptedException e) {
						System.out.println("put Interrupted....");
					}
				}
			}
		}.start();
		
		
	}
}
   在上面的生产者/消费者模型中，缓冲区只有5个，生产者线程想要一次性放入20个，而消费者线程一次只能消费一个，所以通过使用两个Condition实例实现了在队列已满和队列为空时精确的阻塞/唤醒生产者/消费者线程。

 

 

  • 大小: 26.8 KB
分享到：
| Java并发包核心框架AQS之一同步阻塞与唤醒 ...
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics