JAVA并发-中断处理和任务取消

中断处理
在java程序中，当使用Thread.sleep()或者BlockingQueue.take()等阻塞方法时，需要处理InterruptedException。对于这种异常，通常有2种方案进行处理。
1. 传递异常：将异常传递给方法的调用者。示例如下：

BlockingQueue<String> queue;
public String getNextString() throws InterruptedException{
		return queue.take();
} 

2. 恢复异常：在大多数情况下，异常是可以传递的，但有些情况是无法传递异常的，比如在Runnable的run()方法中，我们不可以抛出异常。此时需要我们恢复异常，这样在调用栈中更高层次的代码将看到引发了一个中断。示例如下：

BlockingQueue<String> queue;
public String getNextString() throws InterruptedException{
		String result=null;
     try{
			result=queue.take();
}catch(InterruptedException e){
	Thead.currentThread.interrupt();
}finally{
	Return result;
}
} 

以上是处理InterruptedException的常用方法，对于InterruptedException，千万不要捕获异常但不做任何处理。当我们使用中断来结束线程时，在catch块中也可以使用interrupted()来清除异常。

任务取消
多线程编程时，有时候需要取消某些任务线程，有以下3种方案：
1. 设定一个线程取消的标记，任务线程定期的检查这个标记。示例如下：

class Task implements Runnable{
	private volatile boolean cancel=false;
	@Override
	public void run() {
		while(!cancel){
			System.out.println("...");
		}
	}
	public void cancel(){
		cancel=true;
	}
}

2. 上面的示例描述的是最一般的场景，试想一下，如果while(!cancel)循环中调用了一个阻塞的方法，那么有这样一种可能：程序可能阻塞在某个方法中。示例如下：

class Task implements Runnable{
	private volatile boolean cancel=false;
	private BlockingQueue<String> blockingQueue;
	public Task(BlockingQueue<String> queue){
		this.blockingQueue=queue;
	}
	@Override
	public void run() {
		try{
			while(!cancel){
				System.out.println("...");
				this.blockingQueue.take();//当程序阻塞在此处时，即便cancel被更新了，也无法感知，这种情况下，程序永远无法退出。
			}			
		}catch(InterruptedException e){
			Thread.currentThread().interrupt();
		}
	}
	public void cancel(){
		cancel=true;
	}
}

当while(!cancel)循环中调用了一个阻塞的方法时，使用标记位的方式终止程序就不再使用了，此时使用中断的方式退出程序：

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class CDemo {
        public static void main(String [] args) throws Exception{
        	BlockingQueue<String> b=new ArrayBlockingQueue<String>(3);
        	b.put("a");
        	b.put("ab");
        	b.put("abc");
        	
        	Task task=new Task(b);
        	task.start();
        	Thread.sleep(4000);
        	
        	task.cancel();
        }
}

class Task extends Thread{  
    private BlockingQueue<String> blockingQueue;  
    public Task(BlockingQueue<String> queue){  
        this.blockingQueue=queue;  
    }  
    @Override  
    public void run() {  
        try{  
        	while(true){
//        		if(Thread.currentThread().isInterrupted()) //一定注意，这行是错误做法
        		if(interrupted())//判断当前线程是否被中断
        			break;
                String str=this.blockingQueue.take(); 
                System.out.println(str);
            }  
        }catch(InterruptedException e){  
            //Thread.currentThread().interrupt();  
        	interrupted();//清除中断痕迹
        }finally{
        	System.out.println("线程结束！");
        }
    }  
    public void cancel(){  
        this.interrupt();//中断当前线程
//    	Thread.currentThread().interrupt(); //一定注意，这行是错误做法
    }  
}  

3. 在生产者消费者问题中，使用“毒丸对象”来终止消费者线程。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

public class Coordinator {
	public static final Object POISON_PILL = new Object();//special object to kill consumers
	private int productCount = 1;
	private int consumerCount = 3;

	public void startAll() throws Exception{
		BlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(5);
		CountDownLatch activeProductorNum = new CountDownLatch(productCount);
		CountDownLatch activeConsumerNum = new CountDownLatch(consumerCount);

		for(int i = 0; i < consumerCount; i++){
			new Thread(new Consumer("consumer " + i, queue,activeConsumerNum)).start();
		}

		for(int i = 0; i < productCount; i++){
			new Thread(new Producer("producer " + i, queue, activeProductorNum)).start();
		}
		
		activeProductorNum.await();//等待所有生产者生产结束
		System.out.println("All producer finished, putting POISON_PILL to the queue to stop consumers!");
		queue.put(POISON_PILL);
		
		activeConsumerNum.await();//等待所有生产者生产结束
		System.out.println("All consumer finished!");
	}
	
	public static void main(String[] args) throws Exception{
		new Coordinator().startAll();
	}
}

class Producer implements Runnable {
	private String name;
	private BlockingQueue<Object> queue;
	private CountDownLatch activeProducerNum;
	
	public Producer(String name, BlockingQueue<Object> queue, CountDownLatch activeProducerNum){
		this.name = name;
		this.queue = queue;
		this.activeProducerNum=activeProducerNum;
	}

	@Override
	public void run() {
		try {
			for(int i=0;i<10;i++){
				queue.put(i);
				System.out.println(name + " produced "+i);				
			}
				
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} finally{
			System.out.println(name + " finished.");
			activeProducerNum.countDown();
		}
	}
}


class Consumer implements Runnable {
	private String name;
	private BlockingQueue<Object> queue;
	private CountDownLatch activeConsumerNum;

	public Consumer(String name, BlockingQueue<Object> queue,CountDownLatch activeConsumerNum){
		this.name = name;
		this.queue = queue;
		this.activeConsumerNum=activeConsumerNum;
	}

	@Override
	public void run() {
		try {
			while (true) {
				Object item = queue.take();
				if (item == Coordinator.POISON_PILL) {
					queue.put(item);//放回继续毒害其他消费者
					break;
				}
				System.out.println(name + " consumed "+item);
			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} finally{
			System.out.println(name + " finished");
			activeConsumerNum.countDown();
		}
	}
}