并发（10）：线程之间的协作（上）

蔚落 2023-03-14 10:37 17阅读 0赞

正如你所见到的，当你使用线程来同时运行多个任务时，可以通过使用锁（互斥）来同步两个任务的行为，从而使得一个任务不会干涉另一个任务的资源。也就是说，如果两个任务在交替着步入某项共享资源（通常是内存），你可以使用互斥来使得任何时刻只有一个任务可以访问这项资源。

这个问题已经解决了，下一步学习如何使任务彼此之间可以协作，以使得多个任务可以一起工作去解决某个问题。现在的问题不是彼此之间的干涉，而是彼此之间的协调，因为在这类问题中，某些部分必须在其他部分被解决之前解决。这非常像项目规划：必须先挖房子的地基，但是接下来可以并行地铺设钢结构和构建水泥部件，而这两项任务必须在混凝土浇注之前完成。管道必须在水泥板浇注之前到位，而水泥板必须在开始构筑房屋骨架之前到位，等等。在这些任务中，某些可以并行执行，但是某些步骤需要所有的任务都结束之后才能开动。

当任务协作时，关键问题是这些任务之间的握手。为了实现这种握手，我们使用了相同的基础特性：互斥。在这种情况下，互斥能够确保只有一个任务可以响应某个信号，这样就可以根除任何可能的竞争条件。在互斥之上，我们为任务添加了一种途径，可以将其自身挂起，直指某些外部条件发生变化（例如，管道现在已经到位），表示是时候让这个任务向前开动了为止。在本节，我们将浏览任务间的握手问题，这种握手可以通过Object的方法wait()和notify()来安全地实现。java SE5的并发类库还提供了具有await()和signal()方法的Condition对象。我们将看到产生的各类问题，以及相应的解决方案。

### 一、wait()和notifyAll() ###

wait()使你可以等待某个条件发生变化，而改变这个条件超出了当前方法的控制能力。通常，这种条件将由另一个任务来改变。你肯定不想在你的任务测试这个条件的同时，不断的进行空循环，这被称为忙等待，通常是一种不良的CPU周期使用方式。因此wait()会在等待外部世界产生变化的时候将任务挂起，并且只有在notify()或notifyAll()发生时，即表示发生了某些感兴趣的事物，这个任务才会被唤醒并去检查所产生的变化。因此，wait()提供了一种在任务之间对活动同步的方式。

调用sleep()的时候锁并没有被释放，调用yield()也属于这种情况，理解这一点很重要。另一方面，当一个任务在方法里遇到了对wait()的调用的时候，线程的执行被挂起，对象上的锁被释放。因为wait()将释放锁，这就意味着另一个任务可以获得这个锁，因此在该对象（现在是未锁定的）中的其他synchronized方法可以在wait()期间被调用。这一点至关重要，因为这些其他的方法通常将会产生改变，而这种改变正式使被挂起的任务重新唤醒所感兴趣的变化。因此，当你调用wait()时，就是在声明：“我已经刚刚做完能做的所有事情，因此我要在这里等待，但是我希望其他synchronized操作在条件适合的情况下能够执行。”

有两种形式的wait()。第一种版本接受毫秒数作为参数，含义与sleep()方法里参数的意思相同，都是指“在此期间暂停”。但是与sleep()不同的是，对于wait()而言：

1.  在wait()期间对象锁是释放的。
2.  可以通过notify()、notifyAll()，或者令时间到期，从wait()中恢复执行。

第二种，也是更常用形式的wait()不接受任何参数。这种wait()将无限等待下去，直到线程接收到notify或者notifyAll()消息。

wait()、notify()以及notifyAll()有一个比较特殊的方面，那就是这些方法是基于Object的一部分，而不是属于Thread的一部分。尽管看起来有点奇怪——仅仅是针对线程的功能却作为通用基类的一部分而实现，不过这是有道理的，因为这些方法操作的锁也是所有对象的一部分。所以，你可以把wait()放进任何同步控制方法里，而不用考虑这个类是继承自Thread还是实现了Runnable接口。实际上，只能在同步控制方法或同步控制块里调用wait()、notify()和notifyAll()（因为不用操作锁，所以sleep()可以在非同步控制方法里调用）。如果在非同步控制方法里调用这些方法，程序能通过编译，但运行的时候，将得到IllegalMonitorStateException异常，并伴随着一些含糊的消息，比如“当前线程不是拥有者”。消息的意思是，调用wait()、notify()和notifyAll()的任务在调用这些方法前必须“拥有”（获取）对象的锁。

可以让另一个对象执行某种操作以维护其自己的锁。要这么做的话，必须首先得到对象的锁。比如，如果要向对象x发送notifyAll()，那么就必须在能够取得x的锁的同步控制块中这么做：

synchronized(x) {
        x.notifyAll();
    }

让我们看一个简单的示例，WaxOMatic.java有两个过程：一个是将蜡涂到Car上，一个是抛光它。抛光任务在涂蜡任务完成之前，是不能执行其工作的，而涂蜡任务在涂另一层蜡之前，必须等待抛光任务完成。WaxOn和WaxOff都使用了Car对象，该对象在这些任务等待条件变化的时候，使用wait()和notifyAll()来挂起和重新启动这些任务：

package concurrency.waxomatic;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    class Car {
    	private boolean waxOn = false;
    
    	// 上过蜡的
    	public synchronized void waxed() {
    		waxOn = true; // 准备抛光
    		notifyAll();
    	}
    
    	// 抛光
    	public synchronized void buffed() {
    		waxOn = false; // 准备再涂一层蜡
    		notifyAll();
    	}
    
    	// 等待打蜡
    	public synchronized void waitForWaxing() throws InterruptedException {
    		while (waxOn == false)
    			wait();
    	}
    
    	// 等待抛光
    	public synchronized void waitForBuffing() throws InterruptedException {
    		while (waxOn == true)
    			wait();
    	}
    }
    
    class WaxOn implements Runnable {
    
    	private Car car;
    
    	public WaxOn(Car c) {
    		car = c;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				System.out.print("Wax On! ");
    				TimeUnit.MILLISECONDS.sleep(200);
    				car.waxed();
    				car.waitForBuffing();
    			}
    		} catch (InterruptedException e) {
    			System.out.println("通过中断退出");
    		}
    		System.out.println("Ending Wax On task");
    	}
    }
    
    class WaxOff implements Runnable {
    
    	private Car car;
    
    	public WaxOff(Car c) {
    		car = c;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				car.waitForWaxing();
    				System.out.print("Wax Off!");
    				TimeUnit.MILLISECONDS.sleep(200);
    				car.buffed();
    			}
    		} catch (InterruptedException e) {
    			System.out.println("通过中断退出");
    		}
    
    		System.out.println("Ending Wax Off task");
    	}
    }
    
    public class WaxOMatic {
    	public static void main(String[] args) throws InterruptedException {
    		Car car = new Car();
    		ExecutorService exec = Executors.newCachedThreadPool();
    		exec.execute(new WaxOff(car));
    		exec.execute(new WaxOn(car));
    		TimeUnit.SECONDS.sleep(5); // Run for a while
    		exec.shutdownNow(); // Interrupt all tasks
    	}
    }

这里，Car有一个单一的布尔属性waxOn，表示涂蜡-抛光处理的状态。

在waitForWaxing()中将检查waxOn标志，如果它为false，那么这个调用任务将通过调用wait()而被挂起。这个行为发生在synchronized方法中这一点很重要，因为在这样的方法中，任务已经获得了锁。当你调用wait()时，线程被挂起，而锁被释放。所被释放这一点是本质所在，因为为了安全地改变对象的状态（例如，将waxOn改变为true，如果被挂起的任务要继续执行，就必须执行该动作），其他某个任务就必须能够获得这个锁，从而将waxOn改变为true。之后，wait()中唤醒，它必须首先重新获得当它进入wait()时释放的锁。在这个锁变得可用之前，这个任务是不会被唤醒的。

WaxOn.run()表示给汽车打蜡过程的第一个步骤，因此它将执行它的操作：调用sleep()以模拟需要涂蜡的时间，然后告知汽车涂蜡结束，并调用waitForBuffing()，这个方法会用一个wait()调用来挂起这个任务，直至WaxOff任务调用这辆车的Buffed()，从而改变状态并调用notifyAll()为止。另一方面，WaxOff.run()立即进入waitForWaxing()，并因此而被挂起，直至WaxOn涂完蜡并且waxed()被调用。在运行这个程序时，你可以看到当控制权在两个任务之间来回互相传递时，这两个步骤过程在不断地重复。在5秒钟之后，interrupt()会中止这两个线程；当你调用某个ExecutorService的shutdownNow()时，它会调用所有由它控制的线程的interrupt()。

前面的示例强调你必须用一个检查感兴趣的条件的while循环包围wait()。这很重要，因为：

*  你可能有多个任务出于相同的原因在等待同一个锁，而第一个唤醒任务可能会改变这种状况（即使你没有这么做，有人也会通过继承你的类去这么做）。如果属于这种情况，那么这个任务应该被再次挂起，直至其感兴趣的条件发生变化。
 *  在这个任务从其wait()中被唤醒的时刻，有可能会有某个其他的任务已经做出了改变，从而使得这个任务在此时不能执行，或者执行其操作已显得无关紧要。此时，应该通过再次调用wait()来将其重新挂起。
 *  也有可能某些任务出于不同的原因在等待你的对象上的锁（在这种情况下必须使用notifyAll()）。在这种情况下，你需要检查是否已经由正确的原因唤醒，如果不是，就再次调用wait()。

因此，其本质就是要检查所感兴趣的特定条件，并在条件不满足的情况下返回到wait()中。惯用的方法就是使用while来编写这种代码。

**（1）错失的信号**

当两个线程使用notify()/wait()或notifyAll()/wait()进行协作时，有可能会错失某个信号。假设T1是通知T2的线程，而这两个线程都是使用下面（有缺陷的）方式实现的：

// T1：
    synchronized(sharedMonitor) {
        <setup condition for T2>
        sharedMonitor.notify();
    }
    
    // T2：
    while(someCondition) {
        // Point 1
        synchronized(sharedMonitor) {
            sharedMonitor.wait();
        }
    }

<setup condition for T2>是防止T2调用wait()的一个动作，当然前提是T2还没有调用wait()。

假设T2对someCondition求值并发现是true。在Point1，线程调度器可能切换到了T1。而T1将执行其设置，然后调用notify()。当T2得以继续执行时，此时对T2来说，时机已经太晚了，以至于不能意识到这个条件已经发生了变化，因此会盲目进入wait()。此时notify()将错失，而T2也将无限的等待这个已经发送的信号，从而产生死锁。

该问题的解决方案是防止在someCondition变量上产生竞争条件。下面是T2正确的执行方式：

synchronized(sharedMonitor) {
        while(someCondition) {
            sharedMonitor.wait();
        }
    }

现在，如果T1首先执行，当控制返回T2时，它将发现条件发生了变化，从而不会进入wait()。反过来，如果T2首先执行，那它将进入wait()，并且稍后会由T1唤醒。因此，信号不会错失。

### 二、notify()与notifyAll() ###

因为在技术上，可能会有多个任务在单个Car对象上处于wait()状态，因此调用notifyAll()比只调用notify()要更安全。但是，上面程序的结构只会有一个任务实际处于wait()状态，因此你可以使用notify()来代替notifyAll()。

使用notify()而不是notifyAll()是一种优化。使用notify()时，在众多等待同一个锁的任务中只有一个会被唤醒，因此如果你希望使用notify()就必须保证被唤醒的是恰当的任务。另外，为了使用notify()，所有任务必须等待相同的条件，因为如果你有多个任务在等待不同的条件，那么你就不会知道是否唤醒了恰当的任务。如果使用notify()，当条件发生变化时，必须只有一个任务能够从中受益。最后，这些限制对所有可能存在的子类都必须总是起作用的。如果这些规则中有任何一条不满足，那么你就必须使用notifyAll()而不是notify()。

在有关java的线程机制的讨论中，有一个令人困惑的描述：notifyAll()将唤醒“所有正在等待的任务”。这是否意味着在程序中任何地方，任何处于wait()状态中的任务都将被任何对notifyAll()的调用唤醒呢？在下面的示例中，与Task2相关的代码说明了情况并非如此——事实上，当notifyAll()因某个特定锁而被调用时，只有等待这个锁的任务才会被唤醒：

package concurrency;
    
    import java.util.Timer;
    import java.util.TimerTask;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 阻挡者
     */
    class Blocker {
    	synchronized void waitingCall() {
    		try {
    			while (!Thread.interrupted()) {
    				wait();
    				System.out.println(Thread.currentThread() + " ");
    			}
    		} catch (InterruptedException e) {
    			// OK to exit this way
    		}
    	}
    
    	synchronized void prod() {
    		notify();
    	}
    
    	synchronized void prodAll() {
    		notifyAll();
    	}
    }
    
    class Task implements Runnable {
    
    	static Blocker blokcer = new Blocker();
    
    	@Override
    	public void run() {
    		blokcer.waitingCall();
    	}
    }
    
    class Task2 implements Runnable {
    	// A separate Blocker object:
    	static Blocker blocker = new Blocker();
    
    	@Override
    	public void run() {
    		blocker.waitingCall();
    	}
    }
    
    public class NotifyVsNotifyAll {
    	public static void main(String[] args) throws InterruptedException {
    		ExecutorService exec = Executors.newCachedThreadPool();
    		for (int i = 0; i < 5; i++) {
    			exec.execute(new Task());
    		}
    		exec.execute(new Task2());
    		Timer timer = new Timer();
    		timer.scheduleAtFixedRate(new TimerTask() {
    			boolean prod = true;
    
    			@Override
    			public void run() {
    				if (prod) {
    					System.out.print("\nnotify() ");
    					Task.blokcer.prod();
    					prod = false;
    				} else {
    					System.out.print("\nnotifyAll()");
    					Task.blokcer.prodAll();
    					prod = true;
    				}
    			}
    		}, 400, 400); // Run every 4 second
    		TimeUnit.SECONDS.sleep(5); // Run for a while
    		timer.cancel();
    		System.out.println("\nTimer canceled");
    		TimeUnit.MILLISECONDS.sleep(500);
    		System.out.print("Task2.blocker.prodAll()");
    		Task2.blocker.prodAll();
    		TimeUnit.MILLISECONDS.sleep(500);
    		System.out.println("\nShutting down");
    		exec.shutdownNow(); // Interrupt all tasks
    	}
    }

Task和Task2每个都有其自己的Blocker对象，因此每个Task对象都会在Task.blocker上阻塞，而每个Task2都会在Task2.blocker上阻塞。在main()中，java.util.Timer对象被设置为每4秒执行一次run()方法，而这个run()方法将经由“激励”方法交替地在Task.blocker上调用notify()和notifyAll()。

从输出中你可以看到，即使存在Task2.blocker上阻塞的Task2对象，也没有任何Task.blocker上的notify()或notifyAll()调用会导致Task2对象被唤醒。与此类似，在main()的结尾，调用了timer的cancel()，即使计时器被撤销了，前5个任务也依然在运行，并仍旧在它们对Task.blocker.waitingCall()的调用中被阻塞。对Task2.blocker.prodAll()的调用所产生的输出不包括任何在Task.blocker中的锁上等待的任务。

如果你浏览Blocker中的prod()和prodAll()，就会发现这是有意义的。这些方法是synchronized的，这意味着它们将获取自身的锁，因此它们调用notify()或notifyAll()时，只在这个锁上调用是符合逻辑的——因此，将只唤醒在等待这个特定锁的任务。

Blocker.waitingCall()非常简单，以至于在本例中，你只需声明for(;;)而不是while(!Thread.interrupted())就可以达到相同的效果，因为在本例中，由于异常而离开循环和通过检查interrupted()标志离开循环是没有任何区别的——在两种情况下都要执行相同的代码。但是，事实上，这个示例选择了检查interrupted()，因为存在着两种离开循环的方式。如果在以后的某个时刻，你决定要在循环中添加更多的代码，那么如果没有覆盖从这个循环中退出这两条路径，就会产生引入错误的风险。

**如果本文对您有很大的帮助，还请点赞关注一下。**