多线程中如何取消任务

@ThreadSafe
public class PrimeGenerator implements Runnable {
    private static ExecutorService exec = Executors.newCachedThreadPool();

    @GuardedBy("this") private final List<BigInteger> primes = new ArrayList<BigInteger>();
    //必须是volatile
    private volatile boolean cancelled;

    public void run() {
        BigInteger p = BigInteger.ONE;
        //检查状态，从而取消任务
        while (!cancelled) {
            p = p.nextProbablePrime();
            synchronized (this) {
                primes.add(p);
            }
        }
    }

    public void cancel() {
        cancelled = true;
    }

    public synchronized List<BigInteger> get() {
        return new ArrayList<BigInteger>(primes);
    }

    static List<BigInteger> aSecondOfPrimes() throws InterruptedException {
        PrimeGenerator generator = new PrimeGenerator();
        exec.execute(generator);
        try {
            SECONDS.sleep(1);
        } finally {
            generator.cancel();
        }
        return generator.get();
    }
}

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;
    }
}

public class PrimeProducer extends Thread {
    private final BlockingQueue<BigInteger> queue;

    PrimeProducer(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();
    }
}

这就是为什么大多数可阻塞的库方法都是抛出中断异常（InterruptedException）作为中断响应。它们永远不会在某个由自己拥有的线程中运行，因此它们为任务或库代码实现了最合理的取消策略：尽快退出执行流程，并把中断信息传递给调用者，从而使调用栈中的上层代码可以采取进一步的操作。

当检测到中断请求时，任务并不需要立刻放弃所有的操作，它可以推迟处理中断请求到某个更适合的时刻。既然要推迟处理，就需要记住中断请求，并在完成当前任务后抛出中断异常，或者表示已收到中断请求。

无论任务把中断视为取消，还是其它某个中断响应操作，都应该小心地保存执行线程的中断状态。如果除了将中断异常传递给调用者外，还需要进行其它操作，则因该在捕获中断异常之后回复中断状态。

Thread.currentThread().interrupt();

    BlockingQueue<Task> queue;
    ……
    //抛出InterruptedException，从而将中断传递给调用者
    public Task getNextTask() throws InterruptedException{
        return queue.take();
    }

public TaskgetNextTask(BlockingQueue<Task>queue){
   booleaninterrupted=false;
   try{
      while(true){
          try{
              return queue.take();
          }
          catch (InterruptedException e){
             interrupted=true;
             // 重新尝试
          }
      }
   }
   finaly{
      if (interrupted) 
          Thread.currentThread().interrupt();
   }
}

/**
 * 在外部线程中安排中断(不要这么做)
 */
public class TimedRun1 {
    private static final ScheduledExecutorService cancelExec = Executors.newScheduledThreadPool(1);

    public static void timedRun(Runnable r, long timeout, TimeUnit unit) {
        final Thread taskThread = Thread.currentThread();
        cancelExec.schedule(new Runnable() {//①安排一个取消任务，希望在运行指定的时间间隔后中断②处的任务。
            public void run() {
                taskThread.interrupt();
            }
        }, timeout, unit);
        r.run();//②运行任务
    }
}

/**
 * 在专门的线程中中断任务
 */
public class TimedRun2 {
    private static final ScheduledExecutorService cancelExec = newScheduledThreadPool(1);

    public static void timedRun(final Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
        //方法内部类
        class RethrowableTask implements Runnable {
            //由于Throwable将在两个线程之间共享，因此声明为volatile类型，从而确保安全地将其从任务线程发布到timedRun线程。
            private volatile Throwable t;

            public void run() {
                try {
                    r.run();
                } catch (Throwable t) {
                    this.t = t;
                }
            }

            void rethrow() {
                if (t != null)
                    throw launderThrowable(t);
            }
        }

        RethrowableTask task = new RethrowableTask();
        final Thread taskThread = new Thread(task);
        taskThread.start();//①启动任务
        cancelExec.schedule(new Runnable() {//②
            public void run() {
                taskThread.interrupt();
            }
        }, timeout, unit);
        taskThread.join(unit.toMillis(timeout));//③
        task.rethrow();//④join返回后，它将检查任务中是否有异常抛出，如果有，则会在调用timedRun的线程中再次抛出该异常。
    }
}

/**
 * 通过Future来取消任务
 */
public class TimedRun {
    private static final ExecutorService taskExec = Executors.newCachedThreadPool();

    public static void timedRun(Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
        Future<?> task = taskExec.submit(r);
        try {
            task.get(timeout, unit);
        } catch (TimeoutException e) {
            // 因超时而取消任务
        } catch (ExecutionException e) {
            // 任务异常，重新抛出异常信息
            throw launderThrowable(e.getCause());
        } finally {
            // 如果该任务已经完成，将没有影响
            // 如果任务正在运行，将因为中断而被取消
            task.cancel(true); // interrupt if running
        }
    }
}

/**
 * 通过改写Interrupt方法将非标准的取消操作封装在Thread中。 
 */
public class ReaderThread extends Thread {
    private static final int BUFSZ = 512;
    private final Socket socket;
    private final InputStream in;

    public ReaderThread(Socket socket) throws IOException {
        this.socket = socket;
        this.in = socket.getInputStream();
    }

    /**
     * ReaderThread改写了interrupt方法，使其既能处理标准的中断，也能关闭底层的socket
     */
    public void interrupt() {
        try {
            // 关闭socket。此时in.read会抛出异常
            socket.close();
        } catch (IOException ignored) {
        } finally {
            // 正常的中断
            super.interrupt();
        }
    }

    public void run() {
        try {
            byte[] buf = new byte[BUFSZ];
            while (true) {
                int count = in.read(buf);
                if (count < 0)
                    break;
                else if (count > 0)
                    processBuffer(buf, count);
            }
        } catch (IOException e) { 
            // 如果socket关闭，in.read方法将会抛出异常。借此机会，响应中断，线程退出
        }
    }

    public void processBuffer(byte[] buf, int count) {
    }
}

我们可以通过newTaskFor方法来进一步优化ReaderThread中封装非标准取消的技术， 这是Java 6 在ThreadPoolExecutor 中的新增功能。当把一个Callable 提交给ExecutorService 时，submit 方法会返回一个Future, 我们可以通过这个Future 来取消任务。newTaskFor是一个方法， 它将创建Future 来代表任务。newTaskFor 还能返回一个RunnableFuture 接口， 该接口扩展了Future 和Runnable (并由FutureTask 实现）。

       通过定制表示任务的Future 可以改变Future.cancel 的行为。例如， 定制的取消代码可以实现日志记录或者收集取消操作的统计信息， 以及取消一些不响应中断的操作。通过改写interrupt 方法， ReaderThread 可以取消基于套接字的线程。同样， 通过改写任务的Future.cancel 方法也可以实现类似的功能。

/**
 *  通过newTaskFor将非标准的取消操作封装在一个任务中
 */

/**
 * 扩展了Callable,并增加了一个cancel 方法和一个newTask工厂方法来构造RunnableFuture
 */
interface CancellableTask <T> extends Callable<T> {
    void cancel();

    RunnableFuture<T> newTask();
}

/**
 * 扩展了ThreadPoolExecutor，并通过改写newTaskFor使得CancellableTask可以创建自己的Future
 */
@ThreadSafe
class CancellingExecutor extends ThreadPoolExecutor {

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof CancellableTask)
            return ((CancellableTask<T>) callable).newTask();
        else
            return super.newTaskFor(callable);
    }
}

/**
 * 实现了CancellableTask,并定义了Future.cancel来关闭socket和调用super.cancel。如果SocketUsingTask通过其自己的Future来取消，那么底层的socket将被关闭时确保响应取消操作，而且还能调用可阻塞的socket I/O方法
 */
public abstract class SocketUsingTask <T> implements CancellableTask<T> {
    @GuardedBy("this") private Socket socket;

    protected synchronized void setSocket(Socket s) {
        socket = s;
    }

    public synchronized void cancel() {
        try {
            if (socket != null)
                socket.close();
        } catch (IOException ignored) {
        }
    }

    public RunnableFuture<T> newTask() {
        return new FutureTask<T>(this) {
            public boolean cancel(boolean mayInterruptIfRunning) {
                try {
                    SocketUsingTask.this.cancel();
                } finally {
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
} 

/**
 * 封装ExecutorService实现日志服务
 */
public class LogService {
    private final ExecutorService exec = Executors.newSingleThreadExecutor();
    private final PrintWriter writer;

    public LogService(PrintWriter writer) {
        this.writer = writer;
    }

    public void start(){

    }

    public void log(String msg) {
        try {
            exec.execute(new WriteTask(msg));
        } catch (RejectedExecutionException ignored) {
        }
    }

    public void stop(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            exec.shutdown();
            // 关闭服务后, 阻塞到所有任务被执行完毕或者超时发生，或当前线程被中断
            exec.awaitTermination(timeout, unit);
        } finally {
            writer.close();
        }
    }
}

/**
 *  通过“毒丸”对象来关闭服务
 */
public class IndexingService {
    private static final int CAPACITY = 1000;
    //毒丸
    private static final File POISON = new File("");
    private final IndexerThread consumer = new IndexerThread();
    private final CrawlerThread producer = new CrawlerThread();
    private final BlockingQueue<File> queue;
    //private final FileFilter fileFilter;
    private final File root;

    public IndexingService(File root) {
        this.root = root;
        this.queue = new LinkedBlockingQueue<File>(CAPACITY);
        
    }

    private boolean alreadyIndexed(File f) {
        return false;
    }

    // IndexingService的生产者线程
    class CrawlerThread extends Thread {
        public void run() {
            try {
                crawl(root);
            } catch (InterruptedException e) { /* 发生异常 */
            } finally {
                while (true) {
                    try {
                        System.out.println("放入“毒丸”");
                        queue.put(POISON);//放入毒丸
                        break;
                    } catch (InterruptedException e1) { /* 重试 */
                    }
                }
            }
        }

        private void crawl(File root) throws InterruptedException {
            File[] entries = root.listFiles();
            if (entries != null) {
                for (File entry : entries) {
                    if (entry.isDirectory())
                        crawl(entry);
                    else if (!alreadyIndexed(entry)){
                        System.out.println("放入生产者队列文件："+entry.getName()+" 来自线程："+Thread.currentThread().getName());
                        queue.put(entry);
                    }
                }
            }
        }
    }

    // IndexingService的消费者线程
    class IndexerThread extends Thread {
        public void run() {
            try {
                while (true) {
                    File file = queue.take();
                    // 遇到毒丸，终止
                    if (file == POISON){
                        System.out.println("遇到“毒丸”，终止");
                        break;
                    }   
                    else
                        indexFile(file);
                }
            } catch (InterruptedException consumed) {
            }
        }

        public void indexFile(File file) {
            System.out.println("消费者取出文件："+file.getName()+" 来自线程："+Thread.currentThread().getName());
            /* ... */
        };
    }

    public void start() {
        producer.start();
        consumer.start();
    }

    public void stop() {
        producer.interrupt();
    }

    public void awaitTermination() throws InterruptedException {
        consumer.join();
    }

}

/**
 * 将异常写入日志的UncaughtExceptionHandler
 */
public class UEHLogger implements Thread.UncaughtExceptionHandler {

    public void uncaughtException(Thread t, Throwable e) {
        Logger logger = Logger.getAnonymousLogger();
        logger.log(Level.SEVERE, "Thread terminated with exception: " + t.getName(), e);
    }

}

Thread thread=new Thread();
thread.setUncaughtExceptionHandler(new UEHLogger()); 
thread.start();