使用ExecutorService提交FileInputStream时出现Stream Closed错误如何解决

问题根因 #

1. FileInputStream 是单向、不可复用、非线程安全的流对象，你将同一个流实例传给两个并发执行的Callable任务，会出现两类问题：
   • 只要任意一个任务的storeContent实现执行完成后主动关闭了流（大部分HTTP客户端上传流时默认会主动关闭输入流），第二个任务操作流时就会直接抛出Stream Closed异常
   • 就算两个实现都不关闭流，第一个任务读完流内容后指针会走到EOF，第二个任务调用available()也只能拿到0，既读不到内容也拿不到正确的文件大小
2. 你当前的外层try-with-resources不会导致提前关流：因为submitCallablesWithExecutor中会阻塞等待两个future.get()都执行完成才返回，try-with-resources会等方法返回后才关闭流，所以不是这个机制导致的问题
3. 额外的错误用法：available()返回的是当前无需阻塞即可读取的字节数，本身就不应该用来计算文件总大小，就算流没有关闭，只要流已经被读取过一部分，返回值就会小于文件实际大小。

排查步骤 #

• 先排查两个storeContent的实现逻辑，确认是否存在执行完成后调用InputStream.close()的逻辑，90%以上的同类问题都是这个原因导致
• 可以在两个Callable的call()方法入口、以及calculateFilesize方法入口加日志，打印流的状态，确认流是在哪个执行节点被关闭的

解决方案 #

根据文件大小可以选两种方案：

方案1：小文件场景（文件体积可全部放入内存） #

提前将文件内容读取为字节数组，两个任务各自用ByteArrayInputStream包装数组操作，完全规避流共享的问题：

// 外层逻辑修改
try (final FileInputStream content = new FileInputStream(tmpFile)) {
    // Java 9+ 可直接用readAllBytes，低版本可以用commons-io等工具类读取
    byte[] fileBytes = content.readAllBytes();
    submitCallablesWithExecutor(fileBytes);
}

submitCallablesWithExecutor调整为接收字节数组，两个Callable内部各自生成流对象：

private void submitCallablesWithExecutor(final byte[] fileContent)
    throws InterruptedException, ExecutionException, TimeoutException {
    ExecutorService executorService = null;
    try {
        executorService = Executors.newFixedThreadPool(5);
        // 调用api1
        final Future<String> futureOne = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return implOne.storeContent(new ByteArrayInputStream(fileContent));
            }
        });
        // 调用api2
        final Future<String> futureTwo = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                return implTwo.storeContent(new ByteArrayInputStream(fileContent));
            }
        });
        futureOne.get(4, TimeUnit.SECONDS);
        futureTwo.get(4, TimeUnit.SECONDS);
    } finally {
        executorService.shutdown();
    }
}

同时建议修改calculateFilesize逻辑，直接取字节数组长度即可，不需要操作流。

方案2：大文件场景（避免占太多内存） #

不传流实例，直接把文件对象传给方法，两个Callable各自打开独立的FileInputStream操作，互不干扰：

// 外层逻辑修改
submitCallablesWithExecutor(tmpFile); // 直接传文件对象，不用提前开流

方法调整：

private void submitCallablesWithExecutor(final File tmpFile)
    throws InterruptedException, ExecutionException, TimeoutException {
    ExecutorService executorService = null;
    Future<String> futureOne = null;
    Future<String> futureTwo = null;
    try {
        executorService = Executors.newFixedThreadPool(5);
        futureOne = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                // 每个任务自己管理流生命周期
                try (FileInputStream content = new FileInputStream(tmpFile)) {
                    return implOne.storeContent(content);
                }
            }
        });
        futureTwo = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                try (FileInputStream content = new FileInputStream(tmpFile)) {
                    return implTwo.storeContent(content);
                }
            }
        });
        futureOne.get(4, TimeUnit.SECONDS);
        futureTwo.get(4, TimeUnit.SECONDS);
    } catch (Exception e) {
        // 任意任务失败，主动取消另一个还在执行的任务，避免资源浪费
        if (futureOne != null) futureOne.cancel(true);
        if (futureTwo != null) futureTwo.cancel(true);
        throw e;
    } finally {
        if (executorService != null) executorService.shutdown();
    }
}

calculateFilesize可以直接通过tmpFile.length()取文件大小，不需要读取流计算，准确率更高性能也更好。

内容的提问来源于stack exchange，提问作者FullStack