java流式处置惩罚zip+多线程

概述

流式处置惩罚一个zip，zip里有多个json文件。
流式处置惩罚可以制止解压一个大的zip。再加上多线程，处置惩罚的效率杠杠的。
代码

1. package 多线程.demo05多jsonCountDownLatch;
3. import com.fasterxml.jackson.databind.ObjectMapper;
4. import lombok.SneakyThrows;
5. import lombok.extern.slf4j.Slf4j;
6. import org.springframework.util.StopWatch;
8. import java.io.ByteArrayOutputStream;
9. import java.io.FileNotFoundException;
10. import java.io.IOException;
11. import java.nio.file.Files;
12. import java.nio.file.Path;
13. import java.nio.file.Paths;
14. import java.util.ArrayList;
15. import java.util.List;
16. import java.util.concurrent.ExecutorService;
17. import java.util.concurrent.Executors;
18. import java.util.concurrent.Future;
19. import java.util.concurrent.TimeUnit;
20. import java.util.zip.ZipEntry;
21. import java.util.zip.ZipInputStream;
23. @Slf4j
24. public class ZipProcessor {
26.     private Path path;
28.     private static int numThreads = Runtime.getRuntime().availableProcessors();
29.     private static ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
30.     private static ObjectMapper objectMapper = new ObjectMapper();
32.     @SneakyThrows
33.     public ZipProcessor(String filePath){
34.         path = Paths.get(filePath);
35.         if (!Files.exists(path)) {
36.             throw new FileNotFoundException("The specified ZIP file does not exist: " + filePath);
37.         }
38.     }
40.     public void streamProcess(){
41.         StopWatch stopWatch = new StopWatch();
42.         stopWatch.start();
43.         // 使用 try-with-resources 保证资源关闭
44.         try (ZipInputStream zis = new ZipInputStream(Files.newInputStream(path))) {
45.             ZipEntry entry;
46.             while ((entry = zis.getNextEntry()) != null) {
47.                 // 将当前条目的数据读取到字节数组中
48.                 byte[] byteArray = getByteArray(zis);
49.                 process(byteArray, entry);
50.                 // 关闭当前条目的输入流
51.                 zis.closeEntry();
52.             }
53.             stopWatch.stop();
54.             log.info("zip处理耗时:{}秒", stopWatch.getTotalTimeSeconds());
55.         } catch (IOException e) {
56.             stopWatch.stop();
57.             log.error("zip处理异常，耗时:{}秒", stopWatch.getTotalTimeSeconds(), e);
58.         }
60.     }
62.     public void streamParallelProcess() {
63.         StopWatch stopWatch = new StopWatch();
64.         stopWatch.start();
65.         try (ZipInputStream zis = new ZipInputStream(Files.newInputStream(path))) {
66.             ZipEntry entry;
67.             List<Future<?>> futures = new ArrayList<>();
68.             while ((entry = zis.getNextEntry()) != null) {
69.                 // 为了lambda表达式捕获局部变量
70.                 final ZipEntry currentEntry = entry;
71.                 // 将当前条目的数据读取到字节数组中
72.                 byte[] byteArray = getByteArray(zis);
73.                 Future<?> future = executorService.submit(() -> process(byteArray, currentEntry));
74.                 futures.add(future);
75.             }
77.             // 等待所有任务完成
78.             for (Future<?> future : futures) {
79.                 try {
80.                     future.get();
81.                 } catch (Exception e) {
82.                     log.error("任务执行失败", e);
83.                 }
84.             }
85.         } catch (IOException e) {
86.             log.error("读取ZIP文件异常", e);
87.         } finally {
88.             executorService.shutdown();
89.             try {
90.                 if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
91.                     executorService.shutdownNow();
92.                 }
93.             } catch (InterruptedException ex) {
94.                 executorService.shutdownNow();
95.             }
96.             stopWatch.stop();
97.             log.info("zip处理耗时:{}秒", stopWatch.getTotalTimeSeconds());
98.         }
99.     }
101.     private void process(byte[] entryData, ZipEntry entry) {
102.         try {
103.             // 在这里处理每个条目的数据
104.             ObjectMapper objectMapper = new ObjectMapper();
105.             OriginalObject originalObject = objectMapper.readValue(entryData, OriginalObject.class);
106.             log.info("完成处理:{}，sourceFileId:{}", entry.getName(), originalObject.getSourceFileId());
107.         } catch (IOException e) {
108.             log.error("处理条目 {} 异常", entry.getName(), e);
109.         }
110.     }
112.     private byte[] getByteArray(ZipInputStream zis) throws IOException {
113.         ByteArrayOutputStream baos = new ByteArrayOutputStream();
114.         byte[] buffer = new byte[1024];
115.         int length;
116.         while ((length = zis.read(buffer)) > 0) {
117.             baos.write(buffer, 0, length);
118.         }
119.         return baos.toByteArray();
120.     }
122. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。