Java并行开发笔记4

不可靠的取消操作将把生产者置于阻塞的操作中

 

	class BrokenPrimeProducer extends Thread{
		private final BlockingQueue<BigInteger> queue;
		private volatile boolean cancelled = false;
		
		BrokenPrimeProducer(BlockingQueue<BigInteger> queue){
			this.queue = queue;
		}
		
		public void run(){
			try{
				BigInteger p = BigInteger.ONE;
				while(!cancelled)
					queue.put(p = p.nextProbablePrime());
			}catch(InterruptedException consumed){
				
			}
		}
		
		public void cancel(){ cancelled = true;}
	}
	
	void consumePrimes() throws InterruptedException {
		BlockingQueue<BigInteger> primes = ...;
		BrokenPrimeProducer producer = new BrokenPrimeProducer(primes);
		producer.start();
		try{
			while(needMorePrimes()){
				consume(primes.take());
			}
		}finally{
			producer.cancel();
		}
	}

 上述代码中，生产者线程生成素数，并将它们放入一个阻塞队列。如果生产者的速度超过了消费者的处理速度，队列将被填满，put方法也会阻塞。当生产者在put方法中阻塞时，如果消费者希望取消生产任务，那么将发生什么情况？它可以调用cancel方法来设置cancelled标志，但此时生产者却永远不能检查这个标志，因为它无法从阻塞的put方法中恢复过来（因为消费者此时已经停止从队列中取出素数，所以put方法将一直保存阻塞状态）。

  为了解决上述问题，可通过线程中断代替取消操作，具体代码如下：

class BrokenPrimeProducer extends Thread{
		private final BlockingQueue<BigInteger> queue;
		
		BrokenPrimeProducer(BlockingQueue<BigInteger> queue){
			this.queue = queue;
		}
		
		public void run(){
			try{
				BigInteger p = BigInteger.ONE;
				while(!Thread.currentThread().isInterrupted())
					queue.put(p = p.nextProbablePrime());
			}catch(InterruptedException consumed){
				
			}
		}
		
		public void cancel(){ interrupt();}
	}