通过ExecutorService提交FileInputStream

Question

在此作为指导之后，我将尝试同时向两个API发送FileInputStream，如果其中任何一个失败，我们将出错。我有两个可调用

private void submitCallablesWithExecutor(final FileInputStream content)
throws InterruptedException, ExecutionException, TimeoutException {
ExecutorService executorService = null;
try {
executorService = Executors.newFixedThreadPool(5);
//send to api1
final Future<String> futureOne = executorService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
final String returnedValue = implOne.storeContent(content)
return returnedValue;
}
});
//send to api2
final Future<String> futureTwo = executorService.submit(new Callable<String>() {
@Override
public String call() throws Exception {
final String returnedValue = implTwo.storeContent(content);
return returnedValue;
}
});
final String valueOne = futureOne.get(4, TimeUnit.SECONDS);
final String valueTwo = futureTwo.get(4, TimeUnit.SECONDS);
} finally {
executorService.shutdown();
}
}

在api2s实现中，它尝试使用以下方法计算文件大小

protected Long calculateFilesize(InputStream data) throws StoreException {
try {
return (long) data.available();
} catch (IOException e) {
LOGGER.error("Error determining filesize of InputStream");
throw new StoreException( e);
}
}

我在数据传输过程中遇到以下错误。available()

java.io.IOException:流关闭

我怀疑在某个时候流正在关闭，而此时它无法读取它。如有任何建议，我们将不胜感激。

值得注意的是，调用submitCallablesWithExecutor(content)的方法被try-with-resources包围。我想知道它是否自动关闭了Stream？

try (final FileInputStream content = new FileInputStream(tmpFile)) {
submitCallablesWithExecutor(content)
}

Answer 1

如果没有看到更多的任务代码，很难判断。这是封闭流的一种可能情况。executorService.shutdown();的javadoc说：

This method does not wait for previously submitted tasks to complete execution.

如果您的第一个任务运行时间超过4秒，则在调用futureOne.get(4, TimeUnit.SECONDS)时会得到TimeoutException，但您没有抓住它，因此您可以尝试关闭FileInputStream的资源。然而，第二个可调用程序可能仍在executor服务中运行，因此可能会得到此java.io.IOException: Stream Closed异常。

您可以使用future.get()进行修复，它会在第一个任务完成时阻塞，但听起来您不希望这样，所以最好干净地关闭您的服务线程。

尝试在shutdown()之后添加executorService.awaitTermination(long timeout, TimeUnit unit)，并捕获TimeoutException并进行适当处理，以确保第二个任务在这种超时/错误情况下干净退出，例如使用锁存器或标志来通知结束任务。