取消和关闭

public class PrimeGenerator implements Runnable {
    @GuardedBy("this")
    private final List<BigInteger> primes = new ArrayList<BigInteger>();

    //自定义的flag, 为保证线程可见性, 将其声明为volatile 
    private volatile boolean cancelled;

    @Override
    public void run() {
        BigInteger p = BigInteger.ONE;
        // 每次循环之前检查cancelled标记的值, 如果cancelled为true, 循环终止, 线程也就运行结束了 
        while (!cancelled) {
            p = p.nextProbablePrime();
            synchronized (this) {
                primes.add(p);
            }
        }
    }

    public void cancel() {
        cancelled = true;
    }

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

public static void main(String[] args) {
    PrimeGenerator generator = new PrimeGenerator();
    Thread t = new Thread(generator);
    t.start();

    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        //除非明确知道主线程应该终止，不然通常情况下应该重新抛出InterruptedException或者恢复被中断的线程
    }
    // 通过调用cancel方法, 将自定义的cancelled标记设置为true, 从而使得线程t运行终止  
    generator.cancel();
    System.out.println(generator.get().size());
}

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)
                // 当队列已满时, put方法将会阻塞. 一旦put方法阻塞
                //且没有其他线程从队列中取数据时, 阻塞将一直持续下去  
                queue.put(p = p.nextProbablePrime());
        } catch (InterruptedException consumed) { }
    }
    public void cancel() { cancelled = true; }
}

public static void main(String[] args) {  
    // 设置队列的最大容量为10  
    BlockingQueue<BigInteger> primes = new LinkedBlockingQueue<BigInteger>(10);  
    BrokenPrimeProducer producer = new BrokenPrimeProducer(primes);  
    producer.start();  

    try {  
        Thread.sleep(1000);  
    } catch (InterruptedException e) {
    }  
    producer.cancel();  
}  

public class Thread {   
    // 请求线程中断, 该方法会将线程的中断标记设置为true. 如何处理中断由目标线程决定  
    public void interrupt() { ... }   
    // 返回中断标记的值  
    public boolean isInterrupted() { ... }  
    // 静态方法用于重置当前进程的中断标记(将其设置为false), 并返回重置之前的值  
    public static boolean interrupted() { ... }   
    ...   
}  

private static final ScheduledExecutorService cancelExec = ...;
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.newSingleThreadScheduledExecutor();

public static void timedRun(final Runnable r,
                long timeout, TimeUnit unit) throws InterruptedException {

    class ReThrowableTask implements Runnable {
        //在当前线程和taskThread线程共享异常
        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运行限时时间
    // 如果超过限时，则让cancelExec线程池的线程对taskThread发起中断请求
    taskThread.join(unit.toMillis(timeout));
    //如果taskThread线程内捕获异常，重新抛出
    task.rethrow();
}

public class LogWriter {  
    private final BlockingQueue<String> queue;  
    private final LoggerThread logger;  

    public LogWriter(Writer writer) {  
        this.queue = new LinkedBlockingQueue<String>(CAPACITY);  
        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 void run() {  
            try {  
                while (true)  
                    writer.println(queue.take());  
            } catch (InterruptedException ignored) {  
            } finally {  
                writer.close();  
            }  
        }  
    }  
}  

/** 
 * 该方法用于停止LoggerThread线程 
 */  
public void shutDown() {  
    logger.interrupt();  
}  

public class LogService {
    private final BlockingQueue<String> queue;
    private final LoggerThread loggerThread;
    /**
     * 表示是否关闭Service
     */
    private boolean isShutdown;
    /**
     * 队列中待处理数据的数量
     */
    private int reservations;

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

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

    public void shutDown() {
        synchronized (this) {
            isShutdown = true;
        }
        loggerThread.interrupt();
    }

    public void log(String msg) throws InterruptedException {
        synchronized (this) {
            // service已关闭后调用log方法直接抛出异常  
            if (isShutdown) {
                throw new IllegalStateException("Service has been shut down");
            }
            ++reservations;
        }
        // BlockingQueue本身就是线程安全的, put方法的调用不在同步代码块中  
        // 我们只需要保证isShutdown和reservations是线程安全的即可  
        queue.put(msg);
    }

    private class LoggerThread extends Thread {
        private final PrintWriter writer;
        private LoggerThread(PrintWriter writer) {
            this.writer = writer;
        }

        public void run() {
            try {
                while (true) {
                    try {
                        synchronized (this) {
                            // 当service已关闭且处理完队列中的所有数据时才跳出while循环  
                            if (isShutdown && reservations == 0) {
                                break;
                            }
                        }
                        String msg = queue.take();
                        synchronized (this) {
                            --reservations;
                        }
                        writer.println(msg);
                    } catch (InterruptedException e) {
                        // 发生InterruptedException异常时不应该立刻跳出while循环  
                        // 而应该继续输出log, 直到处理完队列中的所有数据  
                    }
                }
            } finally {
                writer.close();
            }
        }
    }
}

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