java Thread学习（线程间协作）

线程件协作

线程之间除了使用锁同步两个任务的行为外，还需要进行协作，例如任务A必须在B完成后才能执行。

wait()和notify()/notifyAll()

wait()会去等待外部信号，并且在等待时将自身挂起。等待notify()信号唤醒。
wait()挂起的时候，会释放对象锁，这导致其他的线程可以使用synchronized方法，产生改变，将被挂起的线程唤醒。

wait()/wait(t)：使用wait将挂起线程，可以无限等待，也可以指定时间，超时后恢复执行
可以通过notify()/notifyAll()，或者时间到期后

注意：

这些方法都是针对于对象来说的，wait()释放对象锁，挂起当前的线程。notify()在执行完同步代码块后释放锁，然后所有wait对象锁的线程中随机唤醒一个
wait/notify/notifyAll都是Object中的方法。但是只能在同步控制方法或者同步块中使用。否则运行报错IIIegalMonitorStateException。
notify()唤醒一个线程/notifyAll()唤醒所有对象锁。notify()有可能产生死锁。所以尽量使用notifyAll()

使用另一个对象唤醒自身时，需要先获取对象锁。例如

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

举例：给车打蜡+抛光，打蜡后才能抛光，抛光完成后继续打蜡

class Car{private boolean waxOn = false;public synchronized void waxed() {waxOn = true;/*** do something*/notifyAll();}public synchronized void buffed() {waxOn = false;/*** do ...*/notifyAll();}public synchronized void waitForWaxing() throws InterruptedException {while(waxOn == false) {//唤醒是随机的，所以可能依然需要waitwait();}}public synchronized void waitForBuffing() 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!");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 car) {this.car = car;}public void run() {try{while(!Thread.interrupted()) {System.out.println("Wax Off");car.buffed();car.waitForWaxing();}} 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 InterruptedException {Car car = new Car();ExecutorService service = Executors.newCachedThreadPool();service.execute(new WaxOn(car));service.execute(new WaxOff(car));TimeUnit.SECONDS.sleep(5);service.shutdownNow();}
}

Condition对象

在java.util.concurrent类库中包含了Condition类。await()和signal()/signalAll()也可以完成协调的作用。
java Lock lock = new ReentrantLock(); Condition condition = lock.newCondition();//通过使用一个锁来获得Condition condition.await(); condition.signal(); condition.signalAll();
注意：

每一个Lock都应该在try-finally中，保证任何情况都可以释放锁。
任务使用await()或者signal()前必须进行加锁。

notify()/notifyAll()

notify()仅唤醒一个线程，可能出现死锁。举例生产者P-消费者C问题。引用自http://blog.csdn.net/tayanxunhua/article/details/20998809

第一步：P1放入一个对象到buffer中；

第二步：P2试图put一个对象，此时buf中已经有一个了，所以wait

第三步：P3试图put一个对象，仍然wait

第四步：

C1试图从buf中获得一个对象；
C2试图从buf中获得一个对象，但是挡在了get方法外面
C3试图从buf中获得一个对象，同样挡在了get方法外面

第五步：C1执行完get方法，执行notify，退出方法。notify唤醒了P2，但是C2在P2唤醒之前先进入了get方法，所以P2必须再次获得锁，P2被挡在了put方法的外面，C2循环检查buf大小，在buf中没有对象，所以只能wait；C3在C2之后，P2之前进入了方法，由于buf中没有对象，所以也wait；

第六步：现在，有P3，C2，C3在waiting；最后P2获得了锁，在buf中放入了一个对象，执行notify，退出put方法；

第七步：notify唤醒P3；P3检查循环条件，在buf中已经有了一个对象，所以wait；现在没有线程能够notify了，三个线程就会处于死锁状态。

生产者-消费者

饭店呢的一个示例：厨师需要做饭，服务员需要端菜。饭店包含了厨师，服务员，食物。厨师、服务员、食物都包含饭店，通过饭店操纵其他的对象。

```java
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;}public void run() {try {while(!Thread.interrupted()) {synchronized (this) {while(restaurant.getMeal() == null) {wait();}}System.out.println("WaitPerson got " + restaurant.getMeal());synchronized (restaurant.getChef()) {while(restaurant.getMeal() != null) {restaurant.setMeal(null);restaurant.getChef().notifyAll();}}}} catch(InterruptedException e) {System.out.println("WaitPerson Interrupted");}}
}
class Chef implements Runnable {private Restaurant restaurant;private volatile int count = 0;public Chef(Restaurant restaurant) {this.restaurant = restaurant;}public void run() {try {while(!Thread.interrupted()) {synchronized(this) {while(restaurant.getMeal() != null) {wait();}}if(++count == 10) {System.out.println("Out of food, closing");restaurant.getExecutorService().shutdownNow();}System.out.println("Order up!");synchronized (restaurant.getWaitPerson()) {restaurant.setMeal(new Meal(count));restaurant.getWaitPerson().notifyAll();}TimeUnit.MILLISECONDS.sleep(100);}} catch(InterruptedException e) {System.out.println("Chef Interrupted");}}
}
public class Restaurant {private Chef chef;private WaitPerson waitPerson;private Meal meal;private ExecutorService service;public Restaurant() {service = Executors.newCachedThreadPool();chef = new Chef(this);waitPerson = new WaitPerson(this);service.execute(chef);service.execute(waitPerson);}public static void main(String[] args) {new Restaurant();}public void setChef(Chef chef) {this.chef = chef;}public Chef getChef() {return chef;}public void setWaitPerson(WaitPerson waitPerson) {this.waitPerson = waitPerson;}public WaitPerson getWaitPerson() {return waitPerson;}public void setMeal(Meal meal) {this.meal = meal;}public Meal getMeal() {return meal;}public void setExecutorService(ExecutorService service) {this.service = service;}public ExecutorService getExecutorService() {return service;}
}
```

生产者-消费者与队列

同步队列：java.util.concurrent.BlockingQueue接口提供了同步队列，如果消费者从队列中获取对象，而且队列中为空，那么线程就会被挂起。
- LinkedBlockingQueue：不限定队列大小
- ArrayBlockingQueue：需要限定队列的大小
- SynchronizedBlockingQueue：默认为1的队列

常用方法	应用
add(Obj)	放入一个Object，成功返回true，否则返回异常
offer(Obj)	放入一个Object，成功返回true，否则返回false
put(Obj)	放入一个Object，成功返回true，否则挂起线程，有空间时继续
poll(Obj)	取出队首的元素，如果不能立即取出，则等待指定时间，取不到返回null
take(Obj)	取出队首的元素，如果不能立即取出，挂起线程，等待存在数据后再继续

```java
class Toast {public enum Status { DRY, BUTTERED, JAMMED}public Status status = Status.DRY;public final int id;public Toast(int id) { this.id = id; }public void butter() { status = Status.BUTTERED; }public void jam() { status = Status.JAMMED; }public Status getStatus() { return status; }public int getId() { return id; }public String toString() {return "Toast " + id + ": " + status;}
}
class ToastQueue extends LinkedBlockingQueue<Toast> {}
class Toaster implements Runnable {private ToastQueue dryQueue;private int count = 0;public Toaster(ToastQueue dryQueue) {this.dryQueue = dryQueue;}public void run() {try {while(!Thread.interrupted()) {TimeUnit.MILLISECONDS.sleep(100);Toast t = new Toast(++count);System.out.println("Toaster: " + t);dryQueue.put(t);}} catch(InterruptedException e) {System.out.println("Toaster interrupted");}}
}
class Butterer implements Runnable {private ToastQueue dryQueue, butteredQueue;public Butterer(ToastQueue dryQueue, ToastQueue butteredQueue) {this.dryQueue = dryQueue;this.butteredQueue = butteredQueue;}public void run() {try {while(!Thread.interrupted()) {Toast t = dryQueue.take();t.butter();System.out.println("Butter: " + t);butteredQueue.put(t);}} catch(InterruptedException e) {System.out.println("Butterer Interrupted");}}
}
class Jammer implements Runnable {private ToastQueue butteredQueue, finishedQueue;public Jammer(ToastQueue butteredQueue, ToastQueue finishedQueue) {this.butteredQueue = butteredQueue;this.finishedQueue = finishedQueue;}public void run() {try {while(!Thread.interrupted()) {Toast t = butteredQueue.take();t.jam();System.out.println("Jammer: " + t);finishedQueue.put(t);}} catch(InterruptedException e) {System.out.println("Butterer Interrupted");}}
}
public class ToastOMatic {public static void main(String[] args) throws InterruptedException {ToastQueue dryQueue = new ToastQueue();ToastQueue butteredQueue = new ToastQueue();ToastQueue finishedQueue = new ToastQueue();ExecutorService service = Executors.newCachedThreadPool();service.execute(new Toaster(dryQueue));service.execute(new Butterer(dryQueue, butteredQueue));service.execute(new Jammer(butteredQueue, finishedQueue));TimeUnit.SECONDS.sleep(3);service.shutdownNow();}
}
```

管道输入/输出

使用PipedReader,PipedWriter创建一个管道，当PipedReader读不到数据时，管道就会自动阻塞(允许不同的任务从同一个管道中读取)

```java
class Sender implements Runnable {private Random rand = new Random(47);private PipedWriter out = new PipedWriter();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("Sender write exception\n" + e);} catch(InterruptedException e) {System.out.println("Sender write exception\n" + e);}}public PipedWriter getPipedWriter() {return out;}
}
class Receiver implements Runnable {private PipedReader in;public Receiver(Sender s) throws IOException {in = new PipedReader(s.getPipedWriter());}public void run() {try {while(true) {System.out.println("Read: " + (char)in.read());}} catch (IOException e) {System.out.println("Receiver read exception\n" + e);}}
}
public class PipedIO {public static void main(String[] args) throws Exception {Sender sender = new Sender();Receiver receiver = new Receiver(sender);ExecutorService service = Executors.newCachedThreadPool();service.execute(sender);service.execute(receiver);TimeUnit.SECONDS.sleep(4);service.shutdownNow();}
}
```