分片上传、断点上传

分片上传\断点续传：

之前面试的时候被问到过，但是答的很差，今天晚上花时间了解了一下相关的实现，然后写一点总结。 具体的上传方法调用的是我们腾讯云COS桶提供的API。

方法一、初始化分片上传

在这个方法中，我们需要参数bucketName(桶名称)、key(accessKey)、storageClass（存储类型）。

1.创建初始化分块上传的对象，需要我们传递桶名称和key

2.设置存储类型，这里应该使用了数据沉降策略，这是数据存储的一种优化方式。

​ 数据沉降：将访问频次不高，但是存储在标准存储（这片内存区域中更适合访问比较频繁、性能高、成本高）的数据，转换到低频存储或者是归档存储等更冷的存储类型中，以此降低存储成本。这里是根据storageClass来判断设置的存储类型的，如果storageClass参数为空的话，就采用默认的标准存储类型；如果不为空，就根据指定的存储类型进行存储。

3.通过cosClient进行初始化分块上传任务，获取到响应对象后，获取到响应对象的uploadId即可。这个Id是比较重要的，在接下来的分块上传方法中，需要携带上uploadId，以此我们就可以知道上传的分块是属于哪一个分块任务的。服务器端在将分块进行合成的时候，也可以根据uploadId进行合成。

▼
java
复制代码
/**
     * 初始化分块上传
     *
     * @param bucketName 桶名
     * @param key        key
     * @return
     */
    private String initiateMultipartUpload(String bucketName, String key, String storageClass) {
 
        InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, key);
        // 设置存储类型：标准存储（Standard）， 低频存储存储（Standard_IA），归档存储（ARCHIVE）。默认是标准（Standard）
        if (StringUtils.isBlank(storageClass)) {
            request.setStorageClass(StorageClass.Standard);
        } else {
            if (StorageClass.valueOf(storageClass) == StorageClass.Standard) {
                request.setStorageClass(StorageClass.Standard);
            } else {
                request.setStorageClass(StorageClass.Standard_IA);
            }
        }
        String uploadId = null;
        try {
            InitiateMultipartUploadResult initResult = cosClient.initiateMultipartUpload(request);
            // 获取uploadid
            uploadId = initResult.getUploadId();
        } catch (CosServiceException e) {
            logger.error("分块上传失败:", e);
            throw new CdpBaseBusinessException(CdpFileServerError.FILE_UPLOAD_FAILED);
        }
 
        return uploadId;
    }

方法二、上传分片文件方法

这个方法中实现的逻辑就是：首先调用我们的初始化上传分块文件任务方法；然后调用splitBySize()方法，将我们要上传的文件进行一个拆分处理，返回值是一个集合，集合里面存储的是分块文件的路径名；最后通过循环遍历每个分块，对分块进行一个上传。

▼
java
复制代码
public static String uploadPart(String path, File file) {
 
        String key = path + "/" + file.getName();
 
        // 初始化分块上传的请求，调用COS InitiateMultipartUploadRequest 方法
        String uploadId = initiateMultipartUpload(key);
        Long batch = null;
        try {
            // 计算文件总大小
            long totalSize = file.length();
 
            // 设置分块大小：1M
            byte data[] = new byte[1024 * 1024];
            int batchSize = data.length;
 
            // 计算分块数
            batch = totalSize / batchSize + (totalSize % batchSize > 0 ? 1 : 0);
 
            // 文件分块
            List<String> strings = new FileUtil().splitBySize(file.getPath(), batchSize);
            Thread.sleep(1000);
 
            for (int i = 0; i < batch; i++) {
                System.out.println("共" + batch + "块，正在进行第" + (i + 1) + "块");
                // 如果是最后一个分块，需要重新计算分块大小
                long partSize = batchSize;
                if (i == batch - 1) {
                    partSize = totalSize - i * batchSize;
                }
                // 分块上传
                batchUpload(uploadId, strings.get(i), partSize, i + 1, key, false);
            }
 
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
        cosclient.shutdown();
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("uploadId", uploadId);
        jsonObject.put("key", key);
        jsonObject.put("pieceSum", batch);
        return jsonObject.toString();
    }

方法三、拆分文件处理

在这个方法中，用到了线程池并发处理，主要的业务逻辑：将真个文件按照分块大小进行拆分，然后再将每个分块中的内容写到一个随机的文件中，写入的时候只需要将任务交给线程池。

▼
java
复制代码
/**
     * 拆分文件
     *
     * @param fileName 待拆分的完整文件名
     * @param byteSize 按多少字节大小拆分
     * @return 拆分后的文件名列表
     * @throws IOException
     */
    public List<String> splitBySize(String fileName, int byteSize)
            throws IOException {
        //创建一个返回对象 里面存放的是各个分块文件的名称
        List<String> parts = new ArrayList<String>();
        File file = new File(fileName);
        //计算要分多少块
        int count = (int) Math.ceil(file.length() / (double) byteSize);
        int countLen = (count + "").length();
        //创建线程池(不推荐使用Executors工具类)
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(count,
                count * 3, 1, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(count * 2));
        //遍历，进行分块处理
        for (int i = 0; i < count; i++) {
            //创建一个分块文件名
            String partFileName = file.getName() + "."
                    + leftPad((i + 1) + "", countLen, '0') + ".part";
            //将文件对应位置的数据填充到具体文件中，这里的任务交给线程的run方法去做了
            threadPool.execute(new SplitRunnable(byteSize, i * byteSize,
                    partFileName, file));
            //然后将分块文件路径添加到集合中 用于后续xxx
            parts.add(partFileName);
        }
        return parts;
    }
 
 
 
/**
     * 分割处理Runnable
     *
     * @author yjmyzz@126.com
     */
    private class SplitRunnable implements Runnable {
        int byteSize;
        String partFileName;
        File originFile;
        int startPos;
 
        public SplitRunnable(int byteSize, int startPos, String partFileName,
                             File originFile) {
            this.startPos = startPos;
            this.byteSize = byteSize;
            this.partFileName = partFileName;
            this.originFile = originFile;
        }
 
        public void run() {
            RandomAccessFile rFile;
            OutputStream os;
            try {
                //通过一个randomAccessFile类进行操作，在这个类中我们可以移动指针然后来读写某一部分的数据，正好适合我们的分片上传任务
                rFile = new RandomAccessFile(originFile, "r");
                //字节数组 用来存储本次分块文件中的内容
                byte[] bytes = new byte[byteSize];
                // 移动指针到每“段”开头
                rFile.seek(startPos);
                //将内容读取到bytes数组中
                int s = rFile.read(bytes);
                os = new FileOutputStream(partFileName);
                //然后将文件中的内容写到我们分块文件路径对应的文件中
                os.write(bytes, 0, s);
                //我们的数据一般不会立即写到磁盘中，一般都是在缓冲区，这个方法就是用来刷新一下缓冲区，确保写入到磁盘中
                os.flush();
                os.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

方法四、分块上传

在方法三中，我们已经将一个大文件拆分好了，现在就等待上传了。上传的时候需要调用cosClient提供的文件上传方法，因此我们需要封装一个请求类UploadPartRequest，需要传入参数桶名称、key、uploadId(现在用到了)

▼
java
复制代码
/**
     * 分块上传
     	uploadId:分快上传任务的id
     	path:分块文件名称
     	partSize:分块大小
     	partNumber:分块id
     	key:accessKey
     	isLastPart:是否为最后一块
     */
    private static void batchUpload(String uploadId, String path, Long partSize,
                                    Integer partNumber, String key, Boolean isLastPart) {
        try {
            UploadPartRequest uploadPartRequest = new UploadPartRequest();
            uploadPartRequest.setBucketName(bucketName);
            uploadPartRequest.setKey(key);
            uploadPartRequest.setUploadId(uploadId);
 
            // 设置分块的数据来源输入流
            File file = new File(path);
            // 传递一个输入流即可
            uploadPartRequest.setInputStream(new FileInputStream(file));
            // 设置分块的长度
            uploadPartRequest.setPartSize(file.length()); // 设置数据长度
            uploadPartRequest.setPartNumber(partNumber);     // 假设要上传的part编号是10
            uploadPartRequest.setLastPart(isLastPart);

            //调用cosClient上传分块文件
            UploadPartResult uploadPartResult = cosclient.uploadPart(uploadPartRequest);
            PartETag partETag = uploadPartResult.getPartETag();
            //将文件进行删除
            file.delete();
            System.out.println(partETag.getPartNumber());
            System.out.println(partETag.getETag());
        } catch (CosServiceException e) {
            e.printStackTrace();
        } catch (CosClientException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

断点续传：

断点续传其实是分片上传的一个升级版吧！需要考虑的条件更多一些，在断点续传里面呢，需要判断已经上传的分块文件和未上传的分片文件，所以他们前面大致的逻辑还是相似的（那是不是就可以使用模板设计模式？）

▼
java
复制代码
/**
     * 断点续传
     */
    public static String continueUpload(File file, String uploadId, String key) {
        //初始化CosClient
        initCOSClient();
        Long batch = null;
        try (FileInputStream fileInputStream = new FileInputStream(file)) {
            // 计算文件总大小
            long totalSize = file.length();
 
            // 设置每个分块的大小：1M
            byte data[] = new byte[1024 * 1024];
            int batchSize = data.length;
 
            // 计算分块数
            batch = totalSize / batchSize + (totalSize % batchSize > 0 ? 1 : 0);
 
            //对文件进行拆分
            List<String> part = new FileUtil().splitBySize(file.getPath(), batchSize);
            Thread.sleep(1000);
 
            // 创建查询已上传分块的请求对象 通过桶名称、key、uploadId进行查询，我们上传的时候也是用了这三个参数
            ListPartsRequest listPartsRequest = new ListPartsRequest(bucketName, key, uploadId);
            //结果存储在partListing的Parts中
            PartListing partListing = cosclient.listParts(listPartsRequest);
 
            // 将已上传的分块编号放到list 中
            List<Integer> completePiece = new ArrayList<>();
            for (PartSummary partSummary : partListing.getParts()) {
                completePiece.add(partSummary.getPartNumber());
            }
            // 遍历所有分块
            for (int i = 1; i <= batch; i++) {
                // 判断当前分块是否已经上传
                if (!completePiece.contains(i)) {
                    System.out.println("共" + batch + "块，正在进行第" + i + "块");
                    long partSize = batchSize;
                    //如果是最后一个分块，则重新计算分块大小
                    if (i == batch) {
                        //计算文件还剩多少数据没有上传  partSize是肯定小于batchSize的
                        partSize = totalSize - (i - 1) * batchSize;
                        batchUpload(uploadId, part.get(i - 1), partSize, i, key, true);
                    } else {
                        batchUpload(uploadId, part.get(i - 1), partSize, i, key, false);
                    }
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
 
        cosclient.shutdown();
        //创建一个jsonObject 准备返回json字符串
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("uploadId", uploadId);
        jsonObject.put("key", key);
        //分块文件的大小
        jsonObject.put("pieceSum", batch);
        return jsonObject.toString();
    }

总结：

分片上传和断点续传其实实现的逻辑类似，只是断点续传中需要判断的逻辑更多一些，比如说哪些数据已经上传过，需要做一个标记。

分片上传： 1.首先，创建初始化分快上传的请求对象，我们需要传递桶名称和accessKey 2.然后，我们还需要设置存储类型（数据沉降），需要根据参数storageClass进行判断 3.调用cosClient的初始化分开上传任务方法，获取到响应对象中的uploadId 4.计算出文件的总大小，设置分块的大小，计算出分块的数量count。然后做拆分文件处理，for循环遍历count次，每一次循环中创建随机的文件名，并且将该分块对应的数据写到文件名中，这里使用ThreadExecutorPool自定义了线程池，然后将并发处理这么多个分块。 5.文件分块好之后，就要对分块的任务进行上传了，上传的时候仍然使用cosClient的提供的上传文件的方法，这一步比较简单，就是一个封装请求对象的过程。（桶名称、key、uploadId 三者是不可少的）

断点续传： 断点续传呢其实前四步和分片上传是一样的。 在第五步的时候，我们要做一个判断逻辑，已经上传过的文件就不需要上传了，我们需要调用cosClient先将所有已经上传的分块查询出来（桶名称、key、uploadId 三者合一），然后遍历所有的分块，如果该分块已经存在的话，就不需要再上传了，如果不存在的话，再进行上传。这里其实还涉及到了最后分块，需要重新计算它的大小，只需要用文件总大小-已经上传的大小即可。 还有一个疑问，就是如果判断某个分块是已经上传了的呢？在每次上传分块的时候，都会有一个partNumber，表示他是第几块，这个是从1开始计数的。我们只需要将返回的partNumber，和准备上传的分块的number进行一个比对即可。

额外关注的点：

1.创建上传分块任务的时候，设置了存储类型，这里用到了存储沉降，对数据存储的一个优化方式。（图库中提到过）

2.在拆分文件的时候，使用了线程池并发处理（面试题：为什么不推荐使用Executors工具类），然后还使用了RandomAccessFile类，可以对文件随机位置进行读取。

3.还有一个就是分片上传和断点续传两个大致方法都类似，可以使用一个模板设计模式进行优化。

4.分片上传和断点续传的区别：

分片上传：主要是将大文件分割成小片，然后分别上传这些分片，最后在服务器端将这些分片合并成一个完整文件的技术。可以提高效率，避免因为网络波动导致整个文件上传失败的风险。侧重于提高文件上传的效率

断点续传：在文件上传的过程中，如果遇到了上传中断（网络异常、设备异常等引起的）问题，当恢复数据传输时，能够从上次中断的位置进行继续传输，不需要对整个文件进行传输。侧重于文件传输的可靠性

5.关于断点续传的一些其他知识

一个最简单的断点续传的流程如下：

a.客户端准备开始下载一个1024K的文件，服务端发送Accept-Range：bytes来告诉客户端，我支持带Range的请求（也就是说告诉客户端，你在发送请求的时候可以携带请求头Range）

b.假如客户端在下载到512Kb的时候，网络断开了，然后过了一会网络又好了，那此时客户端就要重新发送下载的请求。此时下载的请求中会携带有请求头：Range:bytes=51200 表示通知客户端从文件的512kb开始传输文件，直到文件内容结束

c.服务端接收到断点续传的请求，从文件的512kb开始传输。并且在响应头中添加:Content-Range:bytes 512000-/1024000，Content-Length:512000 ,Content-Length：512000。

请求报文中的请求头，Range：表示要求服务器端从xx位置开始传出文件

响应报文中的请求头：Accept-Range:服务端发送的，表示支持断点续传；Content-Range：表示传输的一个范围；Content-Length：表示传输的长度为多少