首页 > 代码库 > 漫谈并发编程(五)：线程之间的协作
漫谈并发编程(五)：线程之间的协作
编写多线程程序须要进行线程协作。前面介绍的利用相互排斥来防止线程竞速是来解决线程协作的衍生危害的。编写线程协作程序的关键是解决线程之间的协调问题，在这些任务中，某些能够并行运行，可是某些步骤须要全部的任务都结束之后才干开动。
wait()与notifyAll()
wait()使你能够等待某个条件发生变化，wait()会在等待外部世界产生变化的时候将任务挂起，而且仅仅有在notify()或notifyAll()发生时，即表示发生了某些感兴趣的事物，这个任务才会被唤醒并去检查所产生的变化。
调用sleep()的时候锁并没有被释放，调用yield()也属于这样的情况。理解这一点非常关键。还有一方面。当一个任务在方法里遇到了对wait()的调用的时候。线程的运行被挂起，对象上的锁被释放。
因此wait()释放锁。这就意味着还有一个任务能够获得这个锁。因此在该对象中的其它synchronized方法能够在wait()期间被调用，而其它的方法通常将会产生改变，而这样的改变正是使被挂起的任务又一次唤醒所感兴趣的变化。
有两种形式的wait()。第一种版本号接受毫秒数为參数，含义与sleep()方法里的參数的意思同样，都是指"在此期间暂停"。可是与sleep()不同的是，对于wait()而言：
- 在wait()期间对象锁是释放的
- 能够通过notify()、notifyAll()，或者令时间到期。从wait()中恢复运行。
另外一种，也是更常见形式的wait()不接收不论什么參数。
这样的wait()将无限等待下去，直到线程接收到notify()或者notifyAll()消息。
能够想象，wait()、notify()、notifyAll()一定是基于某个"东西"，把自身状态附加上去，来实现这样的通知及状态的变化。
考虑设计方式：1. 这样的东西能够单独被定义出来。 2. 在Object中提供该"东西"的实现。 明显另外一种方式要轻松方便很多。迁移性更强。其次。这样的东西可能不是线程安全的，所以须要锁来支持。
使用synchronized来进行同步的保护是理所应当，由于"东西"的实现就在Object中，其次使用synchronized的优点是一定程度能够避免由于锁不一致的情况下产生的wait()及notifyAll的不正确应。wait()在一把锁中释放了锁，和notifyAll在还有一把锁进行操作毫无相关。
java要求仅仅能在同步控制方法或同步控制块里调用wait()、notify()和notifyAll()。
以下演示一个样例。一个是将蜡涂到Car上，一个是抛光它。抛光任务在涂蜡任务完毕之前，是不能运行其工作的。而涂蜡任务在涂还有一层蜡之前，必须等待抛光任务完毕。WaxOn和WaxOff都使用了Car对象，该对象在这些任务等待条件变化时候，使用wait()和notifyAll()来挂起和又一次启动这些任务：
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;}
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("Exiting via interrupt");
}
System.out.println("Ending Wax On task");
}
}
class WaxOff implements Runnable {
private Car car;
public WaxOff(Car c) {car = c;}
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("Exiting via interrupt");
}
System.out.println("Ending Wax Off task");
}
}
public class WaxOMatic {
public static void main(String[] args) throws Exception{
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
} 前面的演示样例强调你必须用一个检查感兴趣的条件的while循环包围wait()。
这非常重要，由于：
- 你可能有多个任务出于同样的原因在等待一个锁，而第一个唤醒任务可能已经改变这样的状况(即使你没有这么做，有人也会通过继承你的类去这么做)。假设属于这样的情况，那么这个任务应该被再次挂起，直至其感兴趣的条件发生变化。
- 也有可能某些任务处于不同的原因在等待你的对象上锁(在这样的情况下必须使用(notifyAll))。在这样的情况下，你须要检查是否已经由正确的原因唤醒，假设不是，就再次调用wait()。
notify()与notifyAll()
由于在技术上，可能会有多个任务在单个Car对象上处于wait()状态。因此调用notifyAll()比调用notify()要更安全。可是，上面程序的结构仅仅会有一个任务处于wait()状态，因此你能够使用notify()来取代notifyAll()。
使用notify()而不是notifyAll()是一种优化。
使用notify()时。在众多等待同一个锁的任务中仅仅有一个会被唤醒，因此假设你希望使用notify()就必须保证被唤醒的是恰当的任务。
另外，为了使用notify()。全部任务必须等待同样的条件，由于假设你有多个任务在等待不同的条件。那么你就不会知道是否唤醒的恰当的任务。假设使用notify()，当条件发生变化时，必须仅仅有一个任务能从中受益。最后，这些限制对全部可能存在的子类都必须总是起作用的。
假设这些规则中有不论什么一条不满足。那么你就必须使用notifyAll()而不是notify()。
用wait()和notifyAll()实现生产者消费者问题
使用wait()和notifyAll()时一定要注意不能两层嵌套synchronized，假设使用了两层，则外层的sycnhronized加的锁无法释放。
并且须要注意的是不能使用Lock来限制资源的訪问。由于wait时无法释放该锁。假设还要限制在notifyAll时不能notifyAll到同类。那么实现这个问题还是有难度的。
以下贴上一个自己一个粗陋的实现。各位朋友有美丽代码的也能够贴上来交流下。
class Meal {
}
class WaitPerson implements Runnable {
private String name;
private Restaurant restaurant;
public WaitPerson(String name, Restaurant res) {
this.name = name;
this.restaurant = res;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (restaurant.waitPersons) {
while (restaurant.meals.size() < 1) {
restaurant.waitPersons.wait();
}
}
synchronized (restaurant.chefs) {
if (restaurant.meals.size() >= 1) {
restaurant.meals.poll();
restaurant.chefs.notifyAll();
System.out.println(name + " consumed a meal !");
}
}
}
} catch (InterruptedException e) {
System.out.println(name + " is ended via InterruptedException !");
return;
}
System.out.println(name + " is ended via InterruptedException !");
}
}
class Chef implements Runnable {
private String name;
private Restaurant restaurant;
public Chef(String name, Restaurant res) {
this.name = name;
this.restaurant = res;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (restaurant.chefs) {
while (restaurant.meals.size() > 10) {
restaurant.chefs.wait();
}
}
synchronized (restaurant.waitPersons) {
if (restaurant.meals.size() <= 10) {
restaurant.meals.add(new Meal());
restaurant.waitPersons.notifyAll();
System.out.println(name + " produced a meal !");
}
}
}
} catch (InterruptedException e) {
System.out.println(name + " is ended via InterruptedException !");
return;
}
System.out.println(name + " is ended via InterruptedException !");
}
}
public class Restaurant {
public Queue<Meal> meals = new ConcurrentLinkedQueue<Meal>();
public List<WaitPerson> waitPersons = new ArrayList<WaitPerson>();
public List<Chef> chefs = new ArrayList<Chef>();
public static void main(String[] args) throws InterruptedException {
Restaurant res = new Restaurant();
ExecutorService exec = Executors.newCachedThreadPool();
Chef chef1 = new Chef("chef1", res);
Chef chef2 = new Chef("chef2", res);
res.chefs.add(chef1);
res.chefs.add(chef2);
exec.execute(chef1);
exec.execute(chef2);
WaitPerson waitPerson1 = new WaitPerson("waitPerson1", res);
WaitPerson waitPerson2 = new WaitPerson("waitPerson2", res);
res.waitPersons.add(waitPerson1);
res.waitPersons.add(waitPerson2);
exec.execute(waitPerson1);
exec.execute(waitPerson2);
// TimeUnit.MILLISECONDS.sleep(3000);
// exec.shutdownNow();
}
} 上面这个程序能够证明出来是线程安全的。
只是使用这样的方式实在是太晦涩了。生产者消费者问题的机制须要我们去控制，实际上，java并发类库为我们提供了这样的模型的实现，我们待会会用堵塞队列来重写这个问题。
使用显式的Lock和Condition对象
我们能够显式的使用Condition对象来替代我前面提到的"东西"，使用这样的方式将更加灵活，且有更清晰的辩识度，但会添加程序中对象的数量。你能够通过在Condition上调用await()来挂起一个任务。当外部条件发生变化。意味着某个任务应该继续运行时。你能够通过调用signal()来通知这个任务，从而唤醒一个任务，或者调用signalAll()来唤醒全部在这个Condition上被其自身挂起的任务。
以下我们利用此工具重写前面样例中的Car类。
class Car {
private boolean waxOn = false;
private Lock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
public void waxed() {
lock.lock();
try {
waxOn = true;
condition.signalAll();
} finally {
lock.unlock();
}
}
public void buffed( ) {
lock.lock();
try {
waxOn = false;
condition.signalAll();
} finally {
lock.unlock();
}
}
public void waitForWaxing( ) throws InterruptedException{
lock.lock();
try{
while(waxOn == false)
condition.await();
} finally {
lock.unlock();
}
}
public void waitForBuffing( ) throws InterruptedException {
lock.lock();
try {
while(waxOn == true)
condition.await( );
} finally {
lock.unlock();
}
}
}使用BlockingQueue来解决生产者消费者问题
java帮我们抽象了生产者消费者问题。我们能够使用同步队列来解决任务协作的问题。同步队列在不论什么时刻都仅仅同意一个任务插入或移除元素。
在java.util.concurrent.BlockingQueue接口中提供了这个队列，这个接口有大量的实现。你通常能够使用LinkedBlockingQueue。它是一个无界队列，还能够使用ArrayBlockingQueue，它具有固定的尺寸，因此你能够在它被堵塞之前，向当中放置有限数量的元素。
假设消费者任务试图从队列中获取对象，而该队列此时为空，那么这些队列还能够挂起消费者任务，而且当有很多其它的元素可用时恢复消费者任务。堵塞队列能够解决很大量的问题。而其方式与wait()和notifyAll()相比，则简单并可靠太多。
以下利用堵塞队列实现了上面的餐厅问题。
class Meal {
}
class WaitPerson implements Runnable {
private String name;
private RestaurantBlookingQueue restaurant;
public WaitPerson(String name, RestaurantBlookingQueue res) {
this.name = name;
this.restaurant = res;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
restaurant.meals.take();
System.out.println(name + "taked a Meal");
Thread.sleep(100);
}
} catch (InterruptedException e) {
System.out.println(name + " is ended via InterruptedException !");
return;
}
System.out.println(name + " is ended via InterruptedException !");
}
}
class Chef implements Runnable {
private String name;
private RestaurantBlookingQueue restaurant;
public Chef(String name, RestaurantBlookingQueue res) {
this.name = name;
this.restaurant = res;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
restaurant.meals.put(new Meal());
System.out.println(this.name + "made a meal");
Thread.sleep(100);
}
} catch (InterruptedException e) {
System.out.println(name + " is ended via InterruptedException !");
return;
}
System.out.println(name + " is ended via InterruptedException !");
}
}
public class RestaurantBlookingQueue {
public BlockingQueue<Meal> meals = new ArrayBlockingQueue<Meal>(10);
public List<WaitPerson> waitPersons = new ArrayList<WaitPerson>();
public List<Chef> chefs = new ArrayList<Chef>();
public static void main(String[] args) throws InterruptedException {
RestaurantBlookingQueue res = new RestaurantBlookingQueue();
ExecutorService exec = Executors.newCachedThreadPool();
Chef chef1 = new Chef("chef1", res);
Chef chef2 = new Chef("chef2", res);
res.chefs.add(chef1);
res.chefs.add(chef2);
exec.execute(chef1);
exec.execute(chef2);
WaitPerson waitPerson1 = new WaitPerson("waitPerson1", res);
WaitPerson waitPerson2 = new WaitPerson("waitPerson2", res);
res.waitPersons.add(waitPerson1);
res.waitPersons.add(waitPerson2);
exec.execute(waitPerson1);
exec.execute(waitPerson2);
// TimeUnit.MILLISECONDS.sleep(3000);
// exec.shutdownNow();
}
}任务间使用管道进行输入/输出
通过输入/输出在线程间进行通信通常非常实用。
提供线程功能的类库以"管道"的形式对线程的输入/输出提供了支持。
它们在Java输入/输出类库中的相应物就是PipedWriter类(同意任务向管道写)和PipedReader类(同意不同任务从同一个管道读)。这个模型能够看成是"生产者-消费者"问题的变体。
管道基本是一个堵塞队列，存在于多个引入BlookingQueue之前的Java版本号。
class Sender implements Runnable {
private Random rand = new Random(47);
private PipedWriter out = new PipedWriter();
public PipedWriter getPipedWriter( ) {return out;}
public void run( ) {
try {
while(true) {
for(char c = ‘A‘ ; c <= ‘z‘; c++) {
out.write(c);
TimeUnit.MILLISECONDS.sleep( rand.nextInt(500));
}
}
} catch (IOException e) {
System.out.println(e + " Sender write exception");
} catch (InterruptedException e) {
System.out.println(e + " Sender sleep exception");
}
}
}
class Receiver implements Runnable {
private PipedReader in;
public Receiver(Sender sender) throws IOException {
in = new PipedReader(sender.getPipedWriter());
}
public void run( ) {
try {
while(true) {
System.out.print("Read: "+(char)in.read() + ", ");
}
} catch (IOException e) {
System.out.println(e + " Receiver read exception");
}
}
}
public class PipedIO {
public static void main(String []args) throws Exception {
Sender sender = new Sender( );
Receiver receiver = new Receiver( sender );
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(sender);
exec.execute(receiver);
TimeUnit.SECONDS.sleep( 4 );
exec.shutdownNow();
}
}死锁
死锁本是操作系统的中概念，由于操作系统中会遇到非常多可能发生死锁的状况。但我们在并发程序常常也须要预防死锁。特别是多个线程在并发的訪问多个对象的时候。首先，我们须要从逻辑上避免死锁发生的可能性，比如哲学家进餐问题。一般在程序中的解决方案是一次性将资源全然分配给它，为了提供并发度，须要我们进一步缩小并发锁的范围。
除了逻辑上预防并发。我们还须要处理意外情况，比如获取到资源的线程中途挂掉。我们须要释放资源。在程序中即释放锁。在程序中能够通过try-catch实现。
漫谈并发编程(五)：线程之间的协作
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。