Cooperation between tasks
As you’ve seen, when you use threads to run more than one task at a time, you can keep one task from interfering with another task’s resources by using a lock (mutex) to synchronize the behavior of the two tasks. That is, if two tasks are stepping on each other over a shared resource (usually memory), you use a mutex to allow only one task at a time to access that resource.
With that problem solved, the next step is to learn how to make tasks cooperate with each other, so that multiple tasks can work together to solve a problem. Now the issue is not about interfering with one another, but rather about working in unison, since portions of such problems must be solved before other portions can be solved. It’s much like project planning: The footings for the house must be dug first, but the steel can be laid and the concrete forms can be built in parallel, and both of those tasks must be finished before the concrete foundation can be poured. The plumbing must be in place before the concrete slab can be poured, the concrete slab must be in place before you start framing, and so on. Some of these tasks can be done in parallel, but certain steps require all tasks to be completed before you can move ahead.
The key issue when tasks are cooperating is handshaking between those tasks. To accomplish this handshaking, we use the same foundation: the mutex, which in this case guarantees that only one task can respond to a signal. This eliminates any possible race conditions. On top of the mutex, we add a way for a task to suspend itself until some external state changes (e.g., "The plumbing is now in place"), indicating that it’s time for that task to move forward. In this section, we’ll look at the issues of handshaking between tasks, which is safely implemented using the Object methods wait( ) and notifyAll( ). The Java SE5 concurrency library also provides Condition objects with await( ) and signal( ) methods. We’ll see the problems that can arise, and their solutions.
wait() and notifyAll()
wait( ) allows you to wait for a change in some condition that is outside the control of the forces in the current method. Often, this condition will be changed by another task. You don’t want to idly loop while testing the condition inside your task; this is called busy waiting, and it’s usually a bad use of CPU cycles. So wait( ) suspends the task while waiting for the world to change, and only when a notify( ) or notifyAll( ) occurs—suggesting that something of interest may have happened—does the task wake up and check for changes. Thus, wait( ) provides a way to synchronize activities between tasks.
It’s important to understand that sleep( ) does not release the object lock when it is called, and neither does yield( ). On the other hand, when a task enters a call to wait( ) inside a method, that thread’s execution is suspended, and the lock on that object is released. Because wait( ) releases the lock, it means that the lock can be acquired by another task, so other synchronized methods in the (now unlocked) object can be called during a wait( ). This is essential, because those other methods are typically what cause the change that makes it interesting for the suspended task to reawaken. Thus, when you call wait( ), you’re saying, "I’ve done all I can right now, so I’m going to wait right here, but I want to allow other synchronized operations to take place if they can."
There are two forms of wait( ). One version takes an argument in milliseconds that has the same meaning as in sleep( ): "Pause for this period of time." But unlike with sleep( ), with wait(pause):
1.
The object lock is released during the wait( ).
2.
You can also come out of the wait( ) due to a notify( ) or notifyAll( ), in addition to letting the clock run out.
The second, more commonly used form of wait( ) takes no arguments. This wait( ) continues indefinitely until the thread receives a notify( ) or notifyAll( ).
One fairly unique aspect of wait( ), notify( ), and notifyAll( ) is that these methods are part of the base class Object and not part of Thread. Although this seems a bit strange at first—to have something that’s exclusively for threading as part of the universal base class—it’s essential because these methods manipulate the lock that’s also part of every object. As a result, you can put a wait( ) inside any synchronized method, regardless of whether that class extends Thread or implements Runnable. In fact, the only place you can call wait( ), notify( ), or notifyAll( ) is within a synchronized method or block (sleep( ) can be called within non-synchronized methods since it doesn’t manipulate the lock). If you call any of these within a method that’s not synchronized, the program will compile, but when you run it, you’ll get an IllegalMonitorStateException with the somewhat nonintuitive message "current thread not owner." This message means that the task calling wait( ), notify( ), or notifyAll( ) must "own" (acquire) the lock for the object before it can call any of those methods.
You can ask another object to perform an operation that manipulates its own lock. To do this, you must first capture that object’s lock. For example, if you want to send notifyAll( ) to an object x, you must do so inside a synchronized block that acquires the lock for x:
synchronized(x) {
x.notifyAll();
}
Let’s look at a simple example. WaxOMatic.java has two processes: one to apply wax to a Car and one to polish it. The polishing task cannot do its job until the application task is finished, and the application task must wait until the polishing task is finished before it can put on another coat of wax. Both WaxOn and WaxOff use the Car object, which uses wait( ) and notifyAll( ) to suspend and restart tasks while they’re waiting for a condition to change:
Thread.sleep(), Thread.yield()不会释放锁(如果有的话),而wait()会释放锁(必须包含在synchronized的方法或块之中)。
//Car.java
package concurrency;
import java.util.concurrent.TimeUnit;
public class Car {
private volatile boolean isWaxed = false;
public synchronized void waxOn() throws InterruptedException {
isWaxed = true;
System.out.println("Car WaxOn...");
TimeUnit.MILLISECONDS.sleep(1000);
notifyAll();
}
public synchronized void waxOff() throws InterruptedException {
isWaxed = false;
System.out.println("Car WaxOff...");
TimeUnit.MILLISECONDS.sleep(100);
notifyAll();
}
public synchronized void waitForBuffing() throws InterruptedException {
while (isWaxed == true) {
wait();
}
}
public synchronized void waitForWax() throws InterruptedException {
while (isWaxed == false) {
wait();
}
}
}
//WaxOn.java
package concurrency;
public class WaxOn implements Runnable {
private Car car;
public WaxOn(Car car) {
this.car = car;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
car.waxOn();
car.waitForBuffing();
}
} catch (InterruptedException e) {
System.out.println("Wax On Exiting via interrupt");
}
System.out.println("Wax on task done");
}
}
//WaxOff.java
package concurrency;
public class WaxOff implements Runnable {
private Car car;
public WaxOff(Car car) {
this.car = car;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
car.waxOff();
car.waitForWax();
}
} catch (InterruptedException e) {
System.out.println("Wax Off Exiting via interrupt");
}
System.out.println("Wax off task done");
}
}
//WaxOMatic.java
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class WaxOMatic {
public static void main(String[] args) {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOn(car));
exec.execute(new WaxOff(car));
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
exec.shutdownNow();
System.out.println("Main exit");
}
}