漫谈并发编程(五)：线程之间的协作

时间：2014-11-03 22:40:55 阅读：239 评论：0 收藏：0 [点我收藏+]

标签：并发

编写多线程程序需要进行线程协作，前面介绍的利用互斥来防止线程竞速是来解决线程协作的衍生危害的。编写线程协作程序的关键是解决线程之间的协调问题，在这些任务中，某些可以并行执行，但是某些步骤需要所有的任务都结束之后才能开动。

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

漫谈并发编程(五)：线程之间的协作

标签：并发

原文地址：http://blog.csdn.net/troy__/article/details/40748003

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行