处理不可中断阻塞

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

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

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

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

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

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

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

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

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

 (CancellableLoadTask.java)

/**
 * 可中断的下载任务
 * @author hongjie
 */
public class CancellableLoadTask<V> implements CancellableTask<V> {
    private InputStream input;
    private OutputStream output;
    private String filename;

    public CancellableLoadTask(String filename) {
        this.filename = filename;
    }

    /**转存文件方法*/
    public synchronized void download() throws FileNotFoundException,
            IOException {
        File file = new File(filename);
        File saveFile = null;
        String[] names = filename.split("/");
        String saveName = names[names.length - 1];
        saveFile = new File("tmp/" + saveName);
//        System.out.println(saveFile.getAbsolutePath());
        input = new FileInputStream(file);
        output = new FileOutputStream(saveFile);

        // 进行转存
        int len = 0;
        byte[] buffer = new byte[1024];
        while (-1 != (len = input.read(buffer, 0, buffer.length))) {
            output.write(buffer, 0, len);
        }

        input.close();
        output.close();
    }

    /**重写的方法，关闭阻塞的输入输出流*/
    @Override
    public void cancel() {
        try {
            if (null != input)
                input.close();
            if (null != output)
                output.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public RunnableFuture<V> newTask() {
        //重写FutureTask的cancel()方法，首先调用自定义的 cancel方法，停掉阻塞的任务
        return new FutureTask<V>(this) {
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                CancellableLoadTask.this.cancel();
                return super.cancel(mayInterruptIfRunning);
            }
        };
    }

    @Override
    public V call() throws Exception {
        download();
        return null;
    }
}

(DownloadExecutor.java)

/**
 * 下载任务的线程池
 * @author hongjie
 *
 */
public class DownloadExecutor extends ThreadPoolExecutor {
    
    public DownloadExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    /**重写newTaskFor()方法，返回自定义的Future*/
    @Override
    protected <V> RunnableFuture<V> newTaskFor(Callable<V> callable) {
        if(callable instanceof CancellableLoadTask)
            return ((CancellableLoadTask<V>)callable).newTask();
        return super.newTaskFor(callable);
    }
}

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

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

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

(StopLoadTest.java: 测试)

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

程序运行结果：
源文件：

下载超时：

目标文件：