超大文件断点续传，分片上传（java后端）

小鱼儿 2024-02-05 19:43 84阅读 0赞

1. 为什么要使用断点续传（分片上传）  
普通上传，如果文件传输过程中，网络中断或者服务器错误，这个时候后端服务器还没有接受到文件，那么此次上传就会失败， 而下次上传又会从0开始上传；而且超大文件上传，会占用很大的内存来处理文件，非常消耗资源。

2. 怎么实现断点续传：  
前端需要将文件分成多个分片文件，并记录这些分片文件的索引，依次调接口上传这些分片文件，后端拿到这些分片文件，保存并记录分片文件索引。如果传输中断， 只需要告诉前端已上传的分片索引，就可以接着上次的分片位置继续上传。等所有的分片文件上传完后，后端就可以合并文件进行自己的逻辑处理了。

这里只展示后端代码；

我们需要至少两个接口：1.分片上传的接口； 2.查询分片上传的进度接口。

分片上传接口：

>     /**
>          * 上传前方景观相关数据zip-分片上传
>          * @param filePart    单次分片的文件
>          * @param partIndex    当前分片定位
>          * @param partNum    文件分片总数
>          * @param fileName    文件名
>          * @param fileMd5    超大文件的md5值（续传时，区分文件是否更改）
>          * @return
>          */
>         @PostMapping("/partUpload")
>         public R partUpload(@RequestParam("filePart")MultipartFile filePart,
>                             @RequestParam("partIndex") Integer partIndex,
>                             @RequestParam("partNum") Integer partNum,
>                             @RequestParam("fileName") String fileName,
>                             @RequestParam("fileMd5") String fileMd5) {
>             return landscapeInfoService.partUpload(filePart, partIndex, partNum, fileName, fileMd5);
>         }

@Service
    @Slf4j
    public class LandscapeInfoServiceImpl extends ServiceImpl<LandscapeInfoMapper, LandscapeInfo> implements LandscapeInfoService {
    
        @Resource
        private RedisTemplate<String, List<Integer>> redisTemplate;
    
        @Resource
        private RedisTemplate<String, String> redisTemplateStr;
    
    
        private final String path = "export/";
        private final String TempPath = "export/temp";
    
        private final Map<String, Integer> isMergePart = new HashMap<>();
    
    
    
        @Override
        public R partUpload(MultipartFile filePart, Integer partIndex, Integer partNum, String fileName, String fileMd5) {
    
            log.info("单文件分片上传,总片数:{},分片数:{},文件名:{},大小:{},uid:{}", partNum, partIndex, fileName, filePart.getSize(), fileMd5);
            // 保存文件的 fileMd5 到redis中（每次上传分片需要校验 fileMd5 是否一致。不一致需要抛出异常返回前端，调接口清除之前的上传的文件缓存，再重新上传）
            String md5Key = fileName + "_fileMd5";
            String partKey = fileName + "_part";
            if (ObjectUtils.isNotEmpty(redisTemplateStr.opsForValue().get(md5Key))) {
                if (!fileMd5.equals(redisTemplateStr.opsForValue().get(md5Key))) {
                    return R.failed(false, "当前有未完成的上传任务，请等待当前任务上传完成或清除当前上传任务，再继续上传！");
                }
            } else {
                redisTemplateStr.opsForValue().set(md5Key, fileMd5);
            }
            // 实际情况下，这些路径都应该是服务器上面存储文件的路径
            String filePath = path;//文件存放路径
            String tempPath = TempPath;//临时文件存放路径
            File dir = new File(tempPath);
            if (!dir.getParentFile().exists()) {
                dir.getParentFile().mkdirs();
            }
            if (!dir.exists()) {
                dir.mkdirs();
            }
    
    
            //生成一个临时文件名
            String tempFileNamePath = tempPath + File.separator + fileName + "_" + partIndex + part;
            try {
                //将分片存储到临时文件夹中
                File file = new File(tempFileNamePath);
                multipartFileToFile(filePart, file);
                log.info("分片文件" + partIndex + "上传成功。");
                File tempDir = new File(tempPath);
                File[] tempFiles = tempDir.listFiles();
                // 将上传的分片索引记录下来
    
                BoundValueOperations<String, List<Integer>> bounded = redisTemplate.boundValueOps(partKey);
                List<Integer> partList = bounded.get();
                if (CollUtil.isNotEmpty(partList)) {
                    partList.add(partIndex);
                } else {
                    partList = new ArrayList<>();
                    partList.add(partIndex);
                }
                bounded.set(partList);
    
                log.info("分片文件" + partIndex + "索引记录成功！");
                one:
                if (partNum.equals(Objects.requireNonNull(tempFiles).length)) {
                    //需要校验一下,表示已有异步程序正在合并了;如果是分布式这个校验可以加入redis的分布式锁来完成
                    if (isMergePart.get(fileMd5) != null) {
                        break one;
                    }
                    // 后端全部接受到文件
                    isMergePart.put(fileMd5, tempFiles.length);
                    log.info("所有分片上传完成，预计总分片：" + partNum + "; 实际总分片:" + tempFiles.length);
                    // 合并文件
                    mergeFile(partNum, fileName, filePath, tempPath, checkId, deviceType);
                    log.info("===============================  开始处理文件");
                    // todo 开始处理合并后的文件
    
    
    
    
    
    
    
                    
                }
    
            } catch (Exception e) {
                log.error("单文件分片上传失败!", e.getMessage);
                saveRoadCheckFileRecord(checkId, deviceType, checkPlace, roadCheck, 1);
                // 删除所有临时文件
                delTemp(fileName, fileMd5, md5Key, partKey, dir);
                return R.failed(false, "单文件分片上传失败");
            }
            //通过返回成功分片值，来验证分片是否有丢失
            return R.ok(true, partIndex.toString());
    
    
        }
    
        // 删除所有临时文件和redis的记录
        private void delTemp(String fileName, String fileMd5, String md5Key, String partKey, File dir) {
    
            FileUtil.del(new File(path + fileName + zip));              // 删除合并的压缩文件
            FileUtil.del(dir);                                                   // 删除临时分片文件夹
            // 移除此次合并文件保存的key
            isMergePart.remove(fileMd5);
            // 删除对应的redis记录
            redisTemplate.delete(partKey);
            redisTemplateStr.delete(md5Key);
        }
    
    
         * 合并文件
         * @param partNum 文件总数
         * @param fileName
         * @param filePath
         * @param tempPath
         * @throws IOException
         */
        private void mergeFile(Integer partNum, String fileName, String filePath, String tempPath) throws IOException {
            FileOutputStream fileOutputStream = new FileOutputStream(filePath + fileName + zip);
            
            for (int i = 0; i < partNum; i++) {
                //读取分片数据，进行分片合并
                FileInputStream fileInputStream = new FileInputStream(tempPath + File.separator + fileName + "_" + i + ".part");
                byte[] buf = new byte[1024 * 8];//8MB
                int length;
                while ((length = fileInputStream.read(buf)) != -1) {//读取fis文件输入字节流里面的数据
                    fileOutputStream.write(buf, 0, length);//通过fos文件输出字节流写出去
                }
                fileInputStream.close();
            }
            fileOutputStream.flush();
            fileOutputStream.close();
        }
    
    }

查询分片进度的接口（接口比较简单直接给方法）：

@Override
        public R<QryPartVo> qryPart(String fileName) {
            String partKey = fileName + "_part";
            String md5Key = fileName + "_fileMd5";
            BoundValueOperations<String, List<Integer>> bounded = redisTemplate.boundValueOps(partKey);
            List<Integer> list = bounded.get();
            Map<string, Object> map = new HashMap();
            String fileMd5 = redisTemplateStr.opsForValue().get(md5Key);
            if (StrUtil.isNotEmpty(fileMd5)) {
                map.put("fileMd5", fileMd5)
            }
            if (CollUtil.isNotEmpty(list)) {
                Integer max = Collections.max(list);
                map.put("partIndex", max )
            }
            return R.ok(map);
        }

这样就可以实现分片上传了，这是比较贱的分片上传，如果涉及到集群环境的分片上传， 需要吧分片文件统一放到某个位置，比如minio。当然也可以在上传过程中加入进度条，给客户端反馈上传的进度。