java多线程之线程并发库条件阻塞Condition的应用

锁(Lock/synchronized)只能实现互斥不能实现通信，Condition的功能类似于在传统的线程技术中的,Object.wait()和Object.notify()的功能,在等待Condition时,允许发生"虚假唤醒",这通常作为对基础平台语义的让步,对于大多数应用程序,这带来的实际影响很小,因为Condition应该总是在一个循环中被等待,并测试正被等待的状态声明.某个实现可以随意移除可能的虚假唤醒,但是建议程序员总是假定这些虚假唤醒可能发生,因此总是在一个循环中等待.

一个锁内部可以有多个Condition,即有多路等待和通知,可以参看jdk1.5提供的Lock与Condition实现的可阻塞队列的应用案例,从中要体味算法,还要体味面向对象的封装.在传统的线程机制中,一个监视器对象上只能有一路等待和通知,要想实现多路等待和通知,必须嵌套使用多个同步监视器对象.(如果只用一个Condition,两个放的都在等,一旦一个放进去了,那么它会通知可能会导致另一个放的接着往下走).

我们也可以通过Lock和Condition来实现上面我们讲的例子:

子线程循环10次，接着主线程循环100，接着又回到子线程循环10次，接着再回到主线程又循环100，如此循环50次，请写出程序。

package javaplay.thread.test;

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

public class ConditionCommunication {
public static void main(String[] args) {
final Business business = new Business();
new Thread(new Runnable() {

@Override
public void run() {
for (int i = 0; i < 5; i++) {
business.sub(i);
}

}

}).start();

for (int i = 0; i < 5; i++) {
business.main(i);// 此处不能是sub 否则死锁
}

}

static class Business {
// 使用lock与原来的synchronized几乎一模一样的
// 思想和套路是一样的 不在乎用得是什么新技术
// 所以程序员如果崇拜哪种语言或者技术都是很可笑的一件事
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
private boolean bShouldSub = true;// volatile???

public void sub(int i) {
lock.lock();
try {
while (!bShouldSub) {
try {
// this.wait();
condition.await();// 不是wait wait是任何对象都有的，继承自Object
} catch (Exception e) {
e.printStackTrace();
}
}
for (int j = 0; j < 1; j++) {
System.out.println("sub thread sequence of " + j + " ,loop of " + i);
}
bShouldSub = false;
// this.notify();
condition.signal();// 同样不能使用notify
} finally {
lock.unlock();
}
}

public void main(int i) {
lock.lock();
try {
while (bShouldSub) {
try {
// this.wait();
condition.await();
} catch (Exception e) {
e.printStackTrace();
}
}
for (int j = 0; j < 2; j++) {
System.out.println("main thread sequence of " + j + " ,loop of " + i);
}
bShouldSub = true;
condition.signal();
// this.notify();
// wait和notify必须在synchronized块中，synchronized加在谁身上，就调用谁的wait/notify
} finally {
lock.unlock();
}

}
}

}

可以使用Lock和Condition来实现一个缓冲队列(要区别缓冲和缓存的区别), 其实condition doc有这样的例子:

package javaplay.thread.test;

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

/*
* A Condition instance is intrinsically bound to a lock.
* To obtain a Condition instance for a particular Lock instance use
* its newCondition() method.

* As an example, suppose we have a bounded buffer which supports put and take methods.
* If a take is attempted on an empty buffer, then the thread will block until
* an item becomes available; if a put is attempted on a full buffer,
* then the thread will block until a space becomes available.
* We would like to keep waiting put threads and take threads in separate wait-sets
* so that we can use the optimization of only notifying a single thread at a time
* when items or spaces become available in the buffer. This can be achieved using
* two Condition instances.
* 可阻塞队列如下：
*/
class BoundedBuffer {
final Lock lock = new ReentrantLock();
// 此处不能用full/empty 这两种状态不能代表队列所有状态
// 但notFull/notEmpty却可以代表取放过程中队列所有的状态
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
// 此处不能使用一个condition完成 比如说有五个线程把队列放满了
// 此时有一个取线程取走了一个 唤醒了其中一个放线程 则有一个放线程被执行
// 执行完又要唤醒一个放线程 这个线程在执行时队列已经满了 不能再放了
// 有两个condition就能区分是唤醒取线程还是放线程

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();
}
}

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();
}
}
}
// (The ArrayBlockingQueue class provides this functionality,
// so there is no reason to implement this sample usage class.)

/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
据此,我们可以改变“子线程循环10次，接着主线程循环100，接着又回到子线程循环10次，接着再回到主线程又循环100，如此循环50次，请写出程序”的例子,这个例子是两个线程之间的跳转,那么如果实现三个线程之间的轮循,比如:线程1循环10,线程2循环100,线程3循环20次,然后又是线程1,接着线程2...一直轮循50次.

package javaplay.thread.test;

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

public class ThreeConditionCommunication {
public static void main(String[] args) {
final Business business = new Business();
new Thread(new Runnable() {

@Override
public void run() {
for (int i = 0; i < 5; i++) {
business.sub2(i);
}

}

}).start();
new Thread(new Runnable() {

@Override
public void run() {
for (int i = 0; i < 5; i++) {
business.sub3(i);
}

}

}).start();

for (int i = 0; i < 5; 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 bShouldSub = 1;

public void sub2(int i) {
lock.lock();
try {
while (bShouldSub != 2) {
try {
condition2.await();
} catch (Exception e) {
e.printStackTrace();
}
}
for (int j = 0; j < 1; j++) {
System.out.println("sub2 thread sequence of " + j + " ,loop of " + i);
}
bShouldSub = 3;
condition3.signal();
} finally {
lock.unlock();
}
}

public void sub3(int i) {
lock.lock();
try {
while (bShouldSub != 3) {
try {
condition3.await();
} catch (Exception e) {
e.printStackTrace();
}
}
for (int j = 0; j < 1; j++) {
System.out.println("sub3 thread sequence of " + j + " ,loop of " + i);
}
bShouldSub = 1;
condition1.signal();
} finally {
lock.unlock();
}
}

public void main(int i) {
lock.lock();
try {
while (bShouldSub != 1) {
try {
condition1.await();
} catch (Exception e) {
e.printStackTrace();
}
}
for (int j = 0; j < 1; j++) {
System.out.println("main thread sequence of " + j + " ,loop of " + i);
}
bShouldSub = 2;
condition2.signal();
} finally {
lock.unlock();
}

}
}

}
据此可假想的认为，线程在阻塞时类似人去银行取钱排队等候等待被叫号(唤醒)，排的队也不只有一个，业务不一样排的队名也不一样；
多线程争夺的锁可以是任何对象，线程阻塞也要有在哪个对象上阻塞，this.notify只是唤醒在等待this对象这把锁的线程队列中的其中一个线程，很难理解么？多写几遍！

当一个线程执行到wait()方法时，它就进入到一个和该对象相关的等待池中，同时失去（释放）了对象的机锁（暂时失去机锁，wait(long timeout)超时时间到后还需要返还对象锁）；如果能深刻理解，所谓的等待与唤醒都是要指定队名的(或者线程池名)，即在哪个对象相关的等待队列里面等待和在哪个对象相关的等待队列里唤醒线程，不能站错队；