spring - 使用 ResponseEntity 进行流式传输并确保 InputStream 关​​闭的正确方法

Question

我们的一个应用程序泄漏文件句柄，但我们尚未找到原因。

在代码中我可以看到几个与此类似的函数:

public ResponseEntity<InputStreamResource> getFoo( ... ) {
    InputStream content = getContent(...)
    InputStreamResource isr = new InputStreamResource(content);
    return ResponseEntity.status(HttpServletResponse.SC_OK).body(isr);
}

(为简洁起见，删除了 if 检查和 try/catch)

我确信此部分会导致问题，因为当我使用 JMeter 加载测试此特定代码时，我可以看到 getContent() 在此阶段失败:

is = Files.newInputStream(f.toPath());

通常我会关闭InputStream，但由于这段简短而简单的代码，我无法在return或调用body之前关闭流>.

当我运行 lsof(代码在 Linux 上运行)时，我可以看到数千个文件以读取模式打开。所以我确信这个问题是由流没有关闭引起的。

有我应该交易的最佳实践代码吗？

Answer 1

您可以尝试使用StreamingResponseBody

StreamingResponseBody

A controller method return value type for asynchronous request processing where the application can write directly to the response OutputStream without holding up the Servlet container thread.

因为您正在处理单独的线程，直接写入响应，所以您的问题要调用 close()之前return已解决。

也许您可以从以下示例开始

public ResponseEntity<StreamingResponseBody> export(...) throws FileNotFoundException {
    //...

    InputStream inputStream = new FileInputStream(new File("/path/to/example/file"));


    StreamingResponseBody responseBody = outputStream -> {

        int numberOfBytesToWrite;
        byte[] data = new byte[1024];
        while ((numberOfBytesToWrite = inputStream.read(data, 0, data.length)) != -1) {
            System.out.println("Writing some bytes..");
            outputStream.write(data, 0, numberOfBytesToWrite);
        }

        inputStream.close();
    };

    return ResponseEntity.ok()
            .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=generic_file_name.bin")
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .body(responseBody);
}

您还可以尝试使用Files (从java 7开始)

这样你就不必管理InputStream

    File file = new File("/path/to/example/file");

    StreamingResponseBody responseBody = outputStream -> {
        Files.copy(file.toPath(), outputStream);
    };

正如 @Stackee007 在评论中所述，在生产环境的重负载下，定义 @Configuration 也是一个很好的做法。类(class)TaskExecutor调整参数和管理Async流程。

@Configuration
@EnableAsync
@EnableScheduling
public class AsyncConfiguration implements AsyncConfigurer {

    private final Logger log = LoggerFactory.getLogger(AsyncConfiguration.class);

    private final TaskExecutionProperties taskExecutionProperties;

    public AsyncConfiguration(TaskExecutionProperties taskExecutionProperties) {
        this.taskExecutionProperties = taskExecutionProperties;
    }

    //  ---------------> Tune parameters here
    @Override
    @Bean(name = "taskExecutor")
    public Executor getAsyncExecutor() {
        log.debug("Creating Async Task Executor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(taskExecutionProperties.getPool().getCoreSize());
        executor.setMaxPoolSize(taskExecutionProperties.getPool().getMaxSize());
        executor.setQueueCapacity(taskExecutionProperties.getPool().getQueueCapacity());
        executor.setThreadNamePrefix(taskExecutionProperties.getThreadNamePrefix());
        return executor;
    }
    
    //  ---------------> Use this task executor also for async rest methods
    @Bean
    protected WebMvcConfigurer webMvcConfigurer() {
        return new WebMvcConfigurer() {
            @Override
            public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
                configurer.setTaskExecutor(getTaskExecutor());
            }
        };
    }

    @Bean
    protected ConcurrentTaskExecutor getTaskExecutor() {
        return new ConcurrentTaskExecutor(this.getAsyncExecutor());
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new SimpleAsyncUncaughtExceptionHandler();
    }
}

如何使用mockMvc进行测试

您可以在集成测试中简单地遵循此示例代码:

    .andExpect(request().asyncStarted())
    .andDo(MvcResult::getAsyncResult)
    .andExpect(status().isOk()).getResponse().getContentAsByteArray();

内容类型ResponseEntity<StreamingResponseBody>是 MediaType.APPLICATION_OCTET_STREAM在此示例中，您可以获取 byte[] ( .getContentAsByteArray() )，但您可以获取所有内容的 String/Json/plaintext，具体取决于您的正文响应内容类型。