多段相关接口说明(Node.js SDK)

对于较大文件上传，可以切分成段上传。用户可以在如下的应用场景内（但不仅限于此），使用分段上传的模式：

上传超过100MB大小的文件。
网络条件较差，和OBS服务端之间的链接经常断开。
上传前无法确定将要上传文件的大小。

分段上传分为如下3个步骤：

分段上传的主要目的是解决大文件上传或网络条件较差的情况。下面的代码示例展示了如何使用分段上传并发上传大文件：

// 引入obs库
// 使用npm安装
const ObsClient = require('esdk-obs-nodejs');
// 使用源码安装
// var ObsClient = require('./lib/obs');
const fs = require('fs');

// 创建ObsClient实例
var obsClient = new ObsClient({
  // 推荐通过环境变量获取AKSK，这里也可以使用其他外部引入方式传入，如果使用硬编码可能会存在泄露风险
  // 您可以登录访问管理控制台获取访问密钥AK/SK，获取方式请参见https://support.huaweicloud.com/usermanual-ca/ca_01_0003.html
  access_key_id: process.env.ACCESS_KEY_ID,
  secret_access_key: process.env.SECRET_ACCESS_KEY,
  // 【可选】如果使用临时AK/SK和SecurityToken访问OBS，同样建议您尽量避免使用硬编码，以降低信息泄露风险。您可以通过环境变量获取访问密钥AK/SK，也可以使用其他外部引入方式传入
  // security_token: process.env.SECURITY_TOKEN,
  // endpoint填写Bucket对应的Endpoint, 这里以华北-北京四为例，其他地区请按实际情况填写
  server: "https://obs.cn-north-4.myhuaweicloud.com",
});

// 指定存储桶名称
const Bucket = 'bucketname'
// 指定对象名，此处以 example/objectname 为例
const Key = 'objectname'

// 指定分段大小
const DEFAULT_PART_SIZE = 9 * 1024 * 1024;
// 指定并发数为20
const CONCURRENCY = 20

// 准备分段上传的参数
const preparePartParams = (Bucket, Key, UploadId) => (sampleFile, partSize = DEFAULT_PART_SIZE) => {
	try {
		const fileSize = fs.lstatSync(sampleFile).size;
		const partCount = fileSize % partSize === 0 ? Math.floor(fileSize / partSize) : Math.floor(fileSize / partSize) + 1;

		const uploadPartParams = [];
		// 指定并发上传
		for (let i = 0; i < partCount; i++) {
			// 分段在文件中的起始位置
			let Offset = i * partSize;
			// 分段大小
			let currPartSize = (i + 1 === partCount) ? fileSize - Offset : partSize;
			// 分段号
			let PartNumber = i + 1;
			uploadPartParams.push({
				Bucket,
				Key,
				PartNumber,
				UploadId,
				Offset,
				SourceFile: sampleFile,
				PartSize: currPartSize,
			});
		};

		return ({ uploadPartParams, fileSize });
	} catch (error) {
		console.log(error)
	};
};


/**
 * uploadSuccessSize：已经上传成功的大小
 * uploadSuccessCount：已经上传成功的段数量
 * concurrency：当前并发数
 */
let uploadSuccessSize = 0;
let uploadSuccessCount = 0;
let concurrency = 0

const parts = [];

// 上传段
const uploadPart = (uploadPartParam, otherUploadPartInfo) => {
	const partCount = otherUploadPartInfo.partCount;
	const fileSize = otherUploadPartInfo.fileSize;
	concurrency++;
	return obsClient
		.uploadPart(uploadPartParam)
		.then(result => {
			const { PartNumber, PartSize } = uploadPartParam;
			if (result.CommonMsg.Status < 300) {
				uploadSuccessCount++;
				uploadSuccessSize += PartSize;
                                // 打印上传进度
				console.log(`the current concurrent count is ${concurrency} | uploaded segment: ${uploadSuccessCount}/${partCount}. the progress is ${((uploadSuccessSize / fileSize) * 100).toFixed(2)}% | the partNumber ${PartNumber} upload successed.`);
				parts.push({ PartNumber, ETag: result.InterfaceResult.ETag });
			} else {
				console.log(result.CommonMsg.Code, parts);
			};
			concurrency--;
		}).catch(function (err) {
			console.log(err);
			throw err;
		})
};

// 分段上传
const uploadFile = (sourceFile) => {
        // 初始化分段任务
	obsClient.initiateMultipartUpload({
		Bucket,
		Key
	}).then(res => {
		const Status = res.CommonMsg.Status;
		const UploadId = res.InterfaceResult.UploadId;

		if (typeof Status === 'number' && Status > 300) {
			console.log(`initiateMultipartUpload failed! Status:${Status}`);
			return;
		};

		const partParams = preparePartParams(Bucket, Key, UploadId)(sourceFile);
		const uploadPartParams = partParams.uploadPartParams;
		const fileSize = partParams.fileSize;
		const partCount = uploadPartParams.length;
		const otherUploadPartInfo = { fileSize, partCount };

		// 调用并行上传函数
		parallelFunc(uploadPartParams, (param) => uploadPart(param, otherUploadPartInfo), CONCURRENCY)
			.then(() => {
                                // 合并段
				obsClient.completeMultipartUpload({
					Bucket,
					Key,
					UploadId,
					Parts: parts.sort((a, b) => a.PartNumber - b.PartNumber)
				}, (err, result) => {
					if (err) {
						console.log('Error-->' + err);
					} else {
						console.log('Status-->' + result.CommonMsg.Status);
					};
				});
			});

	}).catch(function (err) {
		console.log(err)
	});
};


/**
 * 实现函数并行执行
 * @param {Array} params 回调函数的参数数组
 * @param {Promise} promiseFn 回调函数
 * @param {number} limit 并行数
 */
const parallelFunc = (params, promiseFn, limit) => {
	return new Promise((resolve) => {
		let concurrency = 0;
		let finished = 0;
		const count = params.length;
		const run = (param) => {
			concurrency++;
			promiseFn(param)
				.then(() => {
					concurrency--;
					drainQueue();
					finished++;
					if (finished === count) {
						resolve();
					};
				});
		};
		const drainQueue = () => {
			while (params.length > 0 && concurrency < limit) {
				var param = params.shift();
				run(param);
			};
		};
		drainQueue();
	});

};

uploadFile('localfile');