public class TerminableTaskRunner {
    // 存儲要執(zhí)行的任務(wù)
    private final BlockingQueue<Runnable> tasks;
    // 線程終止標志
    private volatile boolean terminated;
    // 剩余的任務(wù)數(shù)
    private final AtomicInteger count;
    // 實際執(zhí)行任務(wù)的線程
    private volatile Thread workThread;
 
    public TerminableTaskRunner(int capacity) {
        this.tasks = new LinkedBlockingDeque<>(capacity);
        this.count = new AtomicInteger(0);
        this.workThread = new WorkThread();
        workThread.start();
    }
 
    public void submit(Runnable task) {
        this.tasks.add(task);
        this.count.incrementAndGet();
    }
 
    public void shutdown() {
        terminated = true; // 線程終止標志，由于中斷標志可能會被覆蓋，所以需要自己創(chuàng)建一個標志
        if (workThread != null)
            workThread.interrupt(); // 喚醒線程
    }
 
    private class WorkThread extends Thread {
        @Override
        public void run() {
            Runnable task;
            try {
                while (!terminated || tasks.size() >= 1) {
                    task = tasks.take();
                    try {
                        task.run(); // 可能會清空當前線程的中斷標記，如task.run()在內(nèi)部調(diào)用的阻塞方法拋出了InterruptedException
                    } catch (Throwable e) {
                        e.printStackTrace();
                    }
                    count.decrementAndGet();
                }
            } catch (InterruptedException e) {
                // 一旦調(diào)用shutdown且tasks.take()阻塞住，就拋出該異常，沒有任務(wù)要執(zhí)行，直接終止
                workThread = null;
            }
        }
    }
 
    public static void main(String[] args) {
        TerminableTaskRunner taskRunner = new TerminableTaskRunner(4);
        for (int i = 0; i < 4; i++) {
            taskRunner.submit(()->{
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    System.out.println("客戶端調(diào)用了 shutdown 方法");
                }
            });
        }
        taskRunner.shutdown();
 
    }
}