（四）线程之间的协作

前面提到，当我们用多线程同时运行多个任务时，可以通过锁（互斥）来同步两个任务的行为，从而使得一个任务不会干涉另一个任务的资源。

这一章要解决的问题不是彼此间的干涉，而是彼此间的协作。协作时，关键问题是握手。为了实现握手，我们同样使用互斥，因为互斥能够确保只有一个任务可以相应某个信号，这样就可以根除竞争条件。

在互斥之上，我们为任务添加一种途径，可以将自身挂起，直到某些外部条件的变化。这种握手可以使用Object方法的wait()和notify()来实现。Java还提供了具有await()和signal()方法的Condition对象。

wait()和notify()

wait()会在等待外部环境变化的时候将任务挂起，并且只有在notify()或notifyAll()发生时，这个任务才会被唤醒并检查所产生的变化。

sleep()和yield()都不会释放锁资源，但调用wait()时将释放锁，这意味着另一个任务可以获得锁对象。

有意思的是wait(),notify()或notifyAll()是基类Object的方法，而不是Thread。

只能在同步控制方法或同步控制块内调用wait(),notify()或notifyAll()这三个方法。因为调用这三个方法前必须拥有对象的锁。

wait()方法被唤醒时将从获得锁的位置重新执行，因此在锁保护代码内应再次判断是否需要再次进入wait()。

下面举一个煮汤的例子，规则为先加水，再加盐，依次执行下去，两个任务不能同时进行，必须交替进行。因此每个任务都要判断另一个任务是否完成，如果没有，则将自己挂起。当一个任务完成后，执行notify()通知另一个任务。

public class SoupTask{
    public static void main(String[] args) throws InterruptedException{
        Soup soup = new Soup();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new WaterAdd(soup));
        exec.execute(new SaltAdd(soup));
        TimeUnit.SECONDS.sleep(5);
        exec.shutdownNow();
    }  
}

class Soup {
    private boolean waterOk=false;//判断是否已经加好水，false为已经加好盐
    public synchronized void addWater(){
        waterOk=true;
        notifyAll();
    }
    public synchronized void addSalt(){
        waterOk=false;
        notifyAll();
    }
    public synchronized void waitForWater() throws InterruptedException{
        while(waterOk==true) wait();
    }
    public synchronized void waitForSalt() throws InterruptedException{
        while(waterOk==false) wait();
    }  
}

class WaterAdd implements Runnable{
    private Soup soup;
    public WaterAdd(Soup soup){this.soup=soup;}
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                soup.waitForWater();
                System.out.println("WaterAdd on task!");
                TimeUnit.MILLISECONDS.sleep(313);
                soup.addWater();
                soup.waitForSalt();
            }
        }catch(InterruptedException e){
            System.out.println("Exsiting via interrupt!");
        }
        System.out.println("Ending WaterAdd task!");
    }    
}
class SaltAdd implements Runnable{
    private Soup soup;
    public SaltAdd(Soup soup){this.soup=soup;}
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                soup.waitForSalt();
                System.out.println("SaltAdd on task!");
                TimeUnit.MILLISECONDS.sleep(230);
                soup.addSalt();
                soup.waitForWater();
            }
        }catch(InterruptedException e){
            System.out.println("Exsiting via interrupt!");
        }
        System.out.println("Ending SaltAdd task!");
    }    
}
输出：
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
SaltAdd on task!
WaterAdd on task!
Exsiting via interrupt!
Ending SaltAdd task!
Exsiting via interrupt!
Ending WaterAdd task!

notify()和notifyAll()

在众多等待同一个锁的任务中，notify()只会唤醒一个任务，因此我们如果在多任务中（同一个锁对象）使用notify()的话，必须保证所有任务都是等待相同的条件。否则，如果有多个任务等待不同的条件，那么我们就不知道是否唤醒了恰当的任务。

notifyAll()会唤醒所有因同一个锁对象而等待的任务。比如同一个对象上有多个任务等待不同的条件，因为同一个对象只有一个内置锁对象。因此可以使用notifuAll()唤醒这些任务。而程序的其他部分，等待其他锁的任务则不会被唤醒。

生产者——消费者

有一个饭店，饭店有一个厨师和一个服务员，服务员必须等待厨师做好食物，当厨师做好后会通知他，服务员取走食物，返回继续等待。

public class Restaurant {
    Meal meal;//食物
    ExecutorService exec = Executors.newCachedThreadPool();//线程池
    WaitPerson waitPerson = new WaitPerson(this);//服务员
    Chef chef = new Chef(this);//厨师
    public Restaurant(){
        exec.execute(chef);
        exec.execute(waitPerson);
    }
    public static void main(String[] args) {
        new Restaurant();
    }
}
//食物
class Meal{
    private final int orderNum;
    public Meal(int orderNum){
        this.orderNum=orderNum;
    }
    public String toString(){
        return "Meal "+orderNum+" ";
    }
}
//服务员
class WaitPerson implements Runnable{
    private Restaurant restaurant;
    public WaitPerson(Restaurant restaurant){
        this.restaurant=restaurant;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                synchronized(this){
                    //等待厨师做好食物，并通知服务员
                    while(restaurant.meal==null)
                        wait();
                    //取走食物
                    System.out.println("WaitPerson get "+restaurant.meal);
                }
                //通知厨师，食物已被取走
                synchronized(restaurant.chef){
                    restaurant.meal=null;
                    restaurant.chef.notifyAll();
                }
            }
        }catch(InterruptedException e){
            System.out.println("WaitPerson interrupted!");
        }
    }
}

class Chef implements Runnable{
    private Restaurant restaurant;
    private int count =0 ;
    public Chef(Restaurant restaurant){
        this.restaurant=restaurant;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                synchronized(this){
                    //食物未被取走，则等待
                    while(restaurant.meal!=null)
                        wait();
                }
                if(++count == 10){
                    System.out.println("Out of food, closing");
                    restaurant.exec.shutdownNow();
                }
                //订单来到
                System.out.println("Order up!");
                //做好食物，并通知服务员
                synchronized(restaurant.waitPerson){
                    restaurant.meal = new Meal(count);
                    restaurant.waitPerson.notifyAll();
                }
                TimeUnit.MILLISECONDS.sleep(100);
            }
        }catch(InterruptedException e){
            System.out.println("Chef interrupted!");
        }
    }    
}

生产者——消费者队列

wait()和notify()以一种非常低级的方式解决了任务互操作的问题，即每次交互时都握手。

许多情况下，我们可以使用同步队列来解决任务协作的问题，同步队列在任何时刻都只允许一个任务的插入或移除元素。java.util.concurrent.BlockQueue接口提供了这个队列。

如果消费者任务试图从队列获取对象，而此时队列为空，那么队列可以挂起消费者任务。（这是队列的核心）

以下示例为向同步队列持续添加几个任务，然后依次取出执行：

public class TestBlockingQueues {
    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingQueue<LiftOff> queue = new LinkedBlockingQueue<LiftOff>();
        LiftOffRunner runner = new LiftOffRunner(queue);
        Thread t = new Thread(runner);
        t.start();
        for(int i=0;i<5;i++){
            System.out.println("添加任务！");
            Thread.sleep(200);
            runner.add(new LiftOff(5));
        }
    }
}
class LiftOffRunner implements Runnable{
    private BlockingQueue<LiftOff> rockets;
    public LiftOffRunner(BlockingQueue<LiftOff> queue){
        this.rockets=queue;
    }
    public void add(LiftOff lo){
        try{
            rockets.put(lo);
        }catch(InterruptedException e){
            System.out.println("Interrupted during put!");
        }
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                LiftOff rocket = rockets.take();
                rocket.run();
            }
        }catch(InterruptedException e){
            System.out.println("Exiting LiftOffRunner!");
        }
    }
    
}
class LiftOff implements Runnable{
    protected int countDown =10;
    private static int taskCount =0;
    private final int id = taskCount++;
    public LiftOff(){}
    public LiftOff(int countDown){
        this.countDown=countDown;
    }
    @Override
    public void run() {
        while(countDown-->0){
            try {
                System.out.println("#"+id+"("+(countDown>0?countDown:"LidtOff!")+").");
                Thread.currentThread().sleep(100);
                Thread.yield();
            } catch (InterruptedException e) {e.printStackTrace();}
            
        }
    }

}
输出：
添加任务！
run:
添加任务！
#0(4).
#0(3).
添加任务！
#0(2).
#0(1).
添加任务！
#0(LidtOff!).
run:
#1(4).
添加任务！
#1(3).
#1(2).
#1(1).
#1(LidtOff!).
run:
#2(4).
#2(3).
#2(2).
#2(1).
#2(LidtOff!).
run:
#3(4).
#3(3).
#3(2).
#3(1).
#3(LidtOff!).
run:
#4(4).
#4(3).
#4(2).
#4(1).
#4(LidtOff!).

以上代码的任务是LiftOff（火箭发射），我们把它放在BlockingQueue里。在main方法里，启动一个LiftOffRunner线程，通过该线程管理BlockingQueue的任务添加和执行。LiftOff rocket = rockets.take()表示从BlockingQueue取出一个任务，如果没有可执行的任务，则等待。

 

如果有一台机器具有三个任务，制作吐司，抹黄油，抹果酱。我们可以使用同步队列来协调整个过程。

我们建立三个同步队列分别存放干吐司，黄油吐司和果酱吐司。然后从前一个队列取出吐司经过处理后放入后一个队列。这样保证了工作的有序进行。

public class ToastMatic {

    public static void main(String[] args) throws InterruptedException {
        //建立三条流水线
        LinkedBlockingQueue<Toast> toastQueue = new LinkedBlockingQueue<Toast>();
        LinkedBlockingQueue<Toast> butteredQueue = new LinkedBlockingQueue<Toast>();
        LinkedBlockingQueue<Toast> finishedQueue = new LinkedBlockingQueue<Toast>();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Toaster(toastQueue));
        exec.execute(new Butterer(toastQueue,butteredQueue));
        exec.execute(new Jammer(butteredQueue,finishedQueue));
        exec.execute(new Easter(finishedQueue));
        
        TimeUnit.SECONDS.sleep(1);
        exec.shutdownNow();
    }

}

class Toast{
    public enum Status {DRY,BUTTERED,JAMMED};//表示吐司的状态：干的，有黄油的，有果酱的
    private Status status=Status.DRY;//刚生产的吐司是干的
    private final int id;
    public Toast(int id) {this.id=id;}
    public void butter(){this.status=Status.BUTTERED;}//抹黄油
    public void jam(){this.status=Status.JAMMED;}//抹果酱
    public Status getStatue(){return this.status;}
    public int getId(){return this.id;}
    public String toString(){return "Toast "+id+": "+status;}
}
//下面的线程负责生成吐司，并放在ToastQueue中
class Toaster implements Runnable{
    private LinkedBlockingQueue<Toast> toastQueue;
    private int count = 0;
    private Random random= new Random(47);
    public Toaster(LinkedBlockingQueue<Toast> toastQueue){
        this.toastQueue =  toastQueue;
    }
    public void run(){
        try{
            while(!Thread.interrupted()){
                TimeUnit.MICROSECONDS.sleep(500+random.nextInt(500));
                Toast toast = new Toast(count++);
                System.out.println("新的吐司："+toast);
                toastQueue.put(toast);
            }
        }catch(InterruptedException e){
            System.out.println("Toaster interrupted!");
        }
        System.out.println("Toaster off");
    }
}

//取出toastQueue的吐司，抹黄油，并放入butteredQueue
class Butterer implements Runnable{
    private LinkedBlockingQueue<Toast> toastQueue,butteredQueue;
    public Butterer(LinkedBlockingQueue<Toast> toastQueue,LinkedBlockingQueue<Toast> butteredQueue){
        this.toastQueue=toastQueue;
        this.butteredQueue=butteredQueue;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                Toast toast = toastQueue.take();
                toast.butter();
                System.out.println("抹黄油的吐司："+toast);
                butteredQueue.put(toast);
            }
        }catch(InterruptedException e){
            System.out.println("Butterer interrupted!");
        }
        System.out.println("Butterer off");
    }
}

//从butteredQueue取出吐司，抹果酱，放入finishedQueue
class Jammer implements Runnable{
    private LinkedBlockingQueue<Toast> butteredQueue,finishedQueue;
    public Jammer(LinkedBlockingQueue<Toast> butteredQueue,LinkedBlockingQueue<Toast> finishedQueue){
        this.butteredQueue=butteredQueue;
        this.finishedQueue=finishedQueue;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                Toast toast = butteredQueue.take();
                toast.jam();
                System.out.println("抹果酱的吐司："+toast);
                finishedQueue.put(toast);
            }
        }catch(InterruptedException e){
            System.out.println("Jammer interrupted!");
        }
        System.out.println("Jammer off");
    }
}

//输出所有完成的吐司
class Easter implements Runnable{
    private LinkedBlockingQueue<Toast>finishedQueue;
    private int counter =0;
    public Easter(LinkedBlockingQueue<Toast>finishedQueue){
        this.finishedQueue=finishedQueue;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                Toast toast = finishedQueue.take();
                if(toast.getId()!=counter++||toast.getStatue()!=Toast.Status.JAMMED){
                    System.out.println(">>>ERROR :"+toast);
                    System.exit(1);
                }else{
                    System.out.println("吐司完成： :"+toast);
                }
            }
        }catch(InterruptedException e){
            System.out.println("Easter interrupted!");
        }
        System.out.println("Easter off");
    }
}

输出：
新的吐司：Toast 0: DRY
抹黄油的吐司：Toast 0: BUTTERED
抹果酱的吐司：Toast 0: JAMMED
吐司完成： :Toast 0: JAMMED
新的吐司：Toast 1: DRY
抹黄油的吐司：Toast 1: BUTTERED
抹果酱的吐司：Toast 1: JAMMED
吐司完成： :Toast 1: JAMMED
新的吐司：Toast 2: DRY
抹黄油的吐司：Toast 2: BUTTERED
抹果酱的吐司：Toast 2: JAMMED
...
...
新的吐司：Toast 37: DRY
抹黄油的吐司：Toast 37: BUTTERED
抹果酱的吐司：Toast 37: JAMMED
新的吐司：Toast 38: DRY
抹黄油的吐司：Toast 38: BUTTERED
抹果酱的吐司：Toast 38: JAMMED
吐司完成： :Toast 36: JAMMED
吐司完成： :Toast 37: JAMMED
吐司完成： :Toast 38: JAMMED
Butterer interrupted!
Butterer off
Toaster interrupted!
Toaster off
Easter interrupted!
Jammer interrupted!
Jammer off
Easter off

 

任务之间使用管道输入/输出（线程通信）

PipeWriter类：允许任务向管道写

PipeReader类：允许不同任务从同一个管道读

public class PipedIO {

    public static void main(String[] args) throws IOException, InterruptedException {
        Sender sender = new Sender();
        Receiver receiver = new Receiver(sender);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(sender);
        exec.execute(receiver);
        TimeUnit.SECONDS.sleep(10);
        exec.shutdownNow();
    }

}

class Sender implements Runnable{
    private Random random = new Random(47);
    private PipedWriter out = new PipedWriter();
    public PipedWriter getPipedWriter(){return out;}
    @Override
    public void run() {
        try{
            while(true){
                for(char c='A';c<'z';c++){
                    out.write(c);
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
                }
            }
        }catch(IOException e){
            System.out.println("IOException");
        }catch(InterruptedException e){
            System.out.println("InterruptedException");
        }
    }
}
class Receiver implements Runnable{
    private PipedReader in;
    public Receiver(Sender sender) throws IOException{
        in = new PipedReader(sender.getPipedWriter());
    }
    @Override
    public void run() {
        try{
            while(true){
                System.out.print("Read:"+(char)in.read()+"
");
            }
        }catch(IOException e){
            e.printStackTrace();
        }
    }
    
}

使用同步队列实现线程通信

使用两个队列A2BQueue，B2AQueue分别存放A发给B和B发给A的数据。

public class Talk {

    public static void main(String[] args) throws InterruptedException {
        LinkedBlockingQueue<Integer> A2BQueue = new LinkedBlockingQueue<Integer>();
        LinkedBlockingQueue<Integer> B2AQueue = new LinkedBlockingQueue<Integer>();
        //A2BQueue.put(0);//放入初始值
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new A(A2BQueue,B2AQueue));
        exec.execute(new B(A2BQueue,B2AQueue));
        
        TimeUnit.MILLISECONDS.sleep(50);
        exec.shutdownNow();
    }

}
class A implements Runnable{
    LinkedBlockingQueue<Integer> A2BQueue,B2AQueue;
    private Random random = new Random(47);
    public A(LinkedBlockingQueue<Integer> A2BQueue,LinkedBlockingQueue<Integer>B2AQueue){
        this.A2BQueue=A2BQueue;
        this.B2AQueue=B2AQueue;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                int n = random.nextInt(10000);
                System.out.println("A2B: "+n);
                A2BQueue.put(n);
                System.out.println("A read: "+B2AQueue.take());
            }
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }
}
class B implements Runnable{
    LinkedBlockingQueue<Integer> A2BQueue,B2AQueue;
    private Random random = new Random(57);
    public B(LinkedBlockingQueue<Integer> A2BQueue,LinkedBlockingQueue<Integer>B2AQueue){
        this.A2BQueue=A2BQueue;
        this.B2AQueue=B2AQueue;
    }
    @Override
    public void run() {
        try{
            while(!Thread.interrupted()){
                System.out.println("B read: "+A2BQueue.take());
                int n = random.nextInt(10000);
                System.out.println("B2A: "+n);
                B2AQueue.put(n);
            }
        }catch(InterruptedException e){
            e.printStackTrace();
        }
    }
}
输出：
A2B: 9258
B read: 9258
B2A: 1998
A read: 1998
A2B: 555
B read: 555
B2A: 7611
A read: 7611
A2B: 6693
B read: 6693
B2A: 8902
A read: 8902
A2B: 1861
B read: 1861
B2A: 1332
A read: 1332
A2B: 961
B read: 961
B2A: 2121
...
...