Java 线程池处理文件：有界队列、异常与资源关闭

import java.io.BufferedReader;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class FileBatch {
    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            throw new IllegalArgumentException("Usage: java FileBatch <directory>");
        }
        Path directory = Paths.get(args[0]);
        if (!Files.isDirectory(directory)) {
            throw new IOException("Not a directory: " + directory);
        }
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                5, 5, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(64),
                new ThreadPoolExecutor.CallerRunsPolicy());
        Map<Path, Future<Long>> tasks = new LinkedHashMap<>();
        try {
            try (DirectoryStream<Path> stream =
                         Files.newDirectoryStream(directory, "*.txt")) {
                for (Path file : stream) {
                    if (Files.isRegularFile(file)) {
                        tasks.put(file, pool.submit(() -> countLines(file)));
                    }
                }
            }
            pool.shutdown();
            int failed = 0;
            for (Map.Entry<Path, Future<Long>> task : tasks.entrySet()) {
                try {
                    long lines = task.getValue().get();
                    System.out.println(task.getKey() + ": " + lines + " lines");
                } catch (ExecutionException e) {
                    failed++;
                    System.err.println("Failed: " + task.getKey());
                    e.getCause().printStackTrace(System.err);
                }
            }
            if (failed > 0) {
                throw new IllegalStateException(failed + " file(s) failed");
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
            throw e;
        } finally {
            shutdownAndAwait(pool);
        }
    }

    private static long countLines(Path file) throws IOException {
        long count = 0;
        try (BufferedReader reader =
                     Files.newBufferedReader(file, StandardCharsets.UTF_8)) {
            while (reader.readLine() != null) {
                if (Thread.currentThread().isInterrupted()) {
                    throw new IOException("Interrupted: " + file);
                }
                count++;
            }
        }
        return count;
    }

    private static void shutdownAndAwait(ThreadPoolExecutor pool) {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
                    System.err.println("Workers have not terminated");
                }
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}