Thinking in java 之并发其四：线程之间的协作

作者: Tearsweet | 来源:发表于2018-12-02 11:28 被阅读14次

Thinking in java 之并发其四：线程之间的协作

一、前言

在第二章的时候，我们学会了通过锁的方式来同步多个任务，从而使得一个任务不会干涉另一个任务的资源。也就是说，多个任务在交替步入某项共享资源（通常是内存），可以使用互斥来使得任何时刻只有一个任务可以访问这项资源。而接下来，我们需要学习如何使任务彼此之间可以协作，以使得多个任务可以一起工作去解决问题。那么，我们面临的问题不再是彼此之间的干涉，而是彼此之间的协调。

当任务协作时，关键问题是这些任务之间的握手。为了实现这种握手，我们使用了相同的基础特性：互斥。在这种情况下，互斥能够确保只有一个任务可以相应某个信号，这样就可以根除任何可能的竞争条件。在互斥之上，我们为任务添加了一种途径，可以将其自身挂起，直到知道某些外部条件发生变化，表示现在可以让这个任务向前开动了为止。

二、wait() 和 notifyAll()

wait() 可以使任务挂起，直到 notify() 或者 notifyAll() 将其唤起。通常我们在任务需要等待前置任务完成时将其挂起，当前置任务完成时在将其唤醒以继续执行。

在使用 sleep 或者 yeild 的时候，对象的锁并没有被释放，也就是说，其他任务还是不能够访问被锁的资源。而使用 wait() 时，将会释放锁，这意味着另一个任务可以获得这个锁，并且可以执行其他的 synchronized 方法。

wait() 也可以使用时间作为参数，表示，当指定的时间长度过去之后，任务会自动唤醒。当然，在时间到达之前，我们也可以通过 notify() 或者 notifyAll() 将其唤醒。并且在挂起的这段时间里，锁是释放的。

与 sleep() 等方法不同的是，wait()、notify() 和 notifyAll() 不属于 Thread 而是属于 Object。所以，即便是在非同控制的方法里，我们依旧可以调用这三个方法，但是会抛出异常。

现在通过一个简单的示例，来展示任务之间是如何相互配合的。WaxOMatic.java 有两个过程，一个是将蜡涂到 Car 上，一个是抛光它。抛光任务在涂蜡完成之前，是不能执行的。而涂蜡在涂另一层蜡之前，必须等待抛光任务完成。WaxOn 和 WaxOff 都使用了 Car 对象，该对象在这些任务等待条件变化的时候，使用 wait() 和 notufyAll() 来挂起和重启。

package ThreadTest.cooperation;

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 waitFotBuffing() throws InterruptedException {
        while(waxOn == true)
            wait();
    }
}

class WaxOn implements Runnable{
    private Car car;
    public WaxOn(Car car) {
        this.car = car;
    }
    public void run() {
        try {
            while(!Thread.interrupted()) {
                System.out.println("Wax On!");
                TimeUnit.MILLISECONDS.sleep(200);
                car.waxed();
                car.waitFotBuffing();
            }
        }catch(InterruptedException e) {
            System.out.println("Exit via interrupt!");
        }
        System.out.println("Ending Wax On task");
    }
}

class WaxOff implements Runnable{
    private Car car;
    public WaxOff(Car car) {
        this.car=car;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                car.waitForWaxing();
                System.out.println("Wax Off!");
                TimeUnit.MILLISECONDS.sleep(200);
                car.buffed();
            }

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            System.out.println("Exit via interrupt!");
        }
        System.out.println("Ending Wax On 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);
        exec.shutdownNow();

    }

}

ar 的 WaxOn 初始状态为 false 即未涂蜡，这个时候，buff 是不能执行的，而在涂蜡之后，buff 可以执行，但 waxed 不能执行。WaxForWaxing 会循环检测 WaxOn 的状态，当其为false 时，唤醒所有任务，由于此时 WaxOn 为 false 所以只能执行 waxed 。同理 WaxForWaxing 也可以唤醒 Buff。

这里我们使用了 exec.shutdownNow() 他和 exec.shutdown（）的区别是，后者不会中断当前正在执行的任务，而前者会中断当前的任务。当我们调用 exec.shutdownNow() 时，他会立刻执行所有 Thread (由它发起的) 的 Inturrept()；

三、notify() 和 notifyAll()

notify() 和 notifyall() 的区别似乎在于唤醒一个阻塞任务，还是所有阻塞任务。而我们必须要思考的问题是，当做个任务处于阻塞状态是，notify() 唤醒的是哪一个？是随机唤醒？或者一其他方式唤醒任务。而对于 noifyAll() 是指唤醒所有位置的任务吗？先看下面这个示例：

package ThreadTest.cooperation;

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() + " ");
            }
            //System.out.print("\n");
        }catch(InterruptedException e) {
            System.out.println("Exiting");
        }
    }
    synchronized void prod() {notify();}
    synchronized void prodAll() {notifyAll();}
}

class Task implements Runnable{
    static Blocker blocker = new Blocker();

    @Override
    public void run() {
        blocker.waitingCall();
    }
}

class Task2 implements Runnable{
    static Blocker blocker = new Blocker();
    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;
            public void run() {
                if(prod) {
                    System.out.println("notify() ");
                    Task.blocker.prod();
                    prod = false;
                }else {
                    System.out.println("notifyAll() ");
                    Task.blocker.prodAll();
                    prod = true;
                }
            }
        },400,400);
        TimeUnit.SECONDS.sleep(4);
        timer.cancel();
        System.out.println("Timer canceled");
        TimeUnit.MILLISECONDS.sleep(500);
        System.out.println("shuting down");
        exec.shutdown();
    }

}


//output
/*notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
notify()
Thread[pool-1-thread-2,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
notify()
Thread[pool-1-thread-2,5,main]
notifyAll()
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
Timer canceled
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
shuting down
*/

从输出结果来看，有以下结论：

当任务因 wait() 被阻塞时，会释放锁。waitingCall() 被阻塞之后，我们可以使用和它一起被 synchronized 修饰的 prod() 和 prodAll() 方法充分的说明了这点。
notifyAll() 只能唤醒和它拥有同一个锁的任务，比如我们再调用了 Task.prodAll() 之后，Task2的任务并没有被唤醒。那么，不经产生疑问，如果调用 notifyAll() 的方法不被 + synchronized 修饰，会怎么样。答案是会抛出异常，不仅如此，wait()，notify() 也必须被 synchronized 修饰。否则，java 会抛出 java.lang.IllegalMonitorStateException 异常。
对于 notify() 到底唤醒了哪个对象，从结果来看，似乎总是唤醒了第一个线程，也就是最先启动的线程。在 java 对于 notify() 的解释中，可以发现，它是随机的。

Wakes up a single thread that is waiting on this object's monitor. If any threads are waiting on this object, one of them is chosen to be awakened. The choice is arbitrary and occurs at the discretion of the implementation. A thread waits on an object's monitor by calling one of the wait methods.

以上是 java 对于 notify() 的注释。

四、经典协作问题

关于线程之间的写作，有很多经典的例题，其中比较被人熟知的事生产者和消费者问题：

在一个饭店，它有一个厨师和一个服务员。服务员必须等待厨师准备好食物。当厨师准备好后，他会通知服务员，之后服务员上菜，然后返回继续等待。这是一个简单的任务协作示例。厨师代表生产者，服务员代表消费者。两个任务必须在食物被生产和消费时进行握手，而系统必须以有序的方式关闭。以下是对这个叙述建模的代码。

package ThreadTest.cooperation;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

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) {
                        System.out.println("no meal wait");
                        wait();
                    }
                }
                System.out.println("Waitperson got" + restaurant.meal);
                synchronized(restaurant.chef) {
                    //该锁是针对厨师的，告诉厨师没肉了，唤醒厨师生产肉的任务
                    restaurant.meal = null;
                    restaurant.chef.notifyAll();
                }
            }

        }catch(InterruptedException e) {
            System.out.println("WaitPersion 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, close");
                    restaurant.exec.shutdownNow();
                }
                System.out.println("Order Up");
                synchronized(restaurant.waitPerson) {
                    System.out.println("Meal count: " + count);
                    restaurant.meal=new Meal(count);
                    restaurant.waitPerson.notifyAll();
                }
                TimeUnit.MILLISECONDS.sleep(100);
            }
        }catch(InterruptedException e) {
            System.out.println("chef interrupt");
        }
    }

}
public class Restaurant {
    Meal meal;
    Chef chef = new Chef(this);
    WaitPerson waitPerson = new WaitPerson(this);
    ExecutorService exec = Executors.newCachedThreadPool();
    public Restaurant() {
        exec.execute(chef);
        exec.execute(waitPerson);
    }
    public static void main(String[] args) {
        new Restaurant();

    }

}
//Output
/*Order Up
no meal wait
Meal count: 1
Waitperson gotMeal 1
no meal wait
Order Up
Meal count: 2
Waitperson gotMeal 2
no meal wait
Order Up
Meal count: 3
Waitperson gotMeal 3
no meal wait
Order Up
Meal count: 4
Waitperson gotMeal 4
no meal wait
Order Up
Meal count: 5
Waitperson gotMeal 5
no meal wait
Order Up
Meal count: 6
Waitperson gotMeal 6
no meal wait
Order Up
Meal count: 7
Waitperson gotMeal 7
no meal wait
Order Up
Meal count: 8
Waitperson gotMeal 8
no meal wait
Order Up
Meal count: 9
Waitperson gotMeal 9
no meal wait
Out of food, close
Order Up
WaitPersion interrupted
Meal count: 10
chef interrupt*/

这是一个很简单的生产者与消费者的示例。在示例中，chef 和 waitPerson 都和同一个 restaurant 绑定。并通过这个 restaurant 来找到彼此（唤醒彼此的任务）。另外需要注意的点事是 synchronized 的使用。在前面的章节，我们说到过当 synchronized() 括号内指定的对象不同时，它们时属于不同的锁。而 notifyAll(或者是 notify()) 只能唤醒和自己共享同一把锁的代码块。因此 chef 生产肉的代码的 sychronized（）括号里是对应的 waitPerson。只有这样才能唤醒 waitPerson 里 synchronized（this）的代码。同理，waitPerson 的拿走肉的代码，也应该和唤醒 chef 的代码有对应的锁。

也许你会注意到当我们通过 “restaurant.exec.shutdownNow();” 去结束这个任务之后，输出里依旧出现了 " Order Up " 这说明 shutdownNow() 并没有立刻将任务停止。而 shutdownNow() 是立即调用 pool 里所有的 Thread 的 interrupt() 方法，但是对于一个在阻塞状态下的任务，interrupt() 的调用会抛出 InterruptedException。

五、使用显示的 Lock 和 Condition 对象

在 JUC(java.util.concurrent) 中，还有额外的显示工具可以用来重写WaxOMatic.java。使用互斥并允许任务挂起的基本类型是 Condition，可以通过在 Condition 上调用 await() 来挂起一个任务。当外部条件发生变化，意味着某个任务应该继续执行时，你可以通过调用 signal() 来通知这个任务，从而唤醒一个任务，或者调用 signalAll() 来唤醒所有在这个 Condition 上被其自身挂起的任务（与使用 notify() 相比，signalAll() 是更安全的方式）。
下面则是WaxOMatic.java 的重写版本：

package ThreadTest.cooperation;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class Car{
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private boolean waxOn = false;
    public void waxed() {
        lock.lock();
        try {
            waxOn =true;
            condition.signalAll();;
        }finally {
            lock.unlock();
        }

    }
    public void buffed() {
        lock.lock();
        try {
            waxOn = false;
            condition.signal();
        }finally {
            lock.unlock();
        }

    }

    public void waitForWaxing() throws InterruptedException {
        lock.lock();
        try {
            while(waxOn == false)
                condition.await();
        }finally {
            lock.unlock();
        }

    }

    public void waitFotBuffing() throws InterruptedException {
        lock.lock();
        try {
            while(waxOn == true)
                condition.await();
        }finally {
            lock.unlock();
        }
    }
}

class WaxOn implements Runnable{
    private Car car;
    public WaxOn(Car car) {
        this.car = car;
    }
    public void run() {
        try {
            while(!Thread.interrupted()) {
                System.out.println("Wax On!");
                TimeUnit.MILLISECONDS.sleep(200);
                car.waxed();
                car.waitFotBuffing();
            }
        }catch(InterruptedException e) {
            System.out.println("Exit via interrupt!");
        }
        System.out.println("Ending Wax On task");
    }
}

class WaxOff implements Runnable{
    private Car car;
    public WaxOff(Car car) {
        this.car=car;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                car.waitForWaxing();
                System.out.println("Wax Off!");
                TimeUnit.MILLISECONDS.sleep(200);
                car.buffed();
            }

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            System.out.println("Exit via interrupt!");
        }
        System.out.println("Ending Wax On 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);
        exec.shutdownNow();

    }

}

与原来的代码对比来看，新的 WaxOMatic 使用了 Lock 来代替 synchronized（前面有提到 lock 的用法）。同时用 condition 的 signal() 和 await() 来代替 wait() 和 notify()。notify() 只能幻唤醒和自己共享锁的代码块，同样的，condition 的 signal() 也只能唤醒同一个 Lock 的任务。

六、生产者——消费者与队列

在上一篇里，我们用 wait() 和 notifyAll() 完成了生产者——消费者的任务协作问题，即每次交互时都握手(notifyAll() 唤醒对方)。还可以使用更高的抽象级别，使用同步队列来解决任务协作问题，同步队列在任何时刻都只允许一个任务插入或移除元素。在 JUC 的 BlockingQueue 中提供了这个队列，这个接口有大量的标准实现。通常我们使用 LinkedBlockingQueue，它是一个无界队列。LinkedBlockingQueue 包含一个带 int 类型参数的构造方法，它允许我们制定队列的缓存大小。当生产者生产的商品塞满缓存之后，会被阻塞，直到消费者消费了商品之后，才会唤醒它。这样可以保证我们的内存不会被无限的占用。还可以使用 ArrayBlockingQueue，它具有固定的大小。

我们可以这样在生产者——消费者问题中使用阻塞队列。如果消费者任务试图从队列获取对象，而该队列此时为空，那么这些队列还可以挂起消费者任务，并且当条件允许时唤醒消费者任务。以下是阻塞队列使用的简单例子：

package ThreadTest.cooperation;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;

import ThreadTest.LiftOff;

class LiftOffRunner implements Runnable{
    private BlockingQueue<LiftOff> rockets;
    public LiftOffRunner(BlockingQueue<LiftOff> rockets) {
        this.rockets=rockets;
    }
    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("waking form take()");
        }

    }
}
public class TestBlockingQueues {
    static void getKey() {
        try {
            new BufferedReader(new InputStreamReader(System.in)).readLine();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

    }
    static void getKey(String message) {
        System.out.println("message");
        getKey();
    }
    static void test(String msg,BlockingQueue<LiftOff> queue) {
        System.out.println("message");
        LiftOffRunner runner = new LiftOffRunner(queue);
        Thread t = new Thread(runner);
        t.start();
        for(int i=0;i<5;i++) {
            runner.add(new LiftOff(5));
        }
        getKey("press 'Enter' (" + msg + ")");
        t.interrupt();
        System.out.println("finished " + msg + " test");
    }
    public static void main(String[] args) {
        test("LinkedBlockingQueue",new LinkedBlockingQueue<LiftOff>());
        test("ArrayBlockingQueue",new ArrayBlockingQueue<LiftOff>(3));
        test("SynchronousQueue",new SynchronousQueue<LiftOff>());
    }

}

这是一个简单的使用 BlockingQueue 的例子，在 main() 方法中，我们对3中常用的 BlockingQueue 进行了测试，为了输出更明确，我们再实际运行时，需要把另外两个注释掉（每次只测试一个）。当我们测试时，输出会在 5 个 LiftOff 倒计时完成之后停住，并且程序进入阻塞状态，这是因为，take() 方法会取走BlockingQueue 的首位线程并运行，而当首位为空的时候，会进入等待（阻塞）状态，直到有新的数据插入为止。而在该程序中，我们使用了键入 Enter 的方式来中断 LinkedBlockingQueue。

而且，我们可以注意到，5个 LiftOff 任务是在 t.start() 启动之后放进去的，在没有放进去之前，BlockingQueue 也是出于等待状态。

看到 BlockingQueue 的特性，我们或许会想到，能不能用它来重写生产者——消费者的程序。将商品 Meal 放到 BlockingQueue 中，然后由队列来控制生产者和消费者的唤醒与阻塞工作。以下是利用 BlockingQueue 对 Restaurant 的改写。

package ThreadTest.cooperation;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

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()) {
                System.out.println("Waitperson got " + restaurant.mealQueue.take());
            }
        }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()) {
                if(++count== 10) {
                    System.out.println("Out of food, close");
                    restaurant.exec.shutdownNow();
                }
                System.out.println("Order Up");
                Meal meal=new Meal(count);
                restaurant.mealQueue.put(meal);
                System.out.println("Meal count: " + count);
                }
        }catch(InterruptedException e) {
            System.out.println("chef interrupt");
        }
    }

}
public class Restaurant {
    Chef chef = new Chef(this);
    WaitPerson waitPerson = new WaitPerson(this);
    LinkedBlockingQueue<Meal> mealQueue = new LinkedBlockingQueue<Meal>(1);
    ExecutorService exec = Executors.newCachedThreadPool();
    public Restaurant() {
        exec.execute(chef);
        exec.execute(waitPerson);
    }
    public static void main(String[] args) {
        new Restaurant();

    }

}
//Output
/*Order Up
Meal count: 1
Order Up
Meal count: 2
Order Up
Waitperson got Meal 1
Waitperson got Meal 2
Waitperson got Meal 3
Meal count: 3
Order Up
Meal count: 4
Order Up
Meal count: 5
Order Up
Waitperson got Meal 4
Waitperson got Meal 5
Waitperson got Meal 6
Meal count: 6
Order Up
Meal count: 7
Order Up
Meal count: 8
Order Up
Waitperson got Meal 7
Waitperson got Meal 8
Waitperson got Meal 9
Meal count: 9
Out of food, close
Waitperson Interrupted
Order Up
chef interrupt
*/

在上述程序中，我们完全抛弃了 wait() 和 notify() 将工作完全交给了 BlockingQueue 内部去完成。这就是 BlockingQueue 的强大之处。

七、任务间使用管道进行输入/输出

通过输入/输出在线程间进行通信通常很有用。提供线程功能的类库以“管道”的形式对线程间的输入/输出提供支持。它们在 java 输入/输出类库中对应物就是 PipedWriter 类（允许任务向管道写）和 PipedReader 类（允许不同任务从同一管道中读取）。这个模型可以看成是“生产者——消费者”问题的变体，这里管道就是一个封装好的解决方案。管道基本上是一个阻塞队列。

下面是一个简单的列子，两个任务使用一个管道进行通信。

package ThreadTest.cooperation;

import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

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.MICROSECONDS.sleep(rand.nextInt(500));
                }
            }
        }catch(IOException e) {
            System.out.println(e+" Sender write Exception");
        }catch(InterruptedException e) {
            System.out.println(e+" Sender sleep interrupted");
        }
    }
}

class Receiver implements Runnable{
    private PipedReader in;
    public Receiver(Sender sender) throws IOException {
        this.in = new PipedReader(sender.getPipedWriter());
    }
    public void run() {
        try {
            while(true) {
                System.out.println("Read: " + (char)in.read()+",");
            }
        }catch(IOException e) {
            System.out.println(e+" Receiver read Exception");
        }
    }
}
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(5);
        exec.shutdownNow();
    }
}

不难发现，Piped~ 类和BlockingQueue 非常相似，都具备自动阻塞和唤醒任务的功能。其实在BlockingQueue 出现之前，Piped~ 类就在前几个版本出现了。

八、死锁

在锁的那一章，我们了解到，当多个任务共享同一个资源时，当某个任务先获取到该资源，其他任务必须等到该任务使用完之后释放锁才能继续进行自己的任务。那么会不会出现这样一种情况，A需要等待B完成，B需要等待C完成...N又需要等待A完成。多个任务形成了一个死循环，这种情况我们称为死锁。

关于死锁，最经典的莫过于哲学家吃饭问题，我们接下来就来讨论下这个问题。

5个哲学家，围绕着一个圆桌而坐，每两个人中间都会有一只筷子，也就是说每个哲学家的左右边都会有一只筷子（也就是说一共5只筷子）。当某位哲学家想吃饭时，他会去尝试拿起左手和右手的筷子，如果两只筷子有一只或者两只已经被其他哲学家拿起，那么他就会等待，直到其他人吃完，放下筷子，他开始吃饭。

通过问题的描述，我们可以建立这样一个模型，5个哲学家、5根筷子。哲学家有两个方法，吃饭和休息。筷子也有两个方法，拿起和放下。我们给5个哲学家和筷子分别编号 !~5 那么 1 号哲学家左手是1号筷子，右手是2号筷子。2号哲学家左手是2号筷子，右手是3号筷子，以此类推...5号哲学家左手是5号筷子，右手是1号筷子。

哲学家的吃饭和睡觉状态是随机的，当1号哲学家想要吃饭时，会调用1号筷子和2号筷子的拿起方法，吃完饭后会放下两只筷子。筷子的拿起的方法是上锁的，确保每根筷子每次只能被一个哲学家拿起。如果这时候2号哲学家也想吃饭，它会在尝试拿起2号筷子时被阻塞，但是他可以拿起3号筷子，等到1号哲学家吃完，2号筷子被放下，2号哲学家就可以开始用餐了。以下是该问题的代码实现：

package ThreadTest.cooperation;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

class Chopstick{
    private final int id;
    public Chopstick(int id) {
        this.id = id;
    }
    private boolean taken = false;//是否被拿起
    public synchronized void take() throws InterruptedException {
        while(taken) wait();
        taken = true;
    }

    public synchronized void drop(){
        taken = false;
        notifyAll();
    }
}

class Philosopher implements Runnable{
    private Chopstick left;
    private Chopstick right;
    private final int id;
    private final int ponderFactor;
    private Random rand = new Random(47);
    private void pause() throws InterruptedException {
        if(ponderFactor == 0) return;
        TimeUnit.MILLISECONDS.sleep(rand.nextInt(ponderFactor * 500));
    }
    public Philosopher(Chopstick left,Chopstick right,int ident,int ponder) {
        this.left = left;
        this.right = right;
        this.id = ident;
        this.ponderFactor = ponder;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                System.out.println(this + " thinking" );
                pause();
                left.take();
                System.out.println(this + " grabbing right");
                right.take();
                System.out.println(this + " grabbing left");
                System.out.println(this + " eating");
                pause();
                right.drop();
                System.out.println(this + " drop right");
                left.drop();
                System.out.println(this + " drop left");
            }
        }catch(InterruptedException e) {
            System.out.println(this + " exiting via interrupted");
        }
    }

    public String toString() {
        return "Philosopher : " + id;
    }

}
public class DeadLockingDiningPhilosophers {

    public static void main(String[] args) throws IOException {
        int ponder = 5;
        int size = 5;
        Chopstick[] sticks = new Chopstick[size];
        Philosopher[] philosophers = new Philosopher[size];
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i=0;i<size;i++) {
            sticks[i] = new Chopstick(i);
        }
        for(int i=0;i<size;i++) {
            philosophers[i] = new Philosopher(sticks[i],sticks[(i+1)%size],i,ponder);
            exec.execute(philosophers[i]);
        }
        System.out.println("press Enter to quit");
        System.in.read();
        exec.shutdownNow();
    }

}
//output
/*Philosopher : 0 thinking
press Enter to quit
Philosopher : 3 thinking
Philosopher : 2 thinking
Philosopher : 1 thinking
Philosopher : 4 thinking
Philosopher : 0 grabbing right
Philosopher : 4 grabbing right
Philosopher : 3 grabbing right
Philosopher : 2 grabbing right
Philosopher : 1 grabbing right*/

当哲学家们都拿起自己右手的筷子时，程序就进入了死锁。

关于死锁的发生，有下面4个条件：

互斥条件。任务使用的资源中至少一个资源是不能共享的。
资源不能被任务抢占，即哲学家不能从其他人手里抢夺筷子。
至少有一个任务它必须持有一个资源，且正在等待获取一个当前被另一个任务持有的资源。
必须有循环等待。

那么解决死锁的方案就是使上面4个条件至少一个不成立，而最容易破坏的是第4个条件。在哲学家吃饭的例子里。如果最后一个哲学家不是先拿右手边的筷子，而是先拿左手，就不会产生等待的循环。

package ThreadTest.cooperation;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedDiningPhilosophers {
    public static void main(String[] args) throws IOException {
        int ponder = 5;
        int size = 5;
        Chopstick[] sticks = new Chopstick[size];
        Philosopher[] philosophers = new Philosopher[size];
        ExecutorService exec = Executors.newCachedThreadPool();
        for(int i=0;i<size;i++) {
            sticks[i] = new Chopstick(i);
        }
        for(int i=0;i<size-1;i++) {
            philosophers[i] = new Philosopher(sticks[i],sticks[(i+1)%size],i,ponder);
            exec.execute(philosophers[i]);
        }
        philosophers[size-1] = new Philosopher(sticks[size%size],sticks[(size-1)],size-1,ponder);
        exec.execute(philosophers[size-1]);
        System.out.println("press Enter to quit");
        System.in.read();
        exec.shutdownNow();
    }
}

Thinking in java 之并发其四：线程之间的协作

Thinking in java 之并发其四：线程之间的协作

一、前言

二、wait() 和 notifyAll()

三、notify() 和 notifyAll()

四、经典协作问题

五、使用显示的 Lock 和 Condition 对象

六、生产者——消费者与队列

七、任务间使用管道进行输入/输出

八、死锁

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读