使用腾讯云 SCF 云函数压缩 COS 对象存储文件

时间  2019-11-07
标签 使用 腾讯 scf 函数 压缩 cos 对象 存储 文件 栏目 腾讯 繁體版
原文   原文链接

欢迎你们前往腾讯云技术社区，获取更多腾讯海量技术实践干货哦~node

做者：腾讯云Serverless团队git

在使用腾讯云 COS 对象存储的过程当中，咱们常常有想要把整个 Bucket 打包下载的需求，可是 COS 并无提供整个 Bucket 打包下载的能力。这时，咱们能够利用腾讯云的 SCF 无服务器云函数，完成 COS Bucket 的打包，并从新保存压缩后的文件到 COS 中，而后经过 COS 提供的文件访问连接下载文件。github

可是在使用 SCF 云函数进行 COS Bucket 打包的过程当中，偶尔会碰到这样的问题：我指望将某个 COS Bucket 内的文件所有下载下来而后打包压缩，把压缩文件再上传到 COS 中进行备份；可是在这个过程当中，COS Bucket 内的文件可能数量多体积大，而 SCF 云函数的运行环境，实际只有 512MB 的 /tmp 目录是能够读写的。这样算上下载的文件，和生成的 ZIP 包，可能仅支持必定体积的文件处理，知足不了所需。怎么办？算法

在这种状况下，可能有的同窗会想到使用内存，将内存转变为文件系统，即内存文件系统，或者直接读取文件并放置在内存中，或者在内存中生成文件。这种方法能解决一部分问题，但同时也带来了些其余问题：缓存

  1. SCF 云函数的内存配置也是有上限的，目前上限是 1.5GB。
  2. SCF 云函数的收费方式是按配置内存*运行时间。若是使用配置大内存的方法，实际是在为可能偶尔碰到的极端状况支付没必要要的费用，不符合咱们使用 SCF 云函数就是要精简费用的目的。

咱们在这里尝试了一种流式文件处理的方式，经过单个文件压缩后数据当即提交 COS 写的方法，一次处理一个文件，使得被压缩文件无需在 SCF 的缓存空间内堆积，压缩文件也无需放在缓存或内存中，而是直接写入 COS。在这里，咱们实际利用了两种特性：ZIP 文件的数据结构特性和 COS 的分片上传特性。服务器

zip 文件的数据结构

在官方文档中给出的 zip 文件格式以下：数据结构

Overall .ZIP file format:

    [local file header 1]
    [file data 1]
    [data descriptor 1]
    . 
    .
    .
    [local file header n]
    [file data n]
    [data descriptor n]
    [archive decryption header] (EFS)
    [archive extra data record] (EFS)
    [central directory]
    [zip64 end of central directory record]
    [zip64 end of central directory locator] 
    [end of central directory record]

能够看到，实际的 zip 文件格式基本是[文件头+文件数据+数据描述符]{此处可重复n次}+核心目录+目录结束标识组成的，压缩文件的文件数据和压缩数据是在文件头部，相关的目录结构，zip文件信息存储在文件尾部。这样的结构，为咱们后续 COS 分片上传写入带来了方便，能够先写入压缩数据内容，再写入最终文件信息。多线程

COS 分片上传

COS 分片上传按照以下操做便可进行：app

  1. 初始化分片上传：经过初始化动做，获取到这次上传的惟一标识ID。此ID须要保存在本地并在后续上传分片时使用。
  2. 上传分片：经过初始化时获取到的ID，配合文件分片的顺序编号，依次上传文件分片，获取到每一个分片的ETag；COS 会经过 ID 和分片顺序编号，拼接文件。
  3. 结束上传：经过初始化时获取到的ID，结合分片的顺序编号和ETag，通知 COS 分片上传已经完成，能够进行拼接。

在上传过程当中，还随时能够查询已上传分片，或结束取消分片上传。less

文件压缩处理流程设计

利用 zip 文件数据结构中文件压缩数据在前目录和额外标识在后的特性，和 COS 支持分片上传的特性，咱们能够利用流式文件处理方式来依次处理文件，而且作处处理完成一个文件压缩就上传处理后的压缩数据分片。这种处理流程能够简化为以下说明：

  1. 初始化 zip 文件数据结构，并将数据结构保存在内存中。
  2. 初始化 COS 分片上传文件，保存好分片上传 ID。
  3. 下载要放入压缩包的文件至本地，使用 zip 算法，生成压缩文件的数据内容并保存在内存中，并根据目录格式，更新zip数据格式中的目录标识。
  4. 将压缩后的文件数据使用 COS 上传分片，上传至 COS 中。
  5. 清理删除下载至本地的需压缩文件。
  6. 根据须要，重复 3~5 步骤，增长压缩包内的文件。
  7. 在压缩文件处理完成后，使用分片上传，将内存中的 zip 文件数据结构最后的目录结构部分上传至 COS。
  8. 通知 COS 结束上传，完成最终 zip 文件的自动拼接。

在这个处理流程中，一次只处理一个文件，对本地缓存和内存使用都只这一个文件的占用，相比下载所有文件再处理，大大减少了本地缓存占用和内存占用，这种状况下，使用少许缓存和内存就能够完成 COS 中大量文件的压缩打包处理。

使用SCF进行 COS 文件压缩处理实现

流式压缩文件库 archiver

咱们这里使用 node.js 开发语言来实现 COS 文件压缩处理。咱们这里使用了 cos-nodejs-sdk-v5 sdk 和 archiver 模块。其中 archiver 模块是实现zip和tar包压缩的流式处理库，可以经过 append 输入欲压缩文件，经过 stream 输出压缩后的文件流。archiver的简单用法以下：

// require modules
var fs = require('fs');
var archiver = require('archiver');

// create a file to stream archive data to.
var output = fs.createWriteStream(__dirname + '/example.zip');
var archive = archiver('zip', {
    zlib: { level: 9 } // Sets the compression level.
});

// pipe archive data to the file
archive.pipe(output);

// append a file from stream
var file1 = __dirname + '/file1.txt';
archive.append(fs.createReadStream(file1), { name: 'file1.txt' });

// append a file
archive.file('file1.txt', { name: 'file4.txt' });

// finalize the archive (ie we are done appending files but streams have to finish yet)
// 'close', 'end' or 'finish' may be fired right after calling this method so register to them beforehand
archive.finalize();

archiver 将会在每次 append 文件的时候，将文件的压缩数据输出到 pipe 指定的输出流上。所以，咱们在这里能够经过实现咱们自身的 WriteStream，获取到 archiver 的写请求，并把写入内容转移到 COS 模块的分片上传能力上。在这里，咱们实现的 WriteStream 为：

var Writable = require('stream').Writable;
var util = require('util');

module.exports = TempWriteStream;

let handlerBuffer;

function TempWriteStream(options) {
  if (!(this instanceof TempWriteStream))
    return new TempWriteStream(options);
  if (!options) options = {};
  options.objectMode = true;
  handlerBuffer = options.handlerBuffer;
  Writable.call(this, options);
}

util.inherits(TempWriteStream, Writable);

TempWriteStream.prototype._write = function write(doc, encoding, next) {
  handlerBuffer(doc);
  process.nextTick(next)
};

经过集成 nodejs 中的 Writable stream，咱们能够将写操做转移到咱们所需的 handle 上去，handle 能够对接 COS 的分片上传功能。

COS 分片上传

COS 分片上传功能的实现以下，咱们将其封装为 Upload 模块：

const cos = require('./cos')

let Duplex = require('stream').Duplex;
function bufferToStream(buffer) {
    let stream = new Duplex();
    stream.push(buffer);
    stream.push(null);
    return stream;
}

// 大于4M上传
const sliceSize = 4 * 1024 * 1024

function Upload(cosParams) {
    this.cosParams = cosParams;
    this.partNumber = 1;
    this.uploadedSize = 0;
    this.Parts = []
    this.tempSize = 0;
    this.tempBuffer = new Buffer('')
}

Upload.prototype.init = function (next) {
    const _this = this;
    cos.multipartInit(this.cosParams, function (err, data) {
        _this.UploadId = data.UploadId
        next()
    });
}
Upload.prototype.upload = function(partNumber, buffer) {
    const _this = this;
    const params = Object.assign({
        Body: bufferToStream(buffer),
        PartNumber: partNumber,
        UploadId: this.UploadId,
        ContentLength: buffer.length
    }, this.cosParams);
    cos.multipartUpload(params, function (err, data) {
        if (err) {
            console.log(err)
        } else {
            _this.afterUpload(data.ETag, buffer, partNumber)
        }
    });
}


Upload.prototype.sendData = function (buffer) {
    this.tempSize += buffer.length;
    if (this.tempSize >= sliceSize) {
        this.upload(this.partNumber, Buffer.concat([this.tempBuffer, buffer]))
        this.partNumber++;
        this.tempSize = 0;
        this.tempBuffer = new Buffer('')
    } else {
        this.tempBuffer = Buffer.concat([this.tempBuffer, buffer]);
    }
}

Upload.prototype.afterUpload = function (etag, buffer, partNumber) {
    this.uploadedSize += buffer.length
    this.Parts.push({ ETag: etag, PartNumber: partNumber })
    if (this.uploadedSize == this.total) {
        this.complete();
    }
}

Upload.prototype.complete = function () {
    this.Parts.sort((part1, part2) => {
        return part1.PartNumber - part2.PartNumber
    });
    const params = Object.assign({
        UploadId: this.UploadId,
        Parts: this.Parts,
    }, this.cosParams);
    cos.multipartComplete(params, function (err, data) {
        if (err) {
            console.log(err)
        } else {
            console.log('Success!')
        }
    });
}

Upload.prototype.sendDataFinish = function (total) {
    this.total = total;
    this.upload(this.partNumber, this.tempBuffer);
}

module.exports = Upload;

对于 COS 自己已经提供的 SDK，咱们在其基础上封装了相关查询，分片上传初始化，分片上传等功能以下：

const COS = require('cos-nodejs-sdk-v5');

const cos = new COS({
    AppId: '125xxxx227',
    SecretId: 'AKIDutrojxxxxxxx5898Lmciu',
    SecretKey: '96VJ5tnlxxxxxxxl5To6Md2',
});
const getObject = (event, callback) => {
    const Bucket = event.Bucket;
    const Key = event.Key;
    const Region = event.Region
    const params = {
        Region,
        Bucket,
        Key
    };
    cos.getObject(params, function (err, data) {
        if (err) {
            const message = `Error getting object ${Key} from bucket ${Bucket}.`;
            callback(message);
        } else {
            callback(null, data);
        }
    });
};

const multipartUpload = (config, callback) => {
    cos.multipartUpload(config, function (err, data) {
        if (err) {
            console.log(err);
        }
        callback && callback(err, data);
    });
};

const multipartInit = (config, callback) => {
    cos.multipartInit(config, function (err, data) {
        if (err) {
            console.log(err);
        }
        callback && callback(err, data);
    });
};

const multipartComplete = (config, callback) => {
    cos.multipartComplete(config, function (err, data) {
        if (err) {
            console.log(err);
        }
        callback && callback(err, data);
    });
};

const getBucket = (config, callback) => {
    cos.getBucket(config, function (err, data) {
        if (err) {
            console.log(err);
        }
        callback && callback(err, data);
    });
};


module.exports = {
    getObject,
    multipartUpload,
    multipartInit,
    multipartComplete,
    getBucket
};

在具体使用时，须要将文件中 COS 相关登陆信息的APPId，SecretId，SecretKey等替换为自身可用的真实内容。

功能入口实现函数

咱们在最终入口函数 index.js 中使用各个组件来完成最终的目录检索，文件压缩打包上传。在这里，咱们利用函数入参来肯定要访问的 bucket 名称和所属地域，指望压缩的文件夹和最终压缩后文件名。云函数入口函数仍然为 main_handler。

// require modules
const fs = require('fs');
const archiver = require('archiver');

const cos = require('./cos');

const Upload = require('./Upload')

const TempWriteStream = require('./TempWriteStream')

const start = new Date();

const getDirFileList = (region, bucket, dir, next) => {
    const cosParams = {
        Bucket: bucket,
        Region: region,
    }
    const params = Object.assign({ Prefix: dir }, cosParams);

    cos.getBucket(params, function (err, data) {
        if (err) {
            console.log(err)
        } else {
            let fileList = [];
            data.Contents.forEach(function (item) {
                if (!item.Key.endsWith('/')) {
                    fileList.push(item.Key)
                }
            });
            next && next(fileList)
        }
    })
}

const handler = (region, bucket, source, target) => {

    const cosParams = {
        Bucket: bucket,
        Region: region,
    }
    const multipartUpload = new Upload(Object.assign({ Key: target}, cosParams));

    const output = TempWriteStream({ handlerBuffer: multipartUpload.sendData.bind(multipartUpload) })

    var archive = archiver('zip', {
        zlib: { level: 9 } // Sets the compression level.
    });
    output.on('finish', function () {
        multipartUpload.sendDataFinish(archive.pointer());
    });

    output.on('error', function (error) {
        console.log(error);
    });

    archive.on('error', function (err) {
        console.log(err)
    });

    archive.pipe(output);

    multipartUpload.init(function () {
        getDirFileList(region, bucket, source, function(fileList) {
            let count = 0;
            const total = fileList.length;
            for (let fileName of fileList) {
                ((fileName) => {
                    let getParams = Object.assign({ Key: fileName }, cosParams)
                    cos.getObject(getParams, (err, data) => {
                        if (err) {
                            console.log(err)
                            return
                        }
                        var buffer = data.Body;
                        console.log("download file "+fileName);
                        archive.append(buffer, { name: fileName.split('/').pop() });
                        console.log("zip file "+fileName);
                        count++;
                        if (count == total) {
                            console.log("finish zip "+count+" files")
                            archive.finalize();
                        }
                    })
                })(fileName)
            }
        })
    })
}

exports.main_handler = (event, context, callback) => {
    var region = event["region"];
    var bucket = event["bucket"];
    var source = event["source"];
    var zipfile = event["zipfile"];
    //handler('ap-guangzhou', 'testzip', 'pic/', 'pic.zip');
    handler(region, bucket, source, zipfile)
}

测试及输出

最终咱们将如上的代码文件及相关依赖库打包为zip代码包，建立函数并上传代码包。同时咱们准备好一个 COS Bucket命名为 testzip， 在其中建立 pic 文件夹，并在文件夹中传入若干文件。经过函数页面的测试功能，咱们使用以下模版测试函数：

{
"region":"ap-guangzhou",
"bucket":"testzip",
"source":"pic/",
"zipfile":"pic.zip"
}

函数输出日志为：

...
2017-10-13T12:18:18.579Z 9643c683-b010-11e7-a4ea-5254001df6c6 download file pic/DSC_3739.JPG
2017-10-13T12:18:18.579Z 9643c683-b010-11e7-a4ea-5254001df6c6 zip file pic/DSC_3739.JPG
2017-10-13T12:18:18.689Z 9643c683-b010-11e7-a4ea-5254001df6c6 download file pic/DSC_3775.JPG
2017-10-13T12:18:18.690Z 9643c683-b010-11e7-a4ea-5254001df6c6 zip file pic/DSC_3775.JPG
2017-10-13T12:18:18.739Z 9643c683-b010-11e7-a4ea-5254001df6c6 download file pic/DSC_3813.JPG
2017-10-13T12:18:18.739Z 9643c683-b010-11e7-a4ea-5254001df6c6 finish zip 93 files
2017-10-13T12:18:56.887Z 9643c683-b010-11e7-a4ea-5254001df6c6 Success!

能够看到函数执行成功，并从 COS Bucket 根目录看到新增长的 pic.zip 文件。

项目源代码及改进方向

目前项目全部源代码已经放置在 Github 上，路径为 https://github.com/qcloud-scf/demo-scf-compress-cos。能够经过下载或 git clone 项目，获取到项目源代码，根据自身账号信息，修改 cos 文件内的账号 APPId、SecretId、SecretKey这些认证信息，而后将根目录下全部文件打包至 zip 压缩包后，经过 SCF 建立函数并经过 zip 文件上传代码来完成函数建立，根据上面所属的“测试及输出”步骤来测试函数的可用性。

函数在此提供的仍然只是个demo代码，更多的是为你们带来一种新的思路及使用腾讯云 SCF 无服务器云函数和 COS 对象存储。基于此思路，Demo自己后续还有不少能够改进的方法，或根据业务进行变化的思路：

  1. 文件的处理目前仍是下载一个处理一个，其实咱们可使用多线程和队列来加速处理过程，使用若干线程持续下载文件，使用队列对已经下载完成待处理的文件进行排队，而后使用一个压缩线程从队列中读取已下载的文件后进行压缩上传处理。这种方式能够进一步加快大量文件的处理速度，只是须要当心处理好缓存空间被使用占满后的等待和文件处理完成后的删除释放空间。
  2. 目前 Demo 从入参接受的是单个地域、Bucket、目录和输出文件，咱们彻底能够改造为从多个地域或Bucket拉取文件，也能够传递指定的文件列表而不是仅一个目录，同时函数执行触发可使用 COS 触发或 CMQ 消息队列触发，可以造成更加通用的压缩处理函数。

后续对于此 Demo 若是有更多疑问，想法，或改进需求，欢迎你们提交 git pr 或 issue。项目地址：https://github.com/qcloud-scf/demo-scf-compress-cos

相关阅读

使用腾讯云 CDN 、COS 以及万象优图实现HTTP/2样例
如何利用云对象存储 COS 免费托管静态网站
Serverless 初探

此文已由做者受权腾讯云技术社区发布，转载请注明文章出处
原文连接：https://cloud.tencent.com/community/article/810260

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程