（10）java5条件阻塞Condition的应用<包含阻塞队列知识>

java.util.concurrent.locks
接口 Condition
所有已知实现类：
AbstractQueuedLongSynchronizer.ConditionObject, AbstractQueuedSynchronizer.ConditionObject



public interface Condition


Condition 将 Object 监视器方法（wait、notify 和 notifyAll）分解成截然不同的对象，


以便通过将这些对象与任意 Lock 实现组合使用，为每个对象提供多个等待 set（wait-set）。


其中，Lock 替代了 synchronized 方法和语句的使用，Condition 替代了 Object 监视器方法的使用。


条件（也称为条件队列 或条件变量）为线程提供了一个含义，


以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”）。


因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。


等待提供一个条件的主要属性是：以原子方式 释放相关的锁，并挂起当前线程，就像 Object.wait 做的那样。


Condition 实例实质上被绑定到一个锁上。要为特定 Lock 实例获得 Condition 实例，请使用其 newCondition() 方法。




作为一个示例，假定有一个绑定的缓冲区，它支持 put 和 take 方法。


如果试图在空的缓冲区上执行 take 操作，则在某一个项变得可用之前，线程将一直阻塞；


如果试图在满的缓冲区上执行 put 操作，则在有空间变得可用之前，线程将一直阻塞。


我们喜欢在单独的等待 set 中保存 put 线程和 take 线程，这样就可以在缓冲区中的项或空间变得可用时利用最佳规划，一次只通知一个线程。

可以使用两个 Condition 实例来做到这一点。

(阻塞队列)

<span style="font-size: 18px;"> class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock();
     try {
       while (count == items.length) 
         notFull.await();
       items[putptr] = x; 
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally {
       lock.unlock();
     }
   }</span>

<span style="font-size: 18px;"> public Object take() throws InterruptedException {
     lock.lock();
     try {
       while (count == 0) 
         notEmpty.await();
       Object x = items[takeptr]; 
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally {
       lock.unlock();
     }
   } 
 }</span>

【在实际当中 直接用ArrayBlockingQueue 类 就可以了，就不用在手动的写 上面的阻塞队列了】：如下：

java.util.concurrent。。。。---------->>>>>
接口 BlockingQueue<E>

类型参数：

E - 在此 collection 中保持的元素类型

所有超级接口：

Collection<E>, Iterable<E>, Queue<E>

所有已知子接口：

BlockingDeque<E>

所有已知实现类：

ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque,LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue

---

public interface BlockingQueue<E>

extends Queue<E>

支持两个附加操作的 Queue，这两个操作是：获取元素时等待队列变为非空，以及存储元素时等待空间变得可用。

BlockingQueue 方法以四种形式出现，对于不能立即满足但可能在将来某一时刻可以满足的操作，这四种形式的处理方式不同：第一种是抛出一个异常，第二种是返回一个特殊值（null 或false，具体取决于操作），第三种是在操作可以成功前，无限期地阻塞当前线程，第四种是在放弃前只在给定的最大时间限制内阻塞。下表中总结了这些方法：

	抛出异常	特殊值	阻塞	超时
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
检查	element()	peek()	不可用	不可用

BlockingQueue 不接受null 元素。试图 add、put 或 offer 一个 null 元素时，某些实现会抛出NullPointerException。null 被用作指示 poll 操作失败的警戒值。

【1】BlockingQueue 可以是限定容量的。它在任意给定时间都可以有一个 remainingCapacity，超出此容量，便无法无阻塞地 put 附加元素。没有任何内部容量约束的 BlockingQueue 总是报告Integer.MAX_VALUE 的剩余容量。

【2】BlockingQueue 实现主要用于生产者-使用者队列，但它另外还支持 Collection 接口。因此，举例来说，使用 remove(x) 从队列中移除任意一个元素是有可能的。然而，这种操作通常不 会有效执行，只能有计划地偶尔使用，比如在取消排队信息时。

【3】BlockingQueue 实现是线程安全的。所有排队方法都可以使用内部锁或其他形式的并发控制来自动达到它们的目的。然而，大量的 Collection 操作（addAll、containsAll、retainAll 和removeAll）没有 必要自动执行，除非在实现中特别说明。因此，举例来说，在只添加了 c 中的一些元素后，addAll(c) 有可能失败（抛出一个异常）。

【4】BlockingQueue 实质上不 支持使用任何一种“close”或“shutdown”操作来指示不再添加任何项。这种功能的需求和使用有依赖于实现的倾向。例如，一种常用的策略是：对于生产者，插入特殊的end-of-stream 或 poison 对象，并根据使用者获取这些对象的时间来对它们进行解释。

以下是基于典型的生产者-使用者场景的一个用例。注意，BlockingQueue 可以安全地与多个生产者和多个使用者一起使用。

<span style="font-size: 18px;"><strong> class Producer implements Runnable {
   private final BlockingQueue queue;
   Producer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while(true) { queue.put(produce()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   Object produce() { ... }
 }

 class Consumer implements Runnable {
   private final BlockingQueue queue;
   Consumer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while(true) { consume(queue.take()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   void consume(Object x) { ... }
 }

 class Setup {
   void main() {
     BlockingQueue q = new SomeQueueImplementation();
     Producer p = new Producer(q);
     Consumer c1 = new Consumer(q);
     Consumer c2 = new Consumer(q);
     new Thread(p).start();
     new Thread(c1).start();
     new Thread(c2).start();
   }
 }</strong></span>
 

内存一致性效果：当存在其他并发 collection 时，将对象放入 BlockingQueue 之前的线程中的操作happen-before 随后通过另一线程从 BlockingQueue 中访问或移除该元素的操作。

此接口是 Java Collections Framework 的成员。

。。。。结束。

<<<<<--------

（ArrayBlockingQueue 类提供了这项功能，因此没有理由去实现这个示例类。） 可参见：JDK文档。

Condition 实现可以提供不同于 Object 监视器方法的行为和语义，比如受保证的通知排序，或者在执行通知时不需要保持一个锁。


如果某个实现提供了这样特殊的语义，则该实现必须记录这些语义。


注意，Condition 实例只是一些普通的对象，它们自身可以用作 synchronized 语句中的目标，并且可以调用自己的 wait 和 notification 监视器方法。


获取 Condition 实例的监视器锁或者使用其监视器方法，与获取和该 Condition 相关的 Lock 或使用其 waiting 和 signalling 方法没有什么特定的关系。


为了避免混淆，建议除了在其自身的实现中之外，切勿以这种方式使用 Condition 实例。


除非另行说明，否则为任何参数传递 null 值将导致抛出 NullPointerException。




实现注意事项:
在等待 Condition 时，允许发生“虚假唤醒”，这通常作为对基础平台语义的让步。


对于大多数应用程序，这带来的实际影响很小，因为 Condition 应该总是在一个循环中被等待，并测试正被等待的状态声明。


某个实现可以随意移除可能的虚假唤醒，但建议应用程序程序员总是假定这些虚假唤醒可能发生，因此总是在一个循环中等待。


三种形式的条件等待（可中断、不可中断和超时）在一些平台上的实现以及它们的性能特征可能会有所不同。


尤其是它可能很难提供这些特性和维护特定语义，比如排序保证。更进一步地说，中断线程实际挂起的能力在所有平台上并不是总是可行的。


因此，并不要求某个实现为所有三种形式的等待定义完全相同的保证或语义，也不要求其支持中断线程的实际挂起。


要求实现清楚地记录每个等待方法提供的语义和保证，在某个实现不支持中断线程的挂起时，它必须遵从此接口中定义的中断语义。


由于中断通常意味着取消，而又通常很少进行中断检查，因此实现可以先于普通方法的返回来对中断进行响应。


即使出现在另一个操作后的中断可能会释放线程锁时也是如此。实现应记录此行为。






从以下版本开始：
1.5






方法摘要
void await()
造成当前线程在接到信号或被中断之前一直处于等待状态。
boolean await(long time, TimeUnit unit)
造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
long awaitNanos(long nanosTimeout)
造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
void awaitUninterruptibly()
造成当前线程在接到信号之前一直处于等待状态。
boolean awaitUntil(Date deadline)
造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。
void signal()
唤醒一个等待线程。
void signalAll()
唤醒所有等待线程。

应用实例：

package com.itm.thread;

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

/*********
 * 
 * 子线程循环10次，接着主线程循环100次，接着又回到子线程循环10次，接着再回到主线程循环100次，如此循环50次，请写出程序
 * 
 * 
 * @author
 * 
 */
public class ConditionCommunication {

	/**
	 * @param args
	 */
	public static void main(String[] args) {

		final Business business = new Business();

		new Thread(new Runnable() {
			@Override
			public void run() {
				for (int i = 1; i <= 50; i++) {
					business.sub(i);
				}
			}
		}).start();

		// 主线程。
		for (int i = 1; i <= 50; i++) {

			business.main(i);
		}

	}

	static class Business {

		Lock lock = new ReentrantLock();
		Condition condition = lock.newCondition();

		private boolean bShouldSub = true;

		public void sub(int i) {
			lock.lock();
			try {
				while (!bShouldSub) {
					try {
						condition.await();
						// int x = 3/1; // 这里必须写成一句话。。。
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				// 如果 变量就等于 不ShouldSub，则不等待 就直接来了。
				for (int j = 1; j <= 10; j++) {
					System.out.println("sub thread sequece of " + j
							+ ", loop of " + i);
				}

				bShouldSub = false;
				condition.signal();
				// this.notify();
			} finally {
				lock.unlock();
			}
		}

		public void main(int i) {

			lock.lock();
			try {

				while (bShouldSub) {
					try {
						condition.await();
						// this.wait();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				for (int j = 1; j <= 100; j++) {
					System.out.println("main thread sequece of " + j
							+ ", loop of " + i);
				}

				bShouldSub = true;
				condition.signal();
				// this.notify();

			} finally {
				lock.unlock();
			}
		}
	}

}

await
void await()
throws InterruptedException造成当前线程在接到信号或被中断之前一直处于等待状态。
与此 Condition 相关的锁以原子方式释放，并且出于线程调度的目的，将禁用当前线程，且在发生以下四种情况之一 以前，当前线程将一直处于休眠状态：


其他某个线程调用此 Condition 的 signal() 方法，并且碰巧将当前线程选为被唤醒的线程；或者
其他某个线程调用此 Condition 的 signalAll() 方法；或者
其他某个线程中断当前线程，且支持中断线程的挂起；或者
发生“虚假唤醒”
在所有情况下，在此方法可以返回当前线程之前，都必须重新获取与此条件有关的锁。在线程返回时，可以保证 它保持此锁。


如果当前线程：


在进入此方法时已经设置了该线程的中断状态；或者
在支持等待和中断线程挂起时，线程被中断，
则抛出 InterruptedException，并清除当前线程的中断状态。在第一种情况下，没有指定是否在释放锁之前发生中断测试。
实现注意事项


假定调用此方法时，当前线程保持了与此 Condition 有关联的锁。这取决于确定是否为这种情况以及不是时，如何对此作出响应的实现。通常，将抛出一个异常（比如 IllegalMonitorStateException）并且该实现必须对此进行记录。


与响应某个信号而返回的普通方法相比，实现可能更喜欢响应某个中断。在这种情况下，实现必须确保信号被重定向到另一个等待线程（如果有的话）。




抛出：
InterruptedException - 如果当前线程被中断（并且支持中断线程挂起）

新需求：第一个循环100次，第二个循环10，第三个循环20次。

package com.itm.thread;

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

/*********
 * 
 *   
 * 
 * 
 * @author
 * 
 */
public class ThreeConditionCommunication {

	/**
	 * @param args
	 */
	public static void main(String[] args) {

		final Business business = new Business();

		new Thread(new Runnable() {
			@Override
			public void run() {
				for (int i = 1; i <= 50; i++) {
					business.sub2(i);
				}
			}
		}).start();

		new Thread(new Runnable() {
			@Override
			public void run() {
				for (int i = 1; i <= 50; i++) {
					business.sub3(i);
				}
			}
		}).start();

		// 主线程。
		for (int i = 1; i <= 50; i++) {

			business.main(i);
		}

	}

	static class Business {

		Lock lock = new ReentrantLock();
		Condition condition1 = lock.newCondition();
		Condition condition2 = lock.newCondition();
		Condition condition3 = lock.newCondition();

		private int shouldSub = 1;

		public void sub2(int i) {
			lock.lock();
			try {
				while (shouldSub != 2) {
					try {
						condition2.await();
						// int x = 3/1; // 这里必须写成一句话。。。
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				
				for (int j = 1; j <= 10; j++) {
					System.out.println("sub2 thread sequece of " + j
							+ ", loop of " + i);
				}

				shouldSub = 3;
				condition3.signal();
				// this.notify();
			} finally {
				lock.unlock();
			}
		}

		public void sub3(int i) {
			lock.lock();
			try {
				while (shouldSub != 3) {
					try {
						condition3.await();
						// int x = 3/1; // 这里必须写成一句话。。。
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				
				for (int j = 1; j <= 20; j++) {
					System.out.println("sub3 thread sequece of " + j
							+ ", loop of " + i);
				}

				shouldSub = 1;
				condition1.signal();
				// this.notify();
			} finally {
				lock.unlock();
			}
		}

		public void main(int i) {

			lock.lock();
			try {

				while (shouldSub != 1) {
					try {
						condition1.await();
						// this.wait();
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				for (int j = 1; j <= 100; j++) {
					System.out.println("main thread sequece of " + j
							+ ", loop of " + i);
				}

				shouldSub = 2;
				condition2.signal();
				// this.notify();

			} finally {
				lock.unlock();
			}
		}
	}

}