Java并发编程实践读书笔记（4）任务取消和关闭

任务的取消

中断传递原理

Java中没有抢占式中断，就是武力让线程直接中断。

Java中的中断可以理解为就是一种简单的消息机制。某个线程可以向其他线程发送消息，告诉你“你应该中断了”。收到这条消息的线程可以根据这个消息做出反应。

意思是，不是你说让我停我就会停，我愿意停就停！

中断消息的传递其实就是通过Thread的一个布尔类型中断状态变量实现的。

发送中断请求时，线程的中断变量被设置为true：

Thread t = new Thread(new Runnable() {
    @Override
    public void run() {
       //... 
    }        
});
t.start();
//...        
t.interrupt();

任务线程中监听这个状态变量就知道是否有中断请求：

@Override
public void run() {
     while(!Thread.currentThread().isInterrupted()) {
                    
     }         
}

或：

@Override
public void run() {
     while(!Thread.interrupted()) {
                    
     }         
}

注意：阻塞方法一般都会抛出InterruptedException，这个异常一旦被try-catch到了，Thread的interrupt状态就会被自动清除（设置为false）。如果不得不try-catch（比如在Runnable的run方法中，或其他类似接口方法中），但是又不知道应该怎么处理中断异常时，一定要记得恢复中断状态：

Thread.currentThread().interrupt();//重新标记为已中断

这样做的目的，是为了其他能够真的处理这个中断的地方能收到中断请求！

反过来说就是，把InterruptedException异常try-cath到了，但是又什么都不做，就像是拦截了别人的信件，自己看完之后扔垃圾桶了。这种行为破坏了导致中断机制。

 Future的cancel

Future的cancel包含两层意思，一个是如果任务还没启动那么就别启动了，另外一个意思是启动了就中断。

        try {
            future.get(5, TimeUnit.SECONDS);
        } catch (ExecutionException | TimeoutException e) {
            e.printStackTrace();
        }finally {
            future.cancel(true);
        }

处理不可中断的任务

JDK中很多阻塞方法都会考虑到中断，对于不可中断的任务，比如Socket连接，我们能做的是改写Thread的interrupt，来手动做一些事情让任务“中断”掉。

public class ReaderThread extends Thread{
    
    private final Socket socket;
    private final InputStream in;

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


    @Override
    public void interrupt() {
        try {
            //关闭socket，与后面的try-catch配合使用
            this.socket.close();
        }catch(Exception e) {
        }finally{
            super.interrupt();
        }
    }


    @Override
    public void run() {
        byte[] buf = new byte[1024];
        try {
            while(true) {
                int count = in.read(buf);//依赖socket.close异常跳出while
                if(count<0) {
                    break;
                }else if(count>0) {
                    processBuffer(buf,count);
                }
            }
        }catch(Exception e) {
            
        }
    }


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

由此应该知道，在考虑中断的功能时，可以依赖Java的中断消息机制，但更多的是要根据实际的业务需求来合理利用中断机制。

线程服务的取消

 如果我们对外提供了线程服务，那么就应该提供生命周期的方法，比如shutdown、cancel等。

基于前面提到的知识，可以用中断机制来关闭任务线程。但更重要的是线程的关闭不能影响业务。

比如WebLog应用，WebLog使用了一个阻塞队列来缓存WebLog日志，这样对个线程就能同时打日志了，另外有日志处理线程专门负责从队列中取日志来进行记录。这样一个简单的应用在取消的时候就不能盲目的“kill”掉，需要考虑：

1，关闭后已经在阻塞队列中的日志还是应该继续处理完；

2，不能继续接受新的日志存入到队列中了

/**
 * @author huqiao
 */
public class LogService {
    
    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    private final PrintWriter writer;
    private boolean isStoped = false;
    private int logCount = 0;
    
    public LogService() {
        queue = new ArrayBlockingQueue<>(10);
        loggerThread = new LoggerThread();
        writer = new PrintWriter(System.out);
    }
    public void start() {
        loggerThread.start();
    }
    
    public void stop() {
        isStoped = true;
        loggerThread.interrupt();//实现queue.take()抛异常的效果,促使任务线程赶紧去查看“isStoped”状态
    }
    public void log(String log) throws InterruptedException {
        synchronized (this) {
            if(isStoped) {
                //拒绝新日志写入
                throw new IllegalStateException(" logger is stoped");
            }
            logCount++;
        }
        queue.put(log);
    }
    
    private class LoggerThread extends Thread{

        @Override
        public void run() {
            try {
                while(true) {
                    try {
                        String log = null;
                        synchronized (LogService.this) {
                            if(isStoped && logCount == 0) {//logCount确保在stop之后还有机会处理接下来的日志
                                break;
                            }
                        }
                        log = queue.take();
                        synchronized (LogService.this) {
                            logCount--;
                        }
                        writer.write(log);
                    } catch (InterruptedException e) {
                        //忽略异常（try-catch到之后，中断状态已经被标记为false了，再take时就不会出错了），因为已经自己实现了中断机制
                    }
                }
            }finally {
                writer.close();
            }
        }
    }

}

ExecutorService的关闭

shutdown：仅仅表示ExecutorService不再接受新的提交请求，它不会给子任务发中断请求；

shutdownNow：会尝试给任务线程发送中断请求，但如果子线程没有实现中断请求，那发了也没用，子线程照样一直跑；

无论是shutdown还是shutdownNow，如果线程任务没有实现中断，那就根本不会停止。showdownNow会多返回一个List，里面存着还没执行的任务。所以在实现ExecutorService的关闭时，一定要有下面任务线程的配合。

public class ExecutorShutdownTest {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        for(int i= 0;i<1;i++) {
            executorService.execute(new Task(i+""));
        }
        
        List<Runnable> taskList = executorService.shutdownNow();
        boolean successToTerminated = executorService.awaitTermination(10, TimeUnit.SECONDS);//这里等待10秒，能正常等待到任务线程退出
        System.out.println("successToTerminated=" + successToTerminated);
        for(Runnable runnable : taskList) {
            Task task = (Task)runnable;
            System.out.println(task.getName());
        }
    }
    
    private static class Task implements Runnable{

        private String name;
        public String getName() {
            return name;
        }
        public Task(String name) {
            this.name = name;
        }
        @Override
        public void run() {
            int i = 0;
            while(true) {
                try {
                    Thread.sleep(1000);
                    System.out.println("task " + name + " is running...");
                    if(++i>3) {//3秒之后正常退出
                        break;
                    }
                } catch (InterruptedException e) {
                    //e.printStackTrace();
                }
            }
        }

    }
    
}

毒丸对象

书中介绍的这个毒丸对象是一种关闭子线程的巧妙办法。还是以前面的日志服务为例，客户代码想要子任务代码不要在继续跑了，那么可以给子任务扔一个“特殊的消息”。然后子任务会识别这个特殊的消息，让自己退出运行。

处理非正常的线程终止

线程服务中的多个线程有可能异常终止，我们应该要能即使的知道这样的事情。

在创建ExecutorService的时候可以指定线程的创建工程（ThreadFactory）：

public class UnCatchExceptionHandlerTest {
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(10,new MyThreadFactory());
        service.execute(new Runnable() {
            @Override
            public void run() {
                int i = 0;
                System.out.println(100 / i);
            }
        });
    }

}

在这个线程工程中，可以为thread设置UncatchExceptionHandler：

public class MyThreadFactory implements ThreadFactory{

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setUncaughtExceptionHandler(new MyUncatchExceptionHandler());
        return t;
    }

}

public class MyUncatchExceptionHandler implements UncaughtExceptionHandler{

    @Override
    public void uncaughtException(Thread t, Throwable e) {
        System.out.println(t.getName() + " occur exception:" + e.getMessage());
    }

}

注意，这个方法只对execute提交的任务有效，用submit提交的任务，其异常还是会被Future.get()封装在ExecutionException中重新抛出。