续文件断续重传，多线程写入文件

蔚落 2024-03-23 12:37 69阅读 0赞

#### 文章目录 ####

*  前言
 *  一、使用线程池进行多线程操作
 *   *  创建实体类
     *  使用线程池
     *  代码解读
 *  二、使用CompletableFuture实现功能
 *   *  使用默认线程池
     *  代码解读
     *  使用自定义线程池
 *  三、上篇文章遗漏的内容
 *  四、总结

--------------------

## 前言 ##

在上篇文章中，我们使用redis来存储分片信息，做到分片上传和断续重传。我们发现，对于后端代码，大部分时间是使用在mergeChunks方法上，就是将redis存储的分片信息写入磁盘。本篇文章即探讨如何使用多线程来优化写入磁盘的速度。若没看过上篇文章的童鞋可以跳转[java实现文件分片上传并且断点续传][java]看一看。

--------------------

`以下是本篇文章正文内容，下面案例可供参考`

## 一、使用线程池进行多线程操作 ##

先放代码，再一一解释:

### 创建实体类 ###

public class MergeTask {
        
        private final int start;
        private final int end;
    
        public MergeTask(int start, int end) {
        
            this.start = start;
            this.end = end;
        }
    
        public int getStart() {
        
            return start;
        }
    
        public int getEnd() {
        
            return end;
        }
    }

### 使用线程池 ###

private static final Integer TASK_SIZE = 20; // 这里根据自己前端页面分片大小选择一个合适的数字
    
    private boolean mergeChunks(Map<String, byte[]> chunkMap, File destFile) {
        
            // 将分片按照顺序合并
            List<MergeTask> tasks = new ArrayList<>();
            for (int i = 0; i < chunkMap.size(); i += TASK_SIZE) {
        
                int end = Math.min(i + TASK_SIZE, chunkMap.size());
                tasks.add(new MergeTask(i, end));
            }
            List<Future<Boolean>> futures = new ArrayList<>();
            ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
            try (FileOutputStream outputStream = new FileOutputStream(destFile)) {
        
                for (MergeTask task : tasks) {
        
                    futures.add(executorService.submit(() -> {
        
                        for (int i = task.getStart(); i < task.getEnd(); i++) {
        
                            byte[] chunkBytes = chunkMap.get(String.valueOf(i));
                            if (chunkBytes == null) {
        
                                throw new RuntimeException("分片不存在");
                            }
                            outputStream.write(chunkBytes);
                        }
                        return true;
                    }));
                }
                for (Future<Boolean> future : futures) {
        
                    if (!future.get()) {
        
                        return false;
                    }
                }
                return true;
            } catch (Exception e) {
        
                e.printStackTrace();
                return false;
            } finally {
        
                executorService.shutdownNow();
            }
        }

### 代码解读 ###

对于上述代码的部分：

1.  MergeTask 用来记录自己负责哪些分片
2.  对于Executors.newFixedThreadPool();，使用newFixedThreadPool方法创建固定数量的线程池。原因是在磁盘上读写文件消耗的资源是很大的，因此这里不使用newCachedThreadPool方法来根据需要动态创建线程数量。而对于newSingleThreadExecutor方法，只会创建一个线程数，那跟new Thread一样，而且本来的方法就是单线程的，所以也不采用。
3.  Runtime.getRuntime().availableProcessors()是用来获取当前机器的CPU核心数的方法，根据当前环境的cpu来创建线程数。当然这里你也可以根据需求来更改数量。
4.  for (Future future : futures) 这个循环中，由于future.get()方法是阻塞的，所以会阻塞等待每个 Future 的结果，如果想非阻塞等待，可以考虑下面的方法。

## 二、使用CompletableFuture实现功能 ##

由于CompletableFuture 的 allOf() 和 join() 方法可以非阻塞地等待所有任务完成并返回结果，所以最后选择了这个方法。

### 使用默认线程池 ###

private boolean mergeChunks(Map<String, byte[]> chunkMap, File destFile) {
        
            try (FileOutputStream outputStream = new FileOutputStream(destFile)) {
        
                List<CompletableFuture<Void>> futures = new ArrayList<>();
                // 将分片按照顺序合并
                for (int i = 0; i < chunkMap.size(); i++) {
        
                    int finalI = i;
                    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        
                        byte[] chunkBytes = chunkMap.get(String.valueOf(finalI));
                        try {
        
                            outputStream.write(chunkBytes);
                        } catch (IOException e) {
        
                            e.printStackTrace();
                        }
                    });
                    futures.add(future);
                }
                // 等待所有的 CompletableFuture 完成
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
                return true;
            } catch (IOException e) {
        
                e.printStackTrace();
                return false;
            }
        }

### 代码解读 ###

对于上述代码：

1.  创建一个 CompletableFuture 列表，遍历 chunkMap，将每一个分片文件的合并操作放入一个独立的 CompletableFuture 中，并将其添加到列表中。最后调用 CompletableFuture.allOf() 方法，等待所有 CompletableFuture 执行完成。
2.  futures.toArray(new CompletableFuture\[0\])这里，在调用 CompletableFuture.allOf() 方法时，需要传入一个 CompletableFuture<?>... 类型的可变参数列表。因此使用futures.toArray将其转换为 CompletableFuture<?>\[\] 数组才能作为参数传入 allOf() 方法中。
3.  在默认情况下，CompletableFuture 会使用公共的 ForkJoinPool 线程池来执行异步任务。如果chunkMap里的内容太多，可能会创建过多的线程，在上面已经提过线程过多会导致的危害，因此我们最好是指定自定义的线程池来供CompletableFuture使用。

### 使用自定义线程池 ###

Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> 执行的内容,executor);

这样便可以使用自定义的线程池来完成任务了

--------------------

## 三、上篇文章遗漏的内容 ##

对于上篇文章，需要补充的一点是，当mergeChunks返回的结果为false时，应该删除已经写入的内容。需要添加如下代码：

if (flag == true) {
        
         connection.del(key.getBytes());
         return ResponseEntity.ok().body(resource.getURI().toString());
    } else {
        
       File file = new File(uploadPath + fileName);
       if(file.exists()){
        
           file.delete();
       }else{
        
           System.out.println("不存在");
       }
       return ResponseEntity.status(555).build(); // 告诉前端合并失败，需要进行重新点击上传。在redis的信息我们不用删除，这样即使重新上传也不会很久。
    }

## 四、总结 ##

对于多线程操作，可以直接使用线程池来进行处理，但是这样会造成循环等待的结果，因此最后使用了CompletableFuture来解决问题。

[java]: https://blog.csdn.net/a12789sd/article/details/130170748