如何通过多线程与异步优化大Zip文件读取及byte[]生成

问题场景 #

接收最大600~700MB的Multipart格式.zip文件，仅需从中读取4-5个目标JSON文件并存储为String。已为方法添加@Async("taskExecutor")注解（已配置Executor Bean），但当前仍为单线程执行，希望让Zip文件读取、最终byte[]生成两个流程并发执行，缩短整体响应时间。

原代码如下：

@Async("taskExecutor")
public CompletableFuture<ResponseEntity<?>> methodname(MultipartFile file){

    ZipEntry entry = null;
    try(ZipInputStream zipFileStream = new ZipInputStream(file.getInputStream())){
        while((entry = zipFileStream.getNextEntry()) != null){
            String entryName = entry.getName();
            
            if(entryName.contains("<file1name>")){
            BufferedReader br = new BufferedReader(new InputStreamReader(zipFileStream));
            String value1 = br.lines().collect(Collectors.joining("\n"));
            zipFileStream.closeEntry();
            }
            
            if(entryName.contains("<file2name>")){
            BufferedReader br = new BufferedReader(new InputStreamReader(zipFileStream));
            String value2 = br.lines().collect(Collectors.joining("\n"));
            zipFileStream.closeEntry();
            }
            
            if(entryName.contains("<file3name>")){
            BufferedReader br = new BufferedReader(new InputStreamReader(zipFileStream));
            String value3 = br.lines().collect(Collectors.joining("\n"));
            zipFileStream.closeEntry();
            }
        }
    }
    
    //String value1 & String value2 merged based on some condition to finally prepare String value1.
    //some logic to prepare a file
    
    if(fileExists){
        //create byte[] and Httpheaders with content disposition and mediatype and send CompletableFuture<ResponseEntity<?>>
    }
}

解决方案 #

核心思路 #

ZipInputStream是顺序流，无法并行读取不同Entry，因此将整个Zip读取流程封装为独立异步任务，同时让不依赖读取结果的前置准备工作并行执行，待两者完成后再合并结果生成最终byte[]，最大化利用线程资源。

修改后代码 #

1. 抽取异步Zip读取方法

@Async("taskExecutor")
public CompletableFuture<Map<String, String>> extractTargetJsonFiles(MultipartFile file) {
    Map<String, String> jsonContents = new HashMap<>(3);
    try (ZipInputStream zipFileStream = new ZipInputStream(file.getInputStream())) {
        ZipEntry entry;
        while ((entry = zipFileStream.getNextEntry()) != null) {
            String entryName = entry.getName();
            if (entryName.contains("<file1name>")) {
                String content = new BufferedReader(new InputStreamReader(zipFileStream))
                        .lines().collect(Collectors.joining("\n"));
                jsonContents.put("file1", content);
            } else if (entryName.contains("<file2name>")) {
                String content = new BufferedReader(new InputStreamReader(zipFileStream))
                        .lines().collect(Collectors.joining("\n"));
                jsonContents.put("file2", content);
            } else if (entryName.contains("<file3name>")) {
                String content = new BufferedReader(new InputStreamReader(zipFileStream))
                        .lines().collect(Collectors.joining("\n"));
                jsonContents.put("file3", content);
            }
            // 跳过不需要的Entry，减少IO消耗
            zipFileStream.closeEntry();
        }
    } catch (IOException e) {
        return CompletableFuture.failedFuture(e);
    }
    return CompletableFuture.completedFuture(jsonContents);
}

2. 改造原方法实现并发流程

@Async("taskExecutor")
public CompletableFuture<ResponseEntity<?>> methodname(MultipartFile file) {
    // 启动异步读取Zip文件的任务
    CompletableFuture<Map<String, String>> jsonReadFuture = extractTargetJsonFiles(file);

    // 并行执行不依赖JSON内容的前置准备工作（如检查文件、初始化Headers模板）
    CompletableFuture<Void> prePrepareFuture = CompletableFuture.runAsync(() -> {
        // 示例：检查目标文件是否存在、初始化基础HttpHeaders等
        // checkTargetFileExists();
        // initBaseHeaders();
    }, taskExecutor);

    // 等待两个任务完成后，处理结果并生成响应
    return CompletableFuture.allOf(jsonReadFuture, prePrepareFuture)
            .thenApplyAsync(v -> {
                try {
                    Map<String, String> jsonContents = jsonReadFuture.join();
                    String value1 = jsonContents.get("file1");
                    String value2 = jsonContents.get("file2");
                    String value3 = jsonContents.get("file3");

                    // 执行value1与value2的合并逻辑
                    // String mergedValue = mergeValues(value1, value2);

                    // 生成最终byte[]和响应头
                    HttpHeaders headers = new HttpHeaders();
                    headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
                    headers.setContentDisposition(ContentDisposition.attachment()
                            .filename("result.zip")
                            .build());

                    // 替换为实际生成byte[]的逻辑
                    byte[] resultBytes = "processed content".getBytes(StandardCharsets.UTF_8);

                    return new ResponseEntity<>(resultBytes, headers, HttpStatus.OK);
                } catch (Exception e) {
                    return new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR);
                }
            }, taskExecutor);
}

关键说明 #

• 异步任务拆分：将Zip读取与前置准备工作拆分为两个独立异步任务，通过CompletableFuture.allOf等待两者完成，避免串行等待
• 资源优化：跳过不需要的ZipEntry，减少不必要的IO操作和内存占用
• 异常处理：异步任务中捕获IO异常并返回failedFuture，避免未处理异常导致线程池故障
• 线程池复用：所有异步任务均使用配置的taskExecutor线程池，保证资源可控

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