线程协作

/**
 * 线程当中的协作，汽车打蜡-清洗Demo
 * @author pengchengliu
 *
 */
public class WaxOMatic {
 	public static void main(String[] args) throws InterruptedException {
 		Car car = new Car();
 		ExecutorService exec = Executors.newCachedThreadPool();
 		exec.execute(new WaxOn(car));
 		exec.execute(new Buff(car));
 		TimeUnit.SECONDS.sleep(5);
 		exec.shutdownNow();
 	}
}
class Car {
 	private boolean isWaxFinish = false ;
 	
 	public synchronized void waxOn () {
 		isWaxFinish = true ;
 	}
 	
 	public synchronized void buffed () {
 		isWaxFinish = false ;
 		notifyAll();
 	}
 	
 	public synchronized void waitForWaxing () throws InterruptedException {
 		while (!isWaxFinish) {
 			wait();
 		}
 	}
 	
 	public synchronized void waitForBuffing () throws InterruptedException {
 		while (isWaxFinish) {
 			wait();
 		}
 	}
}
class WaxOn implements Runnable {
 	private Car car ;
 	
 	public WaxOn (Car car) {
 		this.car = car ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (!Thread.interrupted()){
 				System.out.println("Wax on!");
 				TimeUnit.MILLISECONDS.sleep(200);
 				car.waxOn();
 				car.waitForBuffing();
 			}
 		} catch (InterruptedException e) {
 			System.out.println("exit via interrupt");
 		}
 		System.out.println("ending wax On task");
 	}
}
class Buff implements Runnable{
 	private Car car ;
 	
 	public Buff (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){
 			System.out.println("exit via interrupt");
 		}
 		System.out.println("ending Wax off");
 	}

}
Wax on!
Wax Off
Wax on!
Wax Off
Wax on!
Wax Off
Wax on!
Wax Off
Wax on!
Wax Off
Wax on!

T1
synchorized(lockBoject){
    someCondition = false ;
    lockBoject.notify();
}


T2
someCondition = true ；
while(someCondition){
    //setup01
    synchorized(lockBoject){
        wait();
    }
}

T1
synchorized(lockBoject){
    someCondition = false ;
    lockBoject.notify();
}


T2
someCondition = true ;
//setup01
synchorized(lockBoject){
    while(someCondition){
        wait();
    }
}

package com.suansuan.cooperation;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 对比Notify和NotifyAll
 * notify用得不好容易导致死锁.
 * @author pengchengliu
 *
 */
public class NotifyVsNotifyAll {
 	public static void main(String[] args) throws InterruptedException {
 		ExecutorService executorService = Executors.newCachedThreadPool();
 		for (int i = 0; i < 5; i++){
 			executorService.execute(new Task01());
 		}
 		executorService.execute(new Task02());
 		
 		Timer timer = new Timer();
 		timer.scheduleAtFixedRate(new TimerTask(){ 
 			boolean prod = true ;
 			public void run() {
 				if (prod) {
 					System.out.println("notify()");
 					Task01.blocker.prod();
 					prod = false ;
 				} else {
 					System.out.println("notifyAll()");
 					Task01.blocker.prodAll();
 					prod = true ;
 				}
 			}
 		}, 400, 400);
 		
 		
 		TimeUnit.SECONDS.sleep(2);
 		timer.cancel();
 		System.out.println("timer is cancel");
 		
 		TimeUnit.MILLISECONDS.sleep(500);
 		System.out.println("Task02 blocker prodAll()");
 		Task02.blocker.prodAll();
 		
 		TimeUnit.MILLISECONDS.sleep(500);
 		System.out.println("shutting down");
 		executorService.shutdownNow();
 	}
}
class Blocker {
 	synchronized void waitingCall () {
 		try{
 			while (!Thread.interrupted())  {
 				wait() ;
 				System.out.println(Thread.currentThread() + "");
 			}
 		} catch (InterruptedException e) {}
 	}
 	
 	synchronized void prod () {
 		notify();
 	}
 	
 	synchronized void prodAll () {
 		notifyAll();
 	}
}
class Task01 implements Runnable {
 	static Blocker blocker = new Blocker();
 	@Override
 	public void run() {
 		blocker.waitingCall();
 	}
}
class Task02 implements Runnable {
 	static Blocker blocker = new Blocker();
 	public void run() {
 		blocker.waitingCall();
 	}
}

notify()
Thread[pool-1-thread-1,5,main]
notifyAll()
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]
Thread[pool-1-thread-2,5,main]
notify()
Thread[pool-1-thread-1,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]
timer is cancel
Task02 blocker prodAll()
Thread[pool-1-thread-6,5,main]
shutting down

package com.suansuan.cooperation;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 最原始的生产者-消费者模型，不具有相应的共享队列
 * @author pengchengliu
 *
 */
public class Restaurant {
 	public Meal meal ;
 	public Consumer consumer = new Consumer(this);
 	public Chef chef = new Chef(this);
 	public ExecutorService exec = Executors.newCachedThreadPool();
 	
 	public Restaurant(){
 		exec.execute(chef);
 		exec.execute(consumer);
 	}
 	
 	public static void main(String[] args) {
 		new Restaurant();
 	}
}
/**
 * 饭菜，也就是我们需要在两个协同任务当中传递的东西
 * @author pengchengliu
 *
 */
class Meal {
 	private final int orderNum ;	//订购的数量
 	public Meal (int orderNumber) {
 		this.orderNum = orderNumber ;
 	}
 	public String toString () {
 		return "Meal " + this.orderNum ;
 	}
}
/**
 * 厨师，生产者，
 * 注意：生产者的唤醒服务员的时机, 还有生产等待的时机
 * @author pengchengliu
 *
 */
class Chef implements Runnable {
 	private Restaurant restaurant ;
 	private int count ;
 	public Chef (Restaurant restaurant) {
 		this.restaurant = restaurant ;
 	}
 	//run方法当中，要确保正确的关闭方式。
 	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();
 				}
 				
 				// 因为服务员就是在这个锁上面等待的，所以我们获取服务员的锁，来唤醒所有等待服务员锁的对象
 				synchronized (restaurant.consumer) {
 					restaurant.meal = new Meal(count);
 					System.out.println("Chef output " + restaurant.meal);
 					restaurant.consumer.notifyAll();
 				}
 				
 				TimeUnit.MILLISECONDS.sleep(100);
 			}
 		} catch (Exception e) {
 			System.out.println("Chef interrupt");
 		}
 	}
}
/**
 * 服务员-消费者
 * 注意：等待的条件，唤醒条件，以及唤醒的时候拿到的锁。
 * @author pengchengliu
 *
 */
class Consumer implements Runnable {
 	private Restaurant restaurant ;
 	public Consumer(Restaurant restaurant) {
 		this.restaurant = restaurant ;
 	}
 	public void run (){
 		try{
 			while (!Thread.interrupted()){
 				synchronized (this) {
 					while (restaurant.meal == null) {
 						wait();
 					}
 				}
 				
 				System.out.println("Consumer input " + restaurant.meal);
 				
 				synchronized (restaurant.chef) {
 					restaurant.meal = null ;
 					restaurant.chef.notifyAll();
 				}
 			}
 		} catch (Exception e) {
 			System.out.println("consumer interrupt");
 		}
 	}
}
Chef output Meal 1
Consumer input Meal 1
Chef output Meal 2
Consumer input Meal 2
Chef output Meal 3
Consumer input Meal 3
Chef output Meal 4
Consumer input Meal 4
Chef output Meal 5
Consumer input Meal 5
Chef output Meal 6
Consumer input Meal 6
Chef output Meal 7
Consumer input Meal 7
Chef output Meal 8
Consumer input Meal 8
Chef output Meal 9
Consumer input Meal 9
Out of food close
consumer interrupt
Chef output Meal 10
Chef interrupt

/**
 * 线程当中的协作，汽车打蜡-清洗Demo 使用显示的Lock和Condition来控制
 * @author pengchengliu
 *
 */
public class WaxOMaticLockCondition {
 	public static void main(String[] args) throws InterruptedException {
 		Cars car = new Cars();
 		ExecutorService exec = Executors.newCachedThreadPool();
 		exec.execute(new Waxon2(car));
 		exec.execute(new Buffed(car));
 		TimeUnit.SECONDS.sleep(5);
 		exec.shutdownNow();
 	}
}
class Buffed implements Runnable {
 	private Cars car ;
 	public Buffed (Cars car) {
 		this.car = car ;
 	}
 	@Override
 	public void run() {
 		try {
 			while (!Thread.interrupted()) {
 				System.out.println("Buffed on");
 				TimeUnit.MILLISECONDS.sleep(200);
 				car.buffed();
 				car.waitForWaxing();
 			}
 		} catch (InterruptedException e) {
 			System.out.println("exit via interrupt");
 		}
 		System.out.println("ending buffed on task");
 	}
}
class Waxon2 implements Runnable {
 	private Cars car ;
 	public Waxon2 (Cars car) {
 		this.car = car ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (!Thread.interrupted()) {
 				System.out.println("Wax on");
 				TimeUnit.MILLISECONDS.sleep(200);
 				car.waxed();
 				car.waitForBuffed();
 			}
 		} catch (InterruptedException e) {
 			System.out.println("exit via interrupt");
 		}
 		System.out.println("ending wax on task");
 	}
}
/**
 * 共享资源类
 * @author pengchengliu
 *
 */
class Cars {
 	private	Lock lock = new ReentrantLock();
 	private Condition condition = lock.newCondition();
 	private boolean isWaxFinish = false ;
 	
 	/**
 	 * 使用try的原因就是怕在显示的锁  不能被释放，所以这么写，确保锁可以被正确的释放掉。
 	 */
 	public void waxed () {
 		lock.lock();
 		try {
 			isWaxFinish = true ;
 			condition.signalAll(); 
 		} finally {
 			lock.unlock();
 		}
 	}
 	
 	public void buffed () {
 		lock.lock();
 		try{
 			isWaxFinish = false ;
 			condition.signalAll();
 		} finally {
 			lock.unlock();
 		}
 	}
 	
 	public void waitForWaxing () throws InterruptedException {
 		lock.lock();
 		try{
 			while (!isWaxFinish) {
 				condition.await();
 			}
 		} finally {
 			lock.unlock();
 		}
 	}
 	
 	public void waitForBuffed () throws InterruptedException {
 		lock.lock();
 		try{
 			while (isWaxFinish) {
 				condition.await();
 			}
 		} finally {
 			lock.unlock();
 		}
 	}
}
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
Wax on
Buffed on
exit via interrupt
ending wax on task
exit via interrupt
ending buffed on task

/**
 * ToastBlockingQueue, 
 * 有一台机器一共具有三个任务，分别是
 * 1、制作吐司，
 * 2、给吐司上黄油、
 * 3、给上了黄油的吐司上果酱。
 * 
 * 我们可以通过各个处理过程之间的BlockingQueue来运行这个吐司制作的程序。
 * @author pengchengliu
 *
 */
public class ToastOMatic {
 	
 	/**
 	 * 开启所有的制作Toast的相关任务，然后运行5秒钟，结束所有的任务
 	 * @param args
 	 * @throws InterruptedException
 	 */
 	public static void main(String[] args) throws InterruptedException {
 		ToastQueue dryQueue = new ToastQueue(),
 				butterQueue = new ToastQueue(),
 				finishQueue = new ToastQueue();
 		
 		ExecutorService exec = Executors.newCachedThreadPool();
 		exec.execute(new Toaster(dryQueue));
 		exec.execute(new Butter(dryQueue, butterQueue));
 		exec.execute(new Jammer(butterQueue, finishQueue));
 		exec.execute(new Eater(finishQueue));
 		
 		TimeUnit.SECONDS.sleep(5);
 		
 		exec.shutdownNow();
 	}
}
/**
 * Toast本身，也就是我们的资源共享类，
 * 思考：为什么关于资源类的状态改变没有加锁，这个类被三个任务所操作，不加锁安全么？
 * @author pengchengliu
 *
 */
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 getStatus () {
 		return this.status ;
 	}
 	
 	public int getId () {
 		return this.id ;
 	}
 	
 	public String toString () {
 		return "Toast " + id + " : " + status ;
 	}
 	
 	public void recyleToast () {
 		this.status = null ;
 	}
}
/**
 * 使用定制的相关队列，如果有特殊的需求，通过继承LinkedBlockingQueue来做自定义
 * 遵守单一职责的原理，应该定制单一的属于队列的相关需求，
 * 这里只是演示Demo，所以没有做任何定制服务。
 * @author pengchengliu
 *
 */
class ToastQueue extends LinkedBlockingQueue<Toast> {}
/**
 * 第一步，制作Toast，将制作出来的Toast至于共享队列当中。
 * 注意：只有一个共享队列
 * @author pengchengliu
 *
 */
class Toaster implements Runnable {
 	
 	private ToastQueue toastQueue ;
 	private int count ;
 	private Random rand = new Random(47);
 	
 	public Toaster (ToastQueue toastQueue) {
 		this.toastQueue = toastQueue ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (! Thread.interrupted()) {
 				TimeUnit.MILLISECONDS.sleep(100 + rand.nextInt(500));
 				Toast toast = new Toast(count ++) ;
 				System.out.println(toast);
 				toastQueue.add(toast);
 			}
 		} catch (InterruptedException e) {
 			System.out.println("Toaster interrupt");
 		}
 		System.out.println("Toaster off");
 	}
}
/**
 * 第二步，上黄油，
 * 注意:需要两个队列，一个是输入队列，一个是输出队列
 * @author pengchengliu
 *
 */
class Butter implements Runnable {
 	
 	private	ToastQueue dryQueue, butterQueue ;
 	
 	public Butter (ToastQueue dryQueue, ToastQueue butterQueue) {
 		this.dryQueue = dryQueue ;
 		this.butterQueue = butterQueue ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (! Thread.interrupted()) {
 				Toast take = dryQueue.take();
 				take.butter();
 				System.out.println(take);
 				butterQueue.add(take);
 			} 
 		} catch (InterruptedException e) {
 			System.out.println("Butter interrupt");
 		}
 		System.out.println("Butter off");
 	}
}
/**
 * 第三步:上果酱,基本操作与步骤二相似
 * @author pengchengliu
 *
 */
class Jammer implements Runnable {
 	
 	private	ToastQueue butterQueue, finishedQueue ;
 	
 	public Jammer (ToastQueue butterQueue, ToastQueue finishedQueue) {
 		this.butterQueue = butterQueue ;
 		this.finishedQueue = finishedQueue ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (!Thread.interrupted()) {
 				Toast take = butterQueue.take();
 				take.jam();
 				System.out.println(take);
 				finishedQueue.add(take);
 			}
 		} catch (InterruptedException e) {
 			System.out.println("Jammer interrupt");
 		}
 		System.out.println("Jammer off");
 	}
}
/**
 * 最后的消费者
 * @author pengchengliu
 *
 */
class Eater implements Runnable {
 	
 	private ToastQueue finishedQueue ;
 	private int counter = 0 ;
 	
 	public Eater (ToastQueue finishedQueue) {
 		this.finishedQueue = finishedQueue ;
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (!Thread.interrupted()) {
 				Toast take = finishedQueue.take();
 				if (take.getId() != counter ++ || take.getStatus() != Toast.Status.JAMMED) {
 					System.out.println(">>> Error: " + take);
 					System.exit(1);
 				} else {
 					System.out.println("chomp! " + take);
 					take.recyleToast();
 					take = null;
 				}
 			}
 		} catch (InterruptedException e) {
 			System.out.println("Eater interrupt");
 		}
 		System.out.println("Eater off");
 	}
}
Toast 0 : DRY
Toast 0 : BUTTERED
Toast 0 : JAMMED
chomp! Toast 0 : JAMMED
Toast 1 : DRY
Toast 1 : BUTTERED
Toast 1 : JAMMED
chomp! Toast 1 : JAMMED
Toast 2 : DRY
Toast 2 : BUTTERED
Toast 2 : JAMMED
chomp! Toast 2 : JAMMED
Toast 3 : DRY
Toast 3 : BUTTERED
Toast 3 : JAMMED
chomp! Toast 3 : JAMMED
Toast 4 : DRY
Toast 4 : BUTTERED
Toast 4 : JAMMED
chomp! Toast 4 : JAMMED
Toast 5 : DRY
Toast 5 : BUTTERED
Toast 5 : JAMMED
chomp! Toast 5 : JAMMED
Toast 6 : DRY
Toast 6 : BUTTERED
Toast 6 : JAMMED
chomp! Toast 6 : JAMMED
Toast 7 : DRY
Toast 7 : BUTTERED
Toast 7 : JAMMED
chomp! Toast 7 : JAMMED
Toast 8 : DRY
Toast 8 : BUTTERED
Toast 8 : JAMMED
chomp! Toast 8 : JAMMED
Toast 9 : DRY
Toast 9 : BUTTERED
Toast 9 : JAMMED
chomp! Toast 9 : JAMMED
Toast 10 : DRY
Toast 10 : BUTTERED
Toast 10 : JAMMED
chomp! Toast 10 : JAMMED
Toast 11 : DRY
Toast 11 : BUTTERED
Toast 11 : JAMMED
chomp! Toast 11 : JAMMED
Toast 12 : DRY
Toast 12 : BUTTERED
Toast 12 : JAMMED
chomp! Toast 12 : JAMMED
Toast 13 : DRY
Toast 13 : BUTTERED
Toast 13 : JAMMED
chomp! Toast 13 : JAMMED
Toast 14 : DRY
Toast 14 : BUTTERED
Toast 14 : JAMMED
chomp! Toast 14 : JAMMED
Toaster interrupt
Toaster off
Butter interrupt
Butter off
Eater interrupt
Eater off
Jammer interrupt
Jammer off

/**
 * 探究Piped 管道 控制任务间的协作
 * @author pengchengliu
 *
 */
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() ;
 	}
}
/**
 * 发送任务
 * @author pengchengliu
 *
 */
class Sender implements Runnable {
 	
 	private	Random rand = new Random(47);
 	private PipedWriter out = new PipedWriter();
 	
 	public PipedWriter getPepedWriter() {
 		return this.out ;
 	}
 	
 	@Override
 	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 writer execption");
 		} catch (InterruptedException e) {
 			System.out.println("Sender interrput");
 		}
 	}
}
/**
 * 接收任务
 * @author pengchengliu
 *
 */
class Receiver implements Runnable {
 	
 	private PipedReader reader ;
 	
 	public Receiver(Sender sender) throws IOException {
 		reader = new PipedReader(sender.getPepedWriter());
 	}
 	
 	@Override
 	public void run() {
 		try {
 			while (true) {
 				System.out.println("Read : " + (char) reader.read() + ".");
 			}
 		} catch (IOException e) {
 			System.out.println(e + "Receiver read exception");
 		}
 	}

}
Read : A.
Read : B.
Read : C.
Read : D.
Read : E.
Read : F.
Read : G.
Read : H.
Read : I.
Read : J.
Read : K.
Read : L.
Read : M.
Sender interrput
java.io.InterruptedIOExceptionReceiver read exception