java并发编程（1）并发程序的取消于关闭

一、任务的取消于关闭

1、中断Thread

　　1.每个线程都有一个boolean类型的中断状态。true则是中断状态中

　　　　interrupt：发出中断请求；isInterrupt：返回中断状态；interrupted：清除中断状态

　　2.JVM中的阻塞方法会检查线程中断状态，其响应方法为：清除中断状态，抛出InterruptedException异常，表示阻塞操作被中断结束；但JVM不保证阻塞方法何时检测到线程的中断状态

　　3.中断的理解：不会真正的中断一个正在运行的线程，而只是发出请求，具体的中断由任务自己处理

　　通过中断来取消线程通常是最好的方法

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) {
            /* Allow thread to exit */
            //如果捕获到中断异常，则由线程自己退出
        }
    }
    public void cancel() {
        interrupt();
    }
}

2、不可中断的阻塞的中断

　　如：Socket I/O操作，即使设置了中断请求，也不会中断，但是close 套接字，会使其抛出异常，达到中断效果；因此我们要重写中断方法

//自定义callable实现类
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 ignored) {
        }
    }
    //新建实例的方法
    public RunnableFuture<T> newTask() {
        return new FutureTask<T>(this) {
            public boolean cancel(boolean mayInterruptIfRunning) {
                try {
                    SocketUsingTask.this.cancel();
                } finally {
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
}

//自定义callable接口
interface CancellableTask <T> extends Callable<T> {
    void cancel();
    RunnableFuture<T> newTask();
}
//自定义 执行池
class CancellingExecutor extends ThreadPoolExecutor {
    ......
    //通过改写newTaskFor 返回自己的Callable
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof CancellableTask)
            return ((CancellableTask<T>) callable).newTask();
        else
            return super.newTaskFor(callable);
    }
}

3、通过自定义取消计时任务

private static final ScheduledExecutorService cancelExec = newScheduledThreadPool(1);
    /**
     *
     * @param r 任务
     * @param timeout 超时时间
     * @param unit TimeUnit
     * @throws InterruptedException
     */
    public static void timedRun(final Runnable r,long timeout, TimeUnit unit) throws InterruptedException {
        class RethrowableTask implements Runnable {
            //通过一个volatile变量，来存储线程是否异常
            private volatile Throwable t;
            public void run() {
                try {
                    r.run();
                } catch (Throwable t) {
                    this.t = t;
                }
            }
            private void rethrow() {
                if (t != null)
                    throw launderThrowable(t);
            }
        }
        RethrowableTask task = new RethrowableTask();
        final Thread taskThread = new Thread(task);
        taskThread.start();
        //延时timeout个unit单位后 执行线程中断
        cancelExec.schedule(() -> taskThread.interrupt(), timeout, unit);
        //无论如何都等待；如果线程不响应中断，那么通过join等待任务线程timeout时间后 不再等待，回到调用者线程
        taskThread.join(unit.toMillis(timeout));
        //如果 任务线程中有异常，则抛出
        task.rethrow();
    }

注意：依赖于join，任务超时join退出和任务正常join推出无法进行判断

4、通过Futrue来实现取消计时任务

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 {
            //通过Futrue.get(超时时间)，捕获相应的异常来处理计时运行和取消任务
            task.get(timeout, unit);
        } catch (TimeoutException e) {
            // task will be cancelled below
        } catch (ExecutionException e) {
            // exception thrown in task; rethrow
            throw launderThrowable(e.getCause());
        } finally {
            // Harmless if task already completed
            task.cancel(true); // interrupt if running
        }
    }

二、停止基于线程的服务

　　1.通常，服务不能直接中断，造成服务数据丢失

　　2.线程池服务也不能直接中断

1、日志服务

标准的生产者，消费者模式

public class LogService {
    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    private final PrintWriter writer;
    private boolean isShutdown;
    private int reservations;

    public LogService(Writer writer) {
        this.queue = new LinkedBlockingQueue<String>();
        this.loggerThread = new LoggerThread();
        this.writer = new PrintWriter(writer);
    }

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

    public void stop() {
        synchronized (this) {
            isShutdown = true;
        }
        loggerThread.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 (LogService.this) {
                            if (isShutdown && reservations == 0) {
                                break;
                            }
                        }
                        String msg = queue.take();
                        synchronized (LogService.this) {
                            --reservations;
                        }
                        writer.println(msg);
                    } catch (InterruptedException e) { /* retry */
                        //捕获了中断请求，但为了将剩余日志输出，不做处理，直到计数器 == 0时，关闭
                    }
                }
            } finally {
                writer.close();
            }
        }
    }
}

2、ExecutorService中断

　　shutDown和shutDownNow

　　通常，将ExecetorService封装；如LogService，使其具有自己的生命周期方法

　　shutDownNow的局限性：不知道当前池中的线程状态，返回未开始的任务，但不能返回已开始未结束的任务

public class TrackingExecutor extends AbstractExecutorService {
    private final ExecutorService exec;
    private final Set<Runnable> tasksCancelledAtShutdown =
            Collections.synchronizedSet(new HashSet<Runnable>());

    public TrackingExecutor() {
        exec = Executors.newSingleThreadExecutor();
    }

    /*public TrackingExecutor(ExecutorService exec) {
        this.exec = exec;
    }*/

    public void shutdown() {
        exec.shutdown();
    }

    public List<Runnable> shutdownNow() {
        return exec.shutdownNow();
    }

    public boolean isShutdown() {
        return exec.isShutdown();
    }

    public boolean isTerminated() {
        return exec.isTerminated();
    }

    public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
        return exec.awaitTermination(timeout, unit);
    }

    public List<Runnable> getCancelledTasks() {
        if (!exec.isTerminated())
            throw new IllegalStateException(/*...*/);
        return new ArrayList<Runnable>(tasksCancelledAtShutdown);
    }

    public void execute(final Runnable runnable) {
        exec.execute(new Runnable() {
            public void run() {
                try {
                    runnable.run();
                } finally {
                    if (isShutdown()
                            && Thread.currentThread().isInterrupted())
                        tasksCancelledAtShutdown.add(runnable);
                }
            }
        });
    }

    @Test
    public void test() throws InterruptedException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        TrackingExecutor trackingExecutor = new TrackingExecutor();
        trackingExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                    System.err.println("123123");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt(); //设置状态 或继续抛，在execute中处理
                    e.printStackTrace();
                } finally {

                }
            }
        });
        List<Runnable> runnables = trackingExecutor.shutdownNow();
        trackingExecutor.awaitTermination(10,TimeUnit.SECONDS);
        List<Runnable> cancelledTasks = trackingExecutor.getCancelledTasks();
        System.err.println(cancelledTasks.size());
    }
}

三、处理非正常线程终止

1.未捕获的Exception导致的线程终止

　　1.手动处理未捕获的异常

　　2.通过Thread的API UncaughExceptionHandler，能检测出某个线程又遇见未捕获而导致异常终止

　　　　注意：默认是将异常的的堆栈信息输出到控制台;自定义的Handler：implements Thread.UncaughExceptionHandler覆写方法

　　　　可以为每个线程设置，也可以设置一个全局的ThreadGroup

　　　　Thread.setUncaughtExceptionHandler/Thread.setDefaultUncaughtExceptionHandler