java并发编程6.取消与关闭

/**
 * 素数生成器
 */
private class PrimeGenerator implements Runnable{
    private final List<BigInteger> primes = new ArrayList<BigInteger>();
    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);
    }
}

    /**
     * cancel方法由finally块调用，从而确保即使在调用sleep时被中断也能取消素数生成器的执行
     */
    List<BigInteger> secondPrimes() throws InterruptedException{
        PrimeGenerator generator = new PrimeGenerator();
        new Thread(generator).start();
        try{
            Thread.sleep(1);
        }finally{
            generator.cancel();
        }
        return generator.get();
    }

/**
 * 如果生产者的速度超过了消费者的处理速度，队列将被填满，put方法也会阻塞。
 * 当生产者在put方法中阻塞时，如果消费者希望取消生产者任务，它会调用cancel()方法来设置cancelled标志，
 * 但此时生产者却永远不能检查这个标志，因为它无法从阻塞的put方法中恢复过来（消费者已经已经停止从队列去除素数，put将一直阻塞）。
 */
private 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 e){
            
        }
    }
    
    public void cancel(){
        cancelled = true;
    }
}

void consumePrimes() throws InterruptedException{
        BlockingQueue<BigInteger> primes = new LinkedBlockingQueue<BigInteger>();
        BrokenPrimeProducer producer = new BrokenPrimeProducer(primes);
        producer.start();
        try{
            while(needMorePrimes()){
                consume(primes.take());
            }
        }finally{
            //如果阻塞，将取消失败
            producer.cancel();
        }
    }

/**
 * interrupt能中断目标线程
 * isInterrupted返回目标线程的中断状态
 * interrupted清除当前线程的中断状态，并返回它之前的值，也是清除中断状态的唯一方法。
 */
public class Thread{
    public void interrupt(){}
    public boolean isInterrupted(){}
    public static boolean interrupted(){}
    //...
}

/**
 * 使用中断而不是boolean标志来请求取消。
 * 有两个位置可以检测出中断：在阻塞的put方法中，以及在循环开始处。
 * 由于调用了阻塞的put方法，因此可以不进行显示地检测（put中会检测并响应中断，抛出InterruptedException），
 * 但执行检测会提高对中断的响应性（如果可阻塞的方法的调用频率并不高，则不足以获得足够的响应性）
 */
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 e){
            
        }
    }
    
    public void cancel(){
        interrupt();
    }
}

    /**
     * 将InterruptedException传递给调用者
     */
    BlockingQueue<Task> queue;
    //...
    public Task getNextTask() throws InterruptedException{
        return queue.take();
    }

public Task getNextTask(BlockingQueue<Taskgt> queue){
        boolean interrupted = false;
        try{
            while(true){
                try{
                    return queue.take();
                }catch(InterruptedException e){
                    interrupted = true;
                }
            }
        }finally{
            if(interrupted){
                Thread.currentThread().interrupt();
            }
        }
    }

/**
     * 在调用线程中安排一个取消任务，在运行指定的时间间隔后中断它。
     * 这解决了从任务中抛出未检查异常的问题，该异常可以做到被timeRun的调用者（也就是运行Runnable r的线程）捕获。
     * 但是由于timedRun可以从任意一个线程中调用，因此它无法知道之歌调用线程的中断策略，如果线程不响应中断，
     * 那么timedRun会在任务结束时才返回，此时可能已经超过了指定的时限。
     * 而且如果任务在超时之前完成，那么中断timedRun所在线程的取消任务将在timedRun返回到调用者之后启动。
     * 虽然不知道那时情况下将运行什么代码，但结果一定是不好的。
     */
    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();
    }

private static final ScheduledExecutorService cancelExec = Executors.newScheduledThreadPool(1);
    /**
     * 执行任务的线程拥有自己的执行策略，即使任务不响应中断，限时运行的方法仍能返回到它的调用者。
     * 在启动任务线程后，timedRun将执行一个限时的join方法（同步等待这个线程结束或者超时，）。
     * 在join返回后，它将检查任务是否有异常抛出。如果有的话，则在调用timedRun的线程中再次抛出
     * （Throwable在两个线程中共享，需声明为volatiel类型）。
     * 但由于依赖于限时join，无法知道执行控制是因为线程正常退出还是join超时而返回，join本身不会返回某个状态来表明它是否成功。
     */
    public static void timedRun(Runnable r,long timeout,TimeUnit unit) throws InterruptedException{
        class RethrowableTask implements Runnable{
            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();
    }

/**
     * 在Future.get()抛出异常时，如果知道不再需要结果，就可以调用Future。cancel来取消任务
     */
    public static void timedRun(Runnable r,long timeout,TimeUnit unit) throws InterruptedException{
        Future<?> task = taskExec.submit(r);
        try{
            
        }catch(TimeoutException e){
            //取消任务
        }catch(ExecutionException e){
            //如果任务中抛出了异常，那么重新抛出异常
            throw launderThrowable(e.getCause());
        }finally{
            //如果任务已经结束，那么执行取消操作也不会带来任何影响
            //如果任务正在运行，那么将被中断
            task.cancel(true);
        }
    }

/**
 * ReaderThread管理了一个套接字连接，采用同步方式从该套接字中读取数据，并将接收到的数据传递给processBuffer。
 * 为了结束某个用户的连接或者关闭服务器，ReaderThread改写了interrupt方法，
 * 使其既能处理标准的中断，也能关闭底层的套接字。
 */
public class ReaderThread extends Thread{
    
    private final Socket socket;
    private final InputStream in;
    
    public ReaderThread(Socket socket) throws IOException{
        this.socket = socket;
        this.in = socket.getInputStream();
    }
    
    public void interrupt(){
        try{
            socket.close();
        }catch(IOException e){
            
        }finally{
            super.interrupt();
        }
    }
    
    public void run(){
        try{
            byte[] buf = new byte[1000];
            while(true){
                int count = in.read(buf);
                if(count < 0){
                    break;
                }else if(count > 0){
                    processBuffer(buf,count);
                }
            }
        }catch(IOException e){
            
        }
    }
}

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

/**
 * CancellingExecutor扩展了ThreadPoolExecutor接口，
 * 并通过改写newTaskFor使得CancellableTask可以自定义自己的Future
 */
public class CancellingExecutor extends ThreadPoolExecutor{

    public CancellingExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    //...
    
    protected<T> RunnableFuture<T> newTaskFor(Callable<T> callable){
        if(callable instanceof CancellableTask){
            return ((CancellableTask<T>)callable).newTask();
        }else{
            return super.newTaskFor(callable);
        }
    }
}

/**
 * SocketUsingTask实现了CancellableTask，并定义了Future.cancel。
 * 如果SocketUsingTask通过其自己的Future来取消，那么底层的套接字将被关闭并且线程将被中断。
 * 因此它提高了任务对取消操作的响应性。
 */
public abstract class SocketUsingTask<T> implements CancellableTask<T>{
    
    private Socket socket;
    
    protected synchronized void setSocket(Socket s){
        socket = s;
    }
    
    public synchronized void cancel(){
        try{
            if(socket != null){
                socket.close();
            }
        }catch(IOException e){
            
        }
    }
    
    public RunnableFuture<T> newTask(){
        return new FutureTask<T>(this){
            public boolean cancel(boolean mayInterruptIfRunning){
                try{
                    SocketUsingTask.this.cancel();
                }finally{
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
}

/**
 * LogWriter给出了一个简单的日志服务示例，其中日志操作在单独的日志线程中执行。
 * 产生日志消息的线程并不会将消息直接写入输出流，
 * 而是由LogWriter通过BlockingQueue将消息提交给日志线程，并由日志线程写入。
 */
public class LogWriter {
    private final BlockingQueue<String> queue;
    private final LoggerThread logger;

    public LogWriter(PrintWriter writer){
        this.queue = new LinkedBlockingQueue<String>();
        this.logger = new LoggerThread(writer);
    }

    public void start(){
        logger.start();
    }

    public void log(String msg) throws InterruptedException{
        queue.put(msg);
    }

    private class LoggerThread extends Thread{
        private final PrintWriter writer;

        public LoggerThread(PrintWriter writer){
            this.writer = writer;
        }
        //...
        public void run(){
            try{
                while(true){
                    writer.println(queue.take());
                }

            }catch(InterruptedException e){

            }finally{    

                writer.close();

            }
        }
    }
}

/**
 * 为LogWriter提供可靠关闭操作的方法需要解决竞态条件问题，因而要使日志消息的提交操作成为原子操作。
 * 然而，并不希望在消息加入队列时去持有一个锁，因为put方法本身就可以阻塞。
 * 可以通过原子方式来检查关闭请求，并且由条件地递增一个计数器来保持提交消息的权利。
 */
public class LogWriter {
    private final BlockingQueue<String> queue;
    private final LoggerThread logger;
    private PrintWriter writer;
    private boolean isShutdown;
    private int reservations;

    public LogWriter(PrintWriter writer){
        this.queue = new LinkedBlockingQueue<String>();
        this.logger = new LoggerThread();
    }

    public void start(){
        logger.start();
    }

    public void stop(){
        synchronized(this){
            isShutdown = true;
            logger.interrupt();
        }
    }

    public void log(String msg) throws InterruptedException{
        synchronized(this){
            if(isShutdown){
                throw new IllegalStateException("...");
            }
            ++reservations;
        }
        queue.put(msg);
    }

    private class LoggerThread extends Thread{
        public void run(){
            try{
                while(true){
                    try{
                        synchronized(LogWriter.this){
                            if(isShutdown && reservations == 0){
                                break;
                            }
                        }
                        String msg = queue.take();
                        synchronized(LogWriter.this){
                            --reservations;
                        }
                        writer.println(msg);
                    }catch(InterruptedException e){

                    }
                }
            }finally{
                writer.close();
            }
        }
    }
}

/**
 * 在复杂的程序中，通常将ExecutorService封装在某个服务中，并且该服务能提供自己的生命周期方法。
 * 通过封装ExecutorService，可以将所有权链从应用程序扩展到服务以及线程，
 * 所有权链上的各个成员都将管理它所拥有的服务或线程的生命周期。
 */
public class LogService {
    private final ExecutorService exec = Executors.newSingleThreadExecutor();
    
    public void start(){
        
    }
    
    public void stop() throws InterruptedException {
        try{
            exec.shutdown();
            exec.awaitTermination(timeout, unit);
        }finally{
            writer.close();
        }
    }
    
    public void log(String msg){
        try{
            exec.execute(new WriteTask(msg));
        }catch(RejectedExecutionException e){
            
        }
    }
}

    /**
     * 使用私有Executor，并且该Executor的生命周期受限于方法调用
     * checkMail能在多台主机上并行地检查新邮件，它创建一个私有的Executor，并向每台主机提交一个任务。
     * 然后，当所有邮件检查任务都执行完成后，关闭Executor，等待结束。
     * 采用AtomicBoolean代替volatile boolean是因为内部的Runnable可以访问hasNewMail。
     */
    boolean checkMail(Set<String> hosts,long timeout,TimeUnit unit) throws InterruptedException{
        ExecutorService exec = Executors.newCachedThreadPool();
        final AtomicBoolean hasNewMail = new AtomicBoolean(false);
        try{
            for(final String host : hosts){
                exec.execute(new Runnable(){
                    public void run(){
                        if(checkMail(host)){
                            hasNewMail.set(true);
                        }
                    }
                });
            }
        }finally{
            exec.shutdown();
            exec.awaitTermination(timeout, unit);
        }
        return hasNewMail.get();
    }

/**
 * 通过封装ExecutorService并使得execute或submit记录哪些任务是在关闭后取消的。
 * TrackingExecutor可以找出哪些任务已经开始但还没正常完成。
 * 在Executor结束后，getCancelledTasks返回被取消的任务清单。
 * 如果要被记录，这些任务在返回时必须维持线程的中断状态。
 *
 */
public class TrackingExecutor extends AbstractExecutorService{
    private final ExecutorService exec;
    private final Set<Runnable> taskCancellAtShutdown = Collections.synchronizedSet(new HashSet<Runnable>());
    
    public TrackingExecutor(ExecutorService exec){
        this.exec = exec;
    }
    
    public List<Runnable> getCancelledTasks(){
        if(!exec.isTerminated()){
            throw new IllegalStateException("");
        }
        return new ArrayList<Runnable>(taskCancellAtShutdown);
    }
    
    public void execute(final Runnable runnable){
        exec.execute(new Runnable(){
            public void run(){
                try{
                    runnable.run();
                }finally{
                    /**
                     * 存在竞态条件，导致误报：一些被认为取消的任务可能实际上已经完成。
                     * 在任务执行最后一条指令以及线程池将任务记录为结束的两个时刻之间线程池可能被关闭（已经结束，但线程池未记录）。
                     * 如果任务两次执行的结果相同，例如网络爬虫，则可以忽略，否则需要考虑这个风险。
                     */
                    if(isShutdown() && Thread.currentThread().isInterrupted()){
                        taskCancellAtShutdown.add(runnable);
                    }
                }
            }
        });
    }

    //将ExecutorService的其他方法委托给exec
}

/**
 * CrawlTask提供了一个getPage方法，该方法能找出正在处理的页面。
 * 当爬虫程序关闭时，无论是还没有开始的任务，还是那些被取消的任务，都将记录它们的URL。
 * 
 * 对于ExecutorService.awaitTermination用法：
 * awaitTermination调用后会阻塞，等待所有子线程任务执行完毕并且shutdown被调用。
 * 参考文章：http://blog.csdn.net/pengbaoxing/article/details/42006679
 */
public abstract class WebCrawler {
    private volatile TrackingExecutor  exec;
    
    private final Set<URL> urlsToCrawl = new HashSet<URL>();
    
    public synchronized void start(){
        exec = new TrackingExecutor(Executors.newCachedThreadPool());
        for(URL url : urlsToCrawl){
            submitCrawlTask(url);
        }
        urlsToCrawl.clear();
    }
    
    public synchronized void stop() throws InterruptedException{
        try{
            saveUncrawled(exec.shutdownNow());
            if(exec.awaitTermination(timeout,unit)){
                saveUncrawled(exec.getCancelledTasks());
            }
        }finally{
            exec = null;
        }
    }
    
    protected abstract List<URL> processPage(URL url);
    
    private void saveUncrawled(List<Runnable> uncrawled){
        for(Runnable task : uncrawled){
            urlsToCrawl.add(((CrawlTask)task).getPage());
        }
    }
    
    private void submitCrawlTask(URL url){
        exec.execute(new CrawlTask(url));
    }
    
    private class CrawlTask implements Runnable{
        private final URL url;
        
        public CrawlTask(URL url) {
            this.url = url;
        }

        public void run(){
            for(URL link : processPage(url)){
                if(Thread.currentThread().isInterrupted()){
                    return;
                }
                submitCrawlTask(link);
            }
        }
        
        public URL getPage(){
            return url;
        }
    }
}

    public void run(){
        Throwable thrown = null;
        try{
            while(!isInterrupted()){
                runTask(getTaskFromWorkQueue());
            }
        }catch(Throwable e){
            thrown = e;
        }finally{
            threadExited(this,thrown);
        }
    }

    /**
     * UncaughtExceptionHandler接口
     */
    public interface UncaughtExceptionHandler{
        void UncaughtExcept(Thread t, Throwable e);
    }

    /**
     * 最常见的响应方式是将一个错误信息以及相应的栈追踪信息写入到应用程序日志中 
     */
    public class UEHLogger implements Thread.UncaughtExceptionHandler{
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            Logger logger = Logger.getAnonymousLogger();
            logger.log(Level.SEVERE,"Thread terminated with exception :" + t.getName(),e); 
        }
    }