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

//自定义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 返回自己的Callableprotected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {if (callable instanceof CancellableTask)return ((CancellableTask<T>) callable).newTask();elsereturn super.newTaskFor(callable);}
}

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

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; rethrowthrow launderThrowable(e.getCause());} finally {// Harmless if task already completedtask.cancel(true); // interrupt if running
        }}

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

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);}}});}@Testpublic void test() throws InterruptedException {ExecutorService executorService = Executors.newSingleThreadExecutor();TrackingExecutor trackingExecutor = new TrackingExecutor();trackingExecutor.execute(new Runnable() {@Overridepublic 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());}
}