论坛首页 Java企业应用论坛

线程任务的取消

浏览 7822 次
精华帖 (7) :: 良好帖 (0) :: 新手帖 (0) :: 隐藏帖 (0)
作者 正文
   发表时间:2007-09-03  
    当外部代码能够在活动自然完成之前,把它的状态更改为完成状态,那么这个活动被称为可取消(cancellable)。取消任务是一个很常见的需求,无论 是由于用户请求还是系统错误引起的服务关闭等等原因。最简单的任务取消策略就是在线程中维持一个bool变量,在run方法中判断此变量的bool值来决 定是否取消任务。显然,这个bool变量需要声明为volatile,以保持多线程环境下可见性(所谓可见性,就是当一个线程修改共享对象的状态变量后,另一个线程 可以马上看到修改的结果)。下面是一个来自《java并发编程实践》的例子:
java 代码
 
  1. package net.rubyeye.concurrency.chapter7;  
  2.   
  3. import java.math.BigInteger;  
  4. import java.util.ArrayList;  
  5. import java.util.List;  
  6. import java.util.concurrent.TimeUnit;  
  7.   
  8. public class PrimeGenerator implements Runnable {  
  9.     private final List<biginteger> primes = </biginteger>new ArrayList<biginteger>();  </biginteger>
  10.   
  11.     private volatile boolean cancelled;  
  12.     public void run() {  
  13.         BigInteger p = BigInteger.ONE;  
  14.         while (!cancelled) {  
  15.             p = p.nextProbablePrime();  
  16.             synchronized (this) {  
  17.                 primes.add(p);  
  18.             }  
  19.         }  
  20.     }  
  21.     public void cancel() {  
  22.         cancelled = true;  
  23.     }  
  24.     public synchronized List<biginteger> get() {  </biginteger>
  25.         return new ArrayList<biginteger>(primes);  </biginteger>
  26.     }  
  27.   
  28.     public static void main(String args[]) throws InterruptedException {  
  29.         PrimeGenerator generator = new PrimeGenerator();  
  30.         new Thread(generator).start();  
  31.         try {  
  32.             TimeUnit.SECONDS.sleep(1);  
  33.         } finally {  
  34.             generator.cancel();  
  35.         }  
  36.     }  
  37. }  

    main中启动一个素数生成的任务,线程运行一秒就取消掉。通过线程中的cancelled变量来表征任务是否继续执行。既然是最简单的策略,那么什么是 例外情况?显然,阻塞操作下(比如调用join,wait,sleep方法),这样的策略会出问题。任务因为调用这些阻塞方法而被阻塞,它将不会去检查 volatile变量,导致取消操作失效。那么解决办法是什么?中断!考虑我们用BlockingQueue去保存生成的素数, BlockingQueue的put方法是阻塞的(当BlockingQueue满的时候,put操作会阻塞直到有元素被take),让我们看看不采用中 断,仍然采用简单策略会出现什么情况:
java 代码
 
  1. package net.rubyeye.concurrency.chapter7;  
  2.   
  3. import java.math.BigInteger;  
  4. import java.util.concurrent.BlockingQueue;  
  5. import java.util.concurrent.CountDownLatch;  
  6. import java.util.concurrent.LinkedBlockingQueue;  
  7. import java.util.concurrent.TimeUnit;  
  8.   
  9. public class BrokenPrimeProducer extends Thread {  
  10.     static int i = 1000;  
  11.   
  12.     private final BlockingQueue<biginteger> queue;  </biginteger>
  13.   
  14.     private volatile boolean cancelled = false;  
  15.   
  16.     BrokenPrimeProducer(BlockingQueue<biginteger> queue) {  </biginteger>
  17.         this.queue = queue;  
  18.     }  
  19.   
  20.     public void run() {  
  21.         BigInteger p = BigInteger.ONE;  
  22.         try {  
  23.             while (!cancelled) {  
  24.                 p = p.nextProbablePrime();  
  25.                 queue.put(p);  
  26.             }  
  27.         } catch (InterruptedException cusumed) {  
  28.         }  
  29.     }  
  30.   
  31.     public void cancel() {  
  32.         this.cancelled = false;  
  33.     }  
  34.   
  35.     public static void main(String args[]) throws InterruptedException {  
  36.         BlockingQueue<biginteger> queue = new LinkedBlockingQueue<biginteger>(  </biginteger></biginteger>
  37.                 10);  
  38.         BrokenPrimeProducer producer = new BrokenPrimeProducer(queue);  
  39.         producer.start();  
  40.         try {  
  41.             while (needMorePrimes())  
  42.                 queue.take();  
  43.         } finally {  
  44.             producer.cancel();  
  45.         }  
  46.     }  
  47.   
  48.     public static boolean needMorePrimes() throws InterruptedException {  
  49.         boolean result = true;  
  50.         i--;  
  51.         if (i == 0)  
  52.             result = false;  
  53.         return result;  
  54.     }  
  55. }  
  56.     

  我们在main中通过queue.take来消费产生的素数(虽然仅仅是取出扔掉),我们只消费了1000个素数,然后尝试取消产生素数的任务,很遗憾, 取消不了,因为产生素数的线程产生素数的速度大于我们消费的速度,我们在消费1000后就停止消费了,那么任务将被queue的put方法阻塞,永远也不 会去判断cancelled状态变量,任务取消不了。正确的做法应当是使用中断(interrupt):
java 代码
 
  1. package net.rubyeye.concurrency.chapter7;  
  2.   
  3. import java.math.BigInteger;  
  4. import java.util.concurrent.BlockingQueue;  
  5. import java.util.concurrent.CountDownLatch;  
  6. import java.util.concurrent.LinkedBlockingQueue;  
  7. import java.util.concurrent.TimeUnit;  
  8.   
  9. public class PrimeProducer extends Thread {  
  10.     static int i = 1000;  
  11.   
  12.     private final BlockingQueue<biginteger> queue;  </biginteger>
  13.   
  14.     private volatile boolean cancelled = false;  
  15.   
  16.     PrimeProducer(BlockingQueue<biginteger> queue) {  </biginteger>
  17.         this.queue = queue;  
  18.     }  
  19.   
  20.     public void run() {  
  21.         BigInteger p = BigInteger.ONE;  
  22.         try {  
  23.             while (!Thread.currentThread().isInterrupted()) {  
  24.                 p = p.nextProbablePrime();  
  25.                 queue.put(p);  
  26.             }  
  27.         } catch (InterruptedException cusumed) {  
  28.         }  
  29.     }  
  30.   
  31.     public void cancel() {  
  32.         interrupt();  
  33.     }  
  34.   
  35.     public static void main(String args[]) throws InterruptedException {  
  36.         BlockingQueue<biginteger> queue = new LinkedBlockingQueue<biginteger>(  </biginteger></biginteger>
  37.                 10);  
  38.         PrimeProducer producer = new PrimeProducer(queue);  
  39.         producer.start();  
  40.         try {  
  41.             while (needMorePrimes())  
  42.                 queue.take();  
  43.         } finally {  
  44.             producer.cancel();  
  45.         }  
  46.     }  
  47.   
  48.     public static boolean needMorePrimes() throws InterruptedException {  
  49.         boolean result = true;  
  50.         i--;  
  51.         if (i == 0)  
  52.             result = false;  
  53.         return result;  
  54.     }  
  55. }  
  56.     

   在run方法中,通过Thread的isInterrupted来判断interrupt status是否已经被修改,从而正确实现了任务的取消。关于interrupt,有一点需要特别说明,调用interrupt并不意味着必然停止目标线 程的正在进行的工作,它仅仅是传递一个请求中断的信号给目标线程,目标线程会在下一个方便的时刻中断。而对于阻塞方法产生的 InterruptedException的处理,两种选择:要么重新抛出让上层代码来处理,要么在catch块中调用Thread的interrupt 来保存中断状态。除非你确定要让工作线程终止(如上所示代码),否则不要仅仅是catch而不做任务处理工作(生吞了 InterruptedException),更详细可以参考这里。如果不清楚外部线程的中断策略,生搬硬套地调用interrupt可能产生不可预料的后果,可参见书中7.1.4例子。

   另外一个取消任务的方法就是采用Future来管理任务,这是JDK5引入的,用于管理任务的生命周期,处理异常等。比如调用ExecutorService的sumit方法会返回一个Future来描述任务,而Future有一个cancel方法用于取消任务。
   那么,如果任务调用了不可中断的阻塞方法,比如Socket的read、write方法,java.nio中的同步I/O,那么该怎么处理呢?简单地,关闭它们!参考下面的例子:
java 代码
 
  1. package net.rubyeye.concurrency.chapter7;  
  2.   
  3. import java.io.IOException;  
  4. import java.io.InputStream;  
  5. import java.net.Socket;  
  6.   
  7. /** 
  8.  * 展示对于不可中断阻塞的取消任务 通过关闭socket引发异常来中断 
  9.  *  
  10.  * @author Admin 
  11.  *  
  12.  */  
  13. public abstract class ReaderThread extends Thread {  
  14.     private final Socket socket;  
  15.   
  16.     private final InputStream in;  
  17.   
  18.     public ReaderThread(Socket socket) throws IOException {  
  19.         this.socket = socket;  
  20.         this.in = socket.getInputStream();  
  21.     }  
  22.   
  23.     // 重写interrupt方法  
  24.     public void interrupt() {  
  25.         try {  
  26.             socket.close();  
  27.         } catch (IOException e) {  
  28.         } finally {  
  29.             super.interrupt();  
  30.         }  
  31.     }  
  32.   
  33.     public void run() {  
  34.         try {  
  35.             byte[] buf = new byte[1024];  
  36.             while (true) {  
  37.                 int count = in.read(buf);  
  38.                 if (count < 0)  
  39.                     break;  
  40.                 else if (count > 0)  
  41.                     processBuff(buf, count);  
  42.             }  
  43.         } catch (IOException e) {  
  44.         }  
  45.     }  
  46.   
  47.     public abstract void processBuff(byte[] buf, int count);  
  48. }  
  49.     

  Reader线程重写了interrupt方法,其中调用了socket的close方法用于中断read方法,最后,又调用了super.interrupt(),防止当调用可中断的阻塞方法时不能正常中断。
   发表时间:2007-09-04  
这个里面如果子线程里面在接收到中断后(interrupt status被改变),如果一直没有阻塞方法 此任务是不是会一直执行下去呢?
也就是说这个"下一个方便的时刻"是不是就是指阻塞方法被调用时呢?


   在run方法中,通过Thread的isInterrupted来判断interrupt status是否已经被修改,从而正确实现了任务的取消。关于interrupt,有一点需要特别说明,调用interrupt并不意味着必然停止目标线 程的正在进行的工作,它仅仅是传递一个请求中断的信号给目标线程,目标线程会在下一个方便的时刻中断。而对于阻塞方法产生的 InterruptedException的处理,两种选择:要么重新抛出让上层代码来处理,要么在catch块中调用Thread的interrupt 来保存中断状态。除非你确定要让工作线程终止(如上所示代码),否则不要仅仅是catch而不做任务处理工作(生吞了 InterruptedException),更详细可以参考这里。如果不清楚外部线程的中断策略,生搬硬套地调用interrupt可能产生不可预料的后果,可参见书中7.1.4例子。

  



0 请登录后投票
   发表时间:2007-09-05  
调用阻塞方法(join,wait,sleep),那么当收到中断信号,会抛出InterruptedException,离开阻塞状态;而如果没有调用 这些阻塞方法,中断信号将可以被忽略,当然你可以通过Thread.currentThread().isInterrupted()判断状态来进行手工 处理。
  





0 请登录后投票
   发表时间:2007-09-05  
我前两天解决过这个问题,给你写了解决办法,结果被可笑的弄成灌水帖子,这个问题我研究了3天才弄明白.你自己在我博客上找吧,有个StreamGobbler的方法,用那个就不会无限 waitFor()再次bs下javaeye.还有有问题可以去javaworld上问我现在一般都去那里了.这里已经走上歧途
0 请登录后投票
   发表时间:2008-02-03  
hellsing42 写道
我前两天解决过这个问题,给你写了解决办法,结果被可笑的弄成灌水帖子,这个问题我研究了3天才弄明白.你自己在我博客上找吧,有个StreamGobbler的方法,用那个就不会无限 waitFor()再次bs下javaeye.还有有问题可以去javaworld上问我现在一般都去那里了.这里已经走上歧途


看来大家的见解基本相同啊。


0 请登录后投票
   发表时间:2008-02-03  
文章写的很不错,线程可中断实现是非常麻烦的事情,通过volatile标志和中断都可能有问题。java concurrent practice 和 java 多线程设计模式都有讲解,其中比较赞同volatile和interrupt中断同时应用来保证cancel成功。

其他的关于有人评灌水有些时候是很讨厌的事情,真的怀疑投票得人弄清楚问题是什么没有。
0 请登录后投票
   发表时间:2008-02-17  
代码有问题,第二个类BrokenPrimeProducer的cancel方法里应该改为:
this.cancelled = true;线程BrokenPrimeProducer会去判断cancelled状态变量的
0 请登录后投票
论坛首页 Java企业应用版

跳转论坛:
Global site tag (gtag.js) - Google Analytics