Object的分块上传

除了通过putObject()方法上传文件到BOS以外,BOS还提供了另外一种上传模式:分块上传(Multipart Upload)。用户可以在如下的应用场景内(但不仅限于此),使用分块上传模式,如:

  • 需要支持断点上传。
  • 上传超过5GB大小的文件。
  • 网络条件较差,和BOS的服务器之间的连接经常断开。
  • 需要流式地上传文件。
  • 上传文件之前,无法确定上传文件的大小。

分块上传比直接上传稍微复杂一点,分块上传需要分为三个阶段:

  • 开始上传(initiateMultipartUpload)
  • 上传分块(uploadPartFromBlob)
  • 上传完成(completeMultipartUpload)

浏览器端代码示例

对文件进行分块

let PART_SIZE = 5 * 1024 * 1024; // 指定分块大小

function getTasks(file, uploadId, bucketName, key) {
    let leftSize = file.size;
    let offset = 0;
    let partNumber = 1;

    let tasks = [];

    while (leftSize > 0) {
        let partSize = Math.min(leftSize, PART_SIZE);
        tasks.push({
            file: file,
            uploadId: uploadId,
            bucketName: bucketName,
            key: key,
            partNumber: partNumber,
            partSize: partSize,
            start: offset,
            stop: offset + partSize - 1
        });

        leftSize -= partSize;
        offset += partSize;
        partNumber += 1;
    }
    return tasks;
}

处理每个分块的上传逻辑

function uploadPartFile(state, client) {
    return function(task, callback) {
        let blob = task.file.slice(task.start, task.stop + 1);
        client.uploadPartFromBlob(task.bucketName, task.key, task.uploadId, task.partNumber, task.partSize, blob)
            .then(function(res) {
                ++state.loaded;
                callback(null, res);
            })
            .catch(function(err) {
                callback(err);
            });
    };
}

初始化uploadID,开始上传分块,并完成上传

let uploadId;
client.initiateMultipartUpload(bucket, key, options)
    .then(function(response) {
        uploadId = response.body.uploadId; // 开始上传,获取服务器生成的uploadId

        let deferred = sdk.Q.defer();
        let tasks = getTasks(blob, uploadId, bucket, key);
        let state = {
            lengthComputable: true,
            loaded: 0,
            total: tasks.length
        };

        // 为了管理分块上传,使用了async(https://github.com/caolan/async)库来进行异步处理
        let THREADS = 2; // 同时上传的分块数量
        async.mapLimit(tasks, THREADS, uploadPartFile(state, client), function(err, results) {
            if (err) {
                deferred.reject(err);
            } else {
                deferred.resolve(results);
            }
        });
        return deferred.promise;
    })
    .then(function(allResponse) {
        let partList = [];
        allResponse.forEach(function(response, index) {
            // 生成分块清单
            partList.push({
                partNumber: index + 1,
                eTag: response.http_headers.etag
            });
        });
        return client.completeMultipartUpload(bucket, key, uploadId, partList); // 完成上传
    })
    .then(function (res) {
        // 上传完成
    })
    .catch(function (err) {
        // 上传失败,添加您的代码
        console.error(err);
    });

Node.js端分块上传

对文件进行分块,并初始化UploadID,上传分块

let PART_SIZE = 5 * 1024 * 1024; // 指定分块大小
let uploadId;
client.initiateMultipartUpload(bucket, key, options)
    .then(function(response) {
        uploadId = response.body.uploadId; // 开始上传,获取服务器生成的uploadId

        let deferred = sdk.Q.defer();
        let tasks = getTasks(blob, uploadId, bucket, key);
        let state = {
            lengthComputable: true,
            loaded: 0,
            total: tasks.length
        };

        // 为了管理分块上传,使用了async(https://github.com/caolan/async)库来进行异步处理
        let THREADS = 2; // 同时上传的分块数量
        async.mapLimit(tasks, THREADS, uploadPartFile(state, client), function(err, results) {
            if (err) {
                deferred.reject(err);
            } else {
                deferred.resolve(results);
            }
        });
        return deferred.promise;
    })
    .then(function(allResponse) {
        let partList = [];
        allResponse.forEach(function(response, index) {
            // 生成分块清单
            partList.push({
                partNumber: index + 1,
                eTag: response.http_headers.etag
            });
        });
        return client.completeMultipartUpload(bucket, key, uploadId, partList); // 完成上传
    })
    .then(function (res) {
        // 上传完成
    })
    .catch(function (err) {
        // 上传失败,添加您的代码
        console.error(err);
    });

function getTasks(file, uploadId, bucketName, key) {
    let leftSize = file.size;
    let offset = 0;
    let partNumber = 1;

    let tasks = [];

    while (leftSize > 0) {
        let partSize = Math.min(leftSize, PART_SIZE);
        tasks.push({
            file: file,
            uploadId: uploadId,
            bucketName: bucketName,
            key: key,
            partNumber: partNumber,
            partSize: partSize,
            start: offset,
            stop: offset + partSize - 1
        });

        leftSize -= partSize;
        offset += partSize;
        partNumber += 1;
    }
    return tasks;
}
function uploadPartFile(state, client) {
    return function(task, callback) {
        return client.uploadPartFromFile(task.bucketName, task.key, task.uploadId, task.partNumber, task.partSize, task.file , task.start)
            .then(function(res) {
                ++state.loaded;
                callback(null, res);
            })
            .catch(function(err) {
                callback(err);
            });
    };
}

取消分块上传事件

用户可以使用abortMultipartUpload方法取消分块上传。

client.abortMultipartUpload(<BucketName>, <Objectkey>, <UploadID>);

获取未完成的分块上传事件

用户可以使用listMultipartUploads方法获取Bucket内未完成的分块上传事件。

client.listMultipartUploads(<bucketName>)
    .then(function (response) {
        // 遍历所有上传事件
        for (var i = 0; i < response.body.multipartUploads.length; i++) {
            console.log(response.body.multipartUploads[i].uploadId);
        }
    });

获取所有已上传的块信息

用户可以使用listParts方法获取某个上传事件中所有已上传的块。

client.listParts(<bucketName>, <key>, <uploadId>)
    .then(function (response) {
        // 遍历所有上传事件
        for (var i = 0; i < response.body.parts.length; i++) {
            console.log(response.body.parts[i].partNumber);
        }
    });