处理不可中断阻塞

在java库中，许多可阻塞的方法都是通过提前返回或者抛出InterruptedException来响应中断请求的，从而使开发人员更容易构建出能响应取消请求的任务。然而并非所有的可阻塞方法或者阻塞机制都能响应中断；如果一个线程由于执行同步的Socket I/O或者等待获得内置锁而阻塞，那么中断请求只能设置线程的中断状态，除此之外没有其他任何作用。

以下是不可中断阻塞的情况：

1. java.io包中的同步Socket I/O
2. java.io包中的同步I/O
3. Selector的异步I/O
4. 获取某个锁

下面是停止不可中断阻塞的方法，这篇文章很好地解释了我上一篇文章(使用Future停止超时任务)中提到的问题。

--------------------------------------------------------------------------------------------------------------------

关键步骤就是重写原来中断线程或者取消任务的方法，在方法里面加入自己的取消操作，比如关闭数据流，关闭套接字等，然后再调用父类的中断方法，这样就可以既关闭了阻塞的任务，又中断了线程。

--------------------------------------------------------------------------------------------------------------------

 下面那是具体的代码：
（CancellableTask.java）

 1 /**
 2  * 处理不可中断阻塞任务的接口
 3  * @author hongjie
 4  */
 5 public interface CancellableTask<V> extends Callable<V> {
 6     /**取消不可中断的阻塞任务*/
 7     public void cancel();
 8     /**创建自定义的FutureTask*/
 9     public RunnableFuture<V> newTask();
10 }

 (CancellableLoadTask.java)

 1 /**
 2  * 可中断的下载任务
 3  * @author hongjie
 4  */
 5 public class CancellableLoadTask<V> implements CancellableTask<V> {
 6     private InputStream input;
 7     private OutputStream output;
 8     private String filename;
 9 
10     public CancellableLoadTask(String filename) {
11         this.filename = filename;
12     }
13 
14     /**转存文件方法*/
15     public synchronized void download() throws FileNotFoundException,
16             IOException {
17         File file = new File(filename);
18         File saveFile = null;
19         String[] names = filename.split("/");
20         String saveName = names[names.length - 1];
21         saveFile = new File("tmp/" + saveName);
22 //        System.out.println(saveFile.getAbsolutePath());
23         input = new FileInputStream(file);
24         output = new FileOutputStream(saveFile);
25 
26         // 进行转存
27         int len = 0;
28         byte[] buffer = new byte[1024];
29         while (-1 != (len = input.read(buffer, 0, buffer.length))) {
30             output.write(buffer, 0, len);
31         }
32 
33         input.close();
34         output.close();
35     }
36 
37     /**重写的方法，关闭阻塞的输入输出流*/
38     @Override
39     public void cancel() {
40         try {
41             if (null != input)
42                 input.close();
43             if (null != output)
44                 output.close();
45         } catch (IOException e) {
46             e.printStackTrace();
47         }
48     }
49 
50     @Override
51     public RunnableFuture<V> newTask() {
52         //重写FutureTask的cancel()方法，首先调用自定义的 cancel方法，停掉阻塞的任务
53         return new FutureTask<V>(this) {
54             @Override
55             public boolean cancel(boolean mayInterruptIfRunning) {
56                 CancellableLoadTask.this.cancel();
57                 return super.cancel(mayInterruptIfRunning);
58             }
59         };
60     }
61 
62     @Override
63     public V call() throws Exception {
64         download();
65         return null;
66     }
67 }

(DownloadExecutor.java)

 1 /**
 2  * 下载任务的线程池
 3  * @author hongjie
 4  *
 5  */
 6 public class DownloadExecutor extends ThreadPoolExecutor {
 7     
 8     public DownloadExecutor(int corePoolSize, int maximumPoolSize,
 9             long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
10         super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
11     }
12 
13     /**重写newTaskFor()方法，返回自定义的Future*/
14     @Override
15     protected <V> RunnableFuture<V> newTaskFor(Callable<V> callable) {
16         if(callable instanceof CancellableLoadTask)
17             return ((CancellableLoadTask<V>)callable).newTask();
18         return super.newTaskFor(callable);
19     }
20 }

DownloadExecutor继承了ThreadPoolExecutor方法，并重写了newTaskFor()方法，因为当调用submit()方法的时候会先调用newTaskFor()方法，然后再执行execute();下面是父类的submit()方法

 1     /**
 2      * @throws RejectedExecutionException {@inheritDoc}
 3      * @throws NullPointerException       {@inheritDoc}
 4      */
 5     public <T> Future<T> submit(Callable<T> task) {
 6         if (task == null) throw new NullPointerException();
 7         RunnableFuture<T> ftask = newTaskFor(task);
 8         execute(ftask);
 9         return ftask;
10     }

因此当调用submit方法的时候会先调用我们重写的newTaskFor()方法，看到DownloadExecutor.java中的16~18行，返回的是CancellableLoadTask.java中自定义的FutureTask,因此会调用我们重写的cancel方法，先停掉阻塞的下载任务，然后中断线程。

(StopLoadTest.java: 测试)

 1 public class StopLoadTest {
 2     public static void main(String[] args) {
 3         CancellableLoadTask<?> loadTask = new CancellableLoadTask("G:/Games/5211install.exe");
 4         //设定线程池中的容量为1，活动时间为2秒
 5         DownloadExecutor executor = new DownloadExecutor(1, 1, 2, TimeUnit.SECONDS, new ArrayBlockingQueue(10));
 6         Future<?> future = executor.submit(loadTask);
 7         try {
 8             //设置下载超时为200毫秒
 9             future.get(200, TimeUnit.MILLISECONDS);
10         } catch (InterruptedException e) {
11             System.out.println("下载任务已经取消");
12         } catch (ExecutionException e) {
13             System.out.println("下载中发生错误，请重新下载");
14         } catch (TimeoutException e) {
15             System.out.println("下载超时，请更换下载点");
16         }
17         finally{
18             future.cancel(true);
19         }
20     }
21 }

 

程序运行结果：
源文件：

下载超时：

目标文件：