通州青岛网站建设,wordpress主题滑动,wordpress如何修改布局,正版厦门网站设计公司分片#xff08;500MB#xff09;进度效果展示
效果展示#xff0c;一个分片是 500MB 的 分片#xff08;10MB#xff09;进度效果展示 大文件分片上传效果展示 前端
思路 前端的思路#xff1a;将大文件切分成多个小文件#xff0c;然后并发给后端。 页面构建
先在页…分片500MB进度效果展示
效果展示一个分片是 500MB 的 分片10MB进度效果展示 大文件分片上传效果展示 前端
思路 前端的思路将大文件切分成多个小文件然后并发给后端。 页面构建
先在页面上写几个组件用来获取文件。
bodyinput typefile idfile /button iduploadButton点击上传/button
/body功能函数生成切片
切分文件的核心函数是 slice没错就是这么的神奇啊
我们把切好的 chunk 放到数组里等待下一步的包装处理
/*** 默认切片大小 10 MB*/
const SIZE 10 * 1024 * 1024;/*** 功能生成切片*/
function handleCreateChunk(file, size SIZE) {const fileChunkList [];progressData [];let cur 0;while (cur file.size) {fileChunkList.push({file: file.slice(cur, cur size),});cur size;}return fileChunkList;
}功能函数请求逻辑
在这里简单封装一下 XMLHttpRequest
/*** 功能封装请求* param {*} param0* returns*/
function request({ url, method post, data, header {}, requestList }) {return new Promise((resolve, reject) {let xhr new XMLHttpRequest();xhr.open(method, url);Object.keys(header).forEach((item) {xhr.setRequestHeader(item, header[item]);});xhr.onloadend function (e) {resolve({data: e.target.response,});};xhr.send(data);});
}功能函数上传切片
/*** 功能 上传切片* 包装好 FormData 之后通过 Promise.all() 并发所有切片*/
async function uploadChunks(hanldleData, fileName) {const requestList hanldleData.map(({ chunk, hash }) {const formData new FormData();formData.append(chunk, chunk);formData.append(hash, hash);formData.append(filename, fileName);return formData;}).map((formData) {request({// url: http://localhost:3001/upload,url: upload,data: formData,});});await Promise.all(requestList);
}/*** 功能触发上传
*/
document.getElementById(uploadButton).onclick async function () {// 切片const file document.getElementById(file).files[0];console.log(file);const fileName file.name;const fileChunkList handleCreateChunk(file);// 包装const hanldleData fileChunkList.map(({ file }, index) {return {chunk: file,hash: ${fileName}_${index},};});await uploadChunks(hanldleData, fileName);
};可以在请求中看到有很多个请求并发的上传 优化进度条的生成
自己简单撸了几个 cube 进度条
style#uploadCube {margin-top: 10px;/* width: 520px; */overflow: hidden;}.cube {width: 50px;height: 50px;background-color: #fff;float: left;border: 1px solid #000;.progress {height: 100%;line-height: 50px;text-align: center;}.uploading {background-color: #409eff;}.success {background-color: #51f400;}.error {background-color: #ff9090;}}
/style
bodyinput typefile idfile /button iduploadButton点击上传/buttondiv iduploadCube/div
/body/*** 功能生成页面进度的 HTML*/
function handleUpdateHTML(progressData) {let uploadCube document.querySelector(#uploadCube);let html ;progressData.forEach((item) {const { presentage } item;let className ;if (presentage 100) {className progress uploading;} else if (presentage 100) {className progress success;}html div classcubediv class${className} stylewidth: ${presentage}%${presentage}%/div/div;});uploadCube.innerHTML html;
}/*** 功能处理每个 chunk 的 xhr.upload.onprogress拿到各个 chunk 的上传进度* - 1. 同时通过 handleUpdateHTML 更新进度页面* - 2. progressData 用来记录各个 chunk 的进度*/
let progressData [];
function handleCreateOnProgress(data) {return (e) {data.presentage ((e.loaded / e.total) * 100).toFixed(2);console.log(JSON.stringify(progressData));handleUpdateHTML(progressData);};
}后端 Koa 后端的思路是 把 Node 暂存的 chunk 文件转移到我想处理的地方也可以直接处理看你的创建写入流把各个 chunk 合并前端会给你每个 chunk 的大小还有 hash 值来定位每个 chunk 的位置 获取 chunk 切片文件
先把上传的接口写好
const Koa require(koa);
const Views require(koa-views);
const Router require(koa-router);
const Static require(koa-static);
const { koaBody } require(koa-body);
const fs require(fs);
const fse require(fs-extra);const app new Koa();
const router new Router();
app.use(Views(__dirname));
app.use(Static(__dirname));
app.use(koaBody({multipart: true,formidable: {maxFields: 1000 * 1024 * 1024,},})
);router.get(/, async (ctx) {await ctx.render(index.html);
});/*** 功能上传接口* - 从 ctx.request.body 中获取 hash 以及 filename* - 从 ctx.request.files 中拿到分片数据* - 然后再把 node 帮我们临时存放的 chunk 文件的 filepath 拿到之后移动到我们想要存放的路径下* - filepath 和 hash 是一一对应的关系*/
router.post(/upload, async (ctx) {const { hash, filename } ctx.request.body;const { filepath } ctx.request.files?.chunk;const chunkPath ${__dirname}/chunkPath/${filename};if (!fse.existsSync(chunkPath)) {await fse.mkdirs(chunkPath);}await fse.move(filepath, ${chunkPath}/${hash});ctx.body {code: 1,};
});app.use(router.routes());
app.listen(3000, () {console.log(server start: http://localhost:3000);
});写完这些就可以拿到 chunk
合并接口
先写一个接口用来拿到 hash、文件名
/*** 功能: merge 接口* - hasMergeChunk 变量是上面用来记录的* - mergePath 定义一下合并后的文件的路径*/
router.post(/merge, async (ctx) {// console.log(ctx.request.body);const { fileName, size } ctx.request.body;hasMergeChunk {};const mergePath ${__dirname}/merge/${fileName};if (!fse.existsSync(${__dirname}/merge)) {fse.mkdirSync(${__dirname}/merge);}await mergeChunk(mergePath, fileName, size);ctx.body {data: 成功,};
});合并分片的功能函数
然后开始合并
/*** 功能合并 Chunk* - 1. chunkDir: 是 chunks 文件们所在的文件夹的路径* - 2. chunkPaths: 是个 Array数组中包含所有的 chunk 的 path* - 3. 因为 每个 chunk 的 path 命名是通过 hash 组成的所以我们先排序一下* - 算是为 createWriteStream 中的 start 做准备* - 4. 为每个 chunk 的 path 创建写入流写到 mergePath 这个路径下。因为已经* - 排序了所以 start 就是每个文件的 index * eachChunkSize* param {*} mergePath* param {*} name* param {*} eachChunkSize*/
async function mergeChunk(mergePath, name, eachChunkSize) {const chunkDir ${__dirname}/chunkPath/${name};const chunkPaths await fse.readdir(chunkDir);chunkPaths.sort((a, b) a.split(_)[1] - b.split(_)[1]);await Promise.all(chunkPaths.map((chunk, index) {const eachChunkPath ${chunkDir}/${chunk};const writeStream fse.createWriteStream(mergePath, {start: index * eachChunkSize,});return pipeStream(eachChunkPath, writeStream);}));console.log(合并完成);fse.rmdirSync(chunkDir);console.log(删除 ${chunkDir} 文件夹);
}接着就是写入流
/*** 功能创建 pipe 写文件流* - 1. [首先了解一下什么是输入输出流](https://www.jmjc.tech/less/111)* - 2. hasMergeChunk 变量用于记录一下那些已经合并完成了也可以写成数组都行。* - 3. 可以检测输出流的 end 事件表示我这个 chunk 已经流完了然后写一下善后逻辑。* param {*} path* param {*} writeStream* returns*/
let hasMergeChunk {};
function pipeStream(path, writeStream) {return new Promise((resolve) {const readStream fse.createReadStream(path); // 输出流readStream.pipe(writeStream); // 输出通过管道流向输入readStream.on(end, () {hasMergeChunk[path] finish;fse.unlinkSync(path); // 删除此文件resolve();console.log(合并 No.${path.split(_)[1]}, 已经合并${Object.keys(hasMergeChunk).length});});});
}至此一个基本的逻辑上传就做好了 后端 Node 原生
想了想还是有必要用原生写一下 复习一下。
基础搭建简单的服务
先写一个基本的服务框架
const http require(http);
const server http.createServer();
server.on(request, async (req, res) {res.setHeader(Access-Control-Allow-Origin, *);res.setHeader(Access-Control-Allow-Headers, *);if (req.method OPTION) {res.status 200;res.end();return;}res.end(hello node);
});
const POST 3000;
server.listen(POST, null, null, () {console.log(server start: http://localhost:${POST});
});基础资源返回
添加页面的返回以及资源的返回
const http require(http);
const server http.createServer();
const url require(url);
const fs require(fs);
const path require(path);
const MIME require(./mime.json);
server.on(request, async (req, res) {res.setHeader(Access-Control-Allow-Origin, *);res.setHeader(Access-Control-Allow-Headers, *);res.setHeader(content-type, text/html;charsetutf-8);if (req.method OPTION) {res.status 200;res.end();return;}let { pathname } url.parse(req.url); // 解析一下 url因为 req.url 可能会带一些参数// console.log(req.url:, req.url);// console.log(url.parse(req.url):, url.parse(req.url));console.log(进入${pathname});switch (pathname) {case /:case /index: {let rs fs.createReadStream(./index.html);rs.pipe(res);break;}case /favicon.ico: {res.end(我没有哦);break;}default: {let ext path.extname(pathname);res.setHeader(Content-Type, MIME[ext]); // 通过请求的资源后缀名来返回对应的 Content-type 的类型let rs fs.createReadStream(.${pathname});rs.pipe(res);}}
});
const POST 3000;
server.listen(POST, null, null, () {console.log(server start: http://localhost:${POST});
});MIME.json 在这个文章的最底下 或者可以自己找个更全的。
功能上传接口
接着写上传的接口这里参考大圣老师的代码写一个类来收集方法。
类定义如下新建一个文件 controller.js
/*** module.exports 方法用于在服务器端导出模块并以 CommonJS 格式提供。* - 参考https://www.delftstack.com/zh/howto/node.js/create-and-export-classes/*/
const multiparty require(multiparty);
const fse require(fs-extra);
class Controller {constructor(dirPath) {this.chunkPath dirPath;}/*** multiparty 使用方法https://www.npmjs.com/package/multiparty* - chunkFileDirPath 为关于文件 chunks 的文件夹路径每个大文件根据文件名生成相关的文件夹* - 注意回调函数里的 this* param {*} url* param {*} path*/async handleUpload(req, res) {const _this this;const form new multiparty.Form();form.parse(req, async function (err, fields, files) {if (err) {console.log(err);return;}const [chunk] files.chunk;const [hash] fields.hash;const [filename] fields.filename;const chunkFileDirPath ${_this.chunkPath}/${filename};if (!fse.existsSync(chunkFileDirPath)) {await fse.mkdirs(chunkFileDirPath);}await fse.move(chunk?.path, ${chunkFileDirPath}/${hash});res.end(收到文件 chunks);});}
}
module.exports Controller;
然后在主服务里引入这个类再上传接口这里调用一下类方法。
// ...
const Controller require(./controller);
const UPLOAD_DIR ${__dirname}/chunkPath; // chunks 上传的文件夹
const controller new Controller(UPLOAD_DIR);
// ...
case /upload: {await controller.handleUpload(req, res);break;
}功能写合并接口
合并的逻辑跟 Koa 几乎没什么差别只不过我都把方法封装到类里了。
首先写路由
case /merge: {await controller.handleMerge(req, res);break;
}然后在类中定义合并的方法
/*** 功能合并* - 1. handlePostData 用来处理 POST 传递的数据具体怎么处理的请查看方法* - 2. 把各个文件的 path 先想清楚要存到哪儿建议自己写一写。* - 我是把所有的 chunks 都放到大目录 chunkPath 中* - 然后在用文件名新建文件夹再把chunks放到子文件夹中。* param {*} req* param {*} res*/
async handleMerge(req, res) {const postData await handlePostData(req);const { fileName, size: eachChunkSize } postData;const mergePath ${__dirname}/merge;const mergeFilePath ${__dirname}/merge/${fileName};if (!fse.existsSync(mergePath)) {fse.mkdirSync(mergePath);}const mergeOptions { chunksPath: this.chunksPath, mergeFilePath, fileName, eachChunkSize };await handleMergeChunks(mergeOptions);console.log(Success Merge);res.end(JSON.stringify({code: 1,message: success merge,}));
}这里的 POST 请求需要处理一下
/*** 功能处理 POST 请求* - 与 GET 数据相比POST 数据量大需要分段。* - 通过 req.on(data, function(data) {}) 监听* - 当有一段数据到达的时候执行回调回调函数参数 data 为每段达到的数据。* - 当数据全部到达时会触发 req.on(end, function() {}) 里面的回调函数。* - 可以通过 JSON.parse(str) 解析成我们想要的 POST 请求数据格式。* - 参考资料https://juejin.cn/post/7142700338414518286#heading-2* param {*} req* returns*/
function handlePostData(req) {return new Promise((resolve, reject) {let allData ;let i 0;req.on(data, function (chunkData) {// console.log(第 ${i} 次收到数据);allData chunkData;});req.on(end, function () {const POST_MESSAGE JSON.parse(allData);resolve(POST_MESSAGE);});});
}然后就是合并 chunks具体的注释我都放到代码里了
/*** 功能合并 chunks* - 1. 首先根据 fileChunksDir 拿到所有 chunks 的文件名* - 2. 然后拼接成 fileAllChunksPaths Array 数组然后一一创建可写流* - 3. fileAllChunksPaths 注意这里需要排序一下不然就是乱的这也是我们创建可写流 srart 位置的基础* - 4. 然后这里通过 pipeStream 函数用 Promise 包装了一下可读流代码需要慢慢读去理解。* - 5. 我们这里的 可写流们是根据 chunks 的不同定义好写入的文件 path* - 以及每个块儿写的开始位置和写入大小每个可写流都是不一样的* - 6. hasMergeChunk 初始化一下数据* param {*} param0*/
async function handleMergeChunks({ chunksPath, mergeFilePath, fileName, eachChunkSize }) {hasMergeChunk {};const fileChunksDir ${chunksPath}/${fileName};const fileAllChunksPaths await fse.readdir(fileChunksDir);console.log(fileAllChunksPaths);fileAllChunksPaths.sort((a, b) a.split(_)[1] - b.split(_)[1]);const promiseArray fileAllChunksPaths.map((chunk, index, array) {const eachChunkPath ${fileChunksDir}/${chunk};const writeStream fse.createWriteStream(mergeFilePath, {start: index * eachChunkSize,});return pipeStream(eachChunkPath, writeStream, array.length);});await Promise.all(promiseArray);
}把创建写文件流功能也拆分出来。不了解流的概念的话首先了解一下什么是输入可读流。
/*** 功能创建 pipe 写文件流* - 1. [首先了解一下什么是输入可读流](https://www.jmjc.tech/less/111)* - 方便记忆可读流 通过 管道 流入 可写流。 可读流 可写流* - 2. hasMergeChunk 变量用于记录一下那些已经合并完成了也可以写成数组都行。* - 3. 可以检测可读流的 end 事件表示我这个 chunk 已经流完了然后写一下善后逻辑。* param {*} path* param {*} writeStream* returns*/
let hasMergeChunk {};
function pipeStream(path, writeStream, length) {return new Promise((resolve) {const readStream fse.createReadStream(path);readStream.pipe(writeStream);readStream.on(end, function () {hasMergeChunk[path] finished;fse.unlinkSync(path);resolve();console.log(doing: No.${path.split(_)[1]} progress: [ ${Object.keys(hasMergeChunk).length} / ${length} ]);});});
}结语
真的只是收藏不点赞嘛…
上传还有合并这两大功能基本上也就完成啦觉得有用的话请点个赞吧~谢谢吴彦祖们 参考文章
字节跳动面试官请你实现一个大文件上传和断点续传字节跳动面试官我也实现了大文件上传和断点续传
Q A
Q: 发送片段之后的合并可能出现错误
这个情况分析了一下是前端的锅啊前端的 await Promise.all() 并不能保证后端的文件流都写完了。 Q: 进度条直接从 0 到了 100
我发现我的请求写错了 完整代码
前端
!DOCTYPE html
html langenheadmeta charsetUTF-8 /meta nameviewport contentwidthdevice-width, initial-scale1.0 /titleDocument/titlescript srcrequest.js/scriptstyle#uploadCube {margin-top: 10px;/* width: 520px; */overflow: hidden;}.cube {width: 50px;height: 50px;background-color: #fff;float: left;border: 1px solid #000;.progress {height: 100%;line-height: 50px;text-align: center;}.uploading {background-color: #409eff;}.success {background-color: #51f400;}.error {background-color: #ff9090;}}/style/headbodyinput typefile idfile /button iduploadButton点击上传/button!-- button idmergeButton点击合并/button --div iduploadCube/div/bodyscript/*** 默认切片大小*/const SIZE 10 * 1024 * 1024;/*** 功能生成切片*/function handleCreateChunk(file, size SIZE) {const fileChunkList [];progressData [];let cur 0;while (cur file.size) {fileChunkList.push({file: file.slice(cur, cur size),});progressData.push({ presentage: 0 });cur size;}return fileChunkList;}/*** 功能生成页面进度的 HTML*/function handleUpdateHTML(progressData) {let uploadCube document.querySelector(#uploadCube);let html ;progressData.forEach((item) {const { presentage } item;let className ;if (presentage 100) {className progress uploading;} else if (presentage 100) {className progress success;}html div classcubediv class${className} stylewidth: ${presentage}%${presentage}%/div/div;});uploadCube.innerHTML html;}/*** 功能处理每个 chunk 的 xhr.upload.onprogress拿到各个 chunk 的上传进度* - 1. 同时通过 handleUpdateHTML 更新进度页面* - 2. progressData 用来记录各个 chunk 的进度*/let progressData [];function handleCreateOnProgress(data) {return (e) {data.presentage ((e.loaded / e.total) * 100).toFixed(2);console.log(JSON.stringify(progressData));handleUpdateHTML(progressData);};}/*** 功能 上传切片* - 注意 map 里别忘了写 return*/async function uploadChunks(hanldleData, fileName) {const requestList hanldleData.map(({ chunk, hash, index }) {const formData new FormData();formData.append(chunk, chunk);formData.append(hash, hash);formData.append(filename, fileName);return { formData, index };}).map(({ formData, index }) {return request({url: upload,data: formData,onprogress: handleCreateOnProgress(progressData[index]),});});await Promise.all(requestList).then((res) {console.log(所有上传结束, res);});console.log(发送合并请求);await request({url: merge,headers: {content-type: application/json,},data: JSON.stringify({size: SIZE,fileName,}),});}document.getElementById(uploadButton).onclick async function () {// 切片const file document.getElementById(file).files[0];const fileName file.name;const fileChunkList handleCreateChunk(file);// 包装const hanldleData fileChunkList.map(({ file }, index) {return {chunk: file,hash: ${fileName}_${index},index,};});await uploadChunks(hanldleData, fileName);};/script
/html
后端 Koa
const Koa require(koa);
const Views require(koa-views);
const Router require(koa-router);
const Static require(koa-static);
const { koaBody } require(koa-body);
const fse require(fs-extra);const app new Koa();
const router new Router();
app.use(Views(__dirname));
app.use(Static(__dirname));
app.use(koaBody({multipart: true,formidable: {maxFileSize: 1000 * 1024 * 1024,},})
);router.get(/, async (ctx) {await ctx.render(index.html);
});/*** 功能上传接口* - 从 ctx.request.body 中获取 hash 以及 filename* - 从 ctx.request.files 中拿到分片数据* - 然后再把 node 帮我们临时存放的 chunk 文件的 filepath 拿到之后移动到我们想要存放的路径下* - filepath 和 hash 是一一对应的关系*/
router.post(/upload, async (ctx) {const { hash, filename } ctx.request.body;const { filepath } ctx.request.files?.chunk;const chunkPath ${__dirname}/chunkPath/${filename};if (!fse.existsSync(chunkPath)) {await fse.mkdirs(chunkPath);}await fse.move(filepath, ${chunkPath}/${hash});ctx.body {code: 1,};
});/*** 功能创建 pipe 写文件流* - 1. [首先了解一下什么是输入可读流](https://www.jmjc.tech/less/111)* - 2. hasMergeChunk 变量用于记录一下那些已经合并完成了也可以写成数组都行。* - 3. 可以检测可读流的 end 事件表示我这个 chunk 已经流完了然后写一下善后逻辑。* param {*} path* param {*} writeStream* returns*/
let hasMergeChunk {};
function pipeStream(path, writeStream) {return new Promise((resolve) {const readStream fse.createReadStream(path); // 可读流readStream.pipe(writeStream); // 可读流通过管道流向可写流readStream.on(end, () {hasMergeChunk[path] finish;fse.unlinkSync(path); // 删除此文件resolve();console.log(合并 No.${path.split(_)[1]}, 已经合并${Object.keys(hasMergeChunk).length});});});
}/*** 功能合并 Chunk* - 1. chunkDir: 是 chunks 文件们所在的文件夹的路径* - 2. chunkPaths: 是个 Array数组中包含所有的 chunk 的 path* - 3. 因为 每个 chunk 的 path 命名是通过 hash 组成的所以我们先排序一下* - 算是为 createWriteStream 中的 start 做准备* - 4. 为每个 chunk 的 path 创建写入流写到 mergePath 这个路径下。因为已经* - 排序了所以 start 就是每个文件的 index * eachChunkSize* - 5. 每个写入流都用 Promise 包装了一下然后用 await Promise.all() 等待处理完* param {*} mergePath* param {*} name* param {*} eachChunkSize*/
async function mergeChunk(mergePath, name, eachChunkSize) {const chunkDir ${__dirname}/chunkPath/${name};const chunkPaths await fse.readdir(chunkDir);chunkPaths.sort((a, b) a.split(_)[1] - b.split(_)[1]);await Promise.all(chunkPaths.map((chunk, index) {const eachChunkPath ${chunkDir}/${chunk};// 创建输入流并为每个 chunk 定好位置const writeStream fse.createWriteStream(mergePath, {start: index * eachChunkSize,});return pipeStream(eachChunkPath, writeStream);}));console.log(合并完成);fse.rmdirSync(chunkDir);console.log(删除 ${chunkDir} 文件夹);
}/*** 功能: merge 接口* - hasMergeChunk 变量是上面用来记录的* - mergePath 定义一下合并后的文件的路径*/
router.post(/merge, async (ctx) {// console.log(ctx.request.body);const { fileName, size } ctx.request.body;hasMergeChunk {};const mergePath ${__dirname}/merge/${fileName};if (!fse.existsSync(${__dirname}/merge)) {fse.mkdirSync(${__dirname}/merge);}await mergeChunk(mergePath, fileName, size);ctx.body {data: 成功,};
});app.use(router.routes());
app.listen(3000, () {console.log(server start: http://localhost:3000);
});request.js 的封装
/*** 功能封装请求* - 1. xhr.upload.onprogress 注意不要拉下 upload* param {*} param0* returns*/
function request({ url, method post, data, headers {}, onprogress (e) e, requestList }) {return new Promise((resolve, reject) {let xhr new XMLHttpRequest();xhr.open(method, url);Object.keys(headers).forEach((item) {xhr.setRequestHeader(item, headers[item]);});xhr.upload.onprogress onprogress;xhr.onloadend function (e) {resolve({data: e.target.response,});};xhr.send(data);});
}后端原生
主服务
const http require(http);
const server http.createServer();
const url require(url);
const fs require(fs);
const path require(path);
const MIME require(./mime.json);const Controller require(./controller);
const UPLOAD_DIR ${__dirname}/chunkPath; // chunks 上传的文件夹
const controller new Controller(UPLOAD_DIR);server.on(request, async (req, res) {res.setHeader(Access-Control-Allow-Origin, *);res.setHeader(Access-Control-Allow-Headers, *);res.setHeader(content-type, text/html;charsetutf-8);if (req.method OPTION) {res.status 200;res.end();return;}let { pathname } url.parse(req.url); // 解析一下 url因为 req.url 可能会带一些参数// console.log(req.url:, req.url); console.log(url.parse(req.url):, url.parse(req.url));console.log(进入${pathname});switch (pathname) {case /:case /index: {let rs fs.createReadStream(./index.html);rs.pipe(res);break;}case /upload: {await controller.handleUpload(req, res);break;}case /merge: {await controller.handleMerge(req, res);break;}case /favicon.ico: {res.end(我没有哦);break;}default: {let ext path.extname(pathname);res.setHeader(Content-Type, MIME[ext]); // 通过请求的资源后缀名来返回对应的 Content-type 的类型let rs fs.createReadStream(.${pathname});rs.pipe(res);}}
});
const POST 3000;
server.listen(POST, null, null, () {console.log(server start: http://localhost:${POST});
});类
/*** module.expothis.chunksPathrts 方法用于在服务器端导出模块并以 CommonJS 格式提供。* - 参考https://www.delftstack.com/zh/howto/node.js/create-and-export-classes/*/
const multiparty require(multiparty);
const fse require(fs-extra);
const { handlePostData, handleMergeChunks } require(./tools);
class Controller {constructor(dirPath) {this.chunksPath dirPath;}/*** 功能合并* - 1. handlePostData 用来处理 POST 传递的数据具体怎么处理的请查看方法* - 2. 把各个文件的 path 先想清楚要存到哪儿建议自己写一写。* - 我是把所有的 chunks 都放到大目录 chunkPath 中* - 然后在用文件名新建文件夹再把chunks放到子文件夹中。* param {*} req* param {*} res*/async handleMerge(req, res) {const postData await handlePostData(req);const { fileName, size: eachChunkSize } postData;const mergePath ${__dirname}/merge;const mergeFilePath ${__dirname}/merge/${fileName};if (!fse.existsSync(mergePath)) {fse.mkdirSync(mergePath);}const mergeOptions { chunksPath: this.chunksPath, mergeFilePath, fileName, eachChunkSize };await handleMergeChunks(mergeOptions);console.log(Success Merge);res.end(JSON.stringify({code: 1,message: success merge,}));}/*** multiparty 使用方法https://www.npmjs.com/package/multiparty* - chunkFileDirPath 为关于文件 chunks 的文件夹路径每个大文件根据文件名生成相关的文件夹* - 注意回调函数里的 this* param {*} url* param {*} path*/async handleUpload(req, res) {const _this this;const form new multiparty.Form();form.parse(req, async function (err, fields, files) {if (err) {console.log(err);return;}const [chunk] files.chunk;const [hash] fields.hash;const [filename] fields.filename;const chunkFileDirPath ${_this.chunksPath}/${filename};if (!fse.existsSync(chunkFileDirPath)) {await fse.mkdirs(chunkFileDirPath);}await fse.move(chunk?.path, ${chunkFileDirPath}/${hash});res.end(收到文件 chunks);});}
}
module.exports Controller;工具函数
/*** 学习__dirname 就是跟文件一起的不会因为引用关系而恒定*/
// console.log(__dirname);
const fse require(fs-extra);
/*** 功能处理 POST 请求* - 与 GET 数据相比POST 数据量大需要分段。* - 通过 req.on(data, function(data) {}) 监听* - 当有一段数据到达的时候执行回调回调函数参数 data 为每段达到的数据。* - 当数据全部到达时会触发 req.on(end, function() {}) 里面的回调函数。* - 可以通过 JSON.parse(str) 解析成我们想要的 POST 请求数据格式。* - 参考资料https://juejin.cn/post/7142700338414518286#heading-2* param {*} req* returns*/
function handlePostData(req) {return new Promise((resolve, reject) {let allData ;let i 0;req.on(data, function (chunkData) {// console.log(第 ${i} 次收到数据);allData chunkData;});req.on(end, function () {const POST_MESSAGE JSON.parse(allData);resolve(POST_MESSAGE);});});
}/*** 功能创建 pipe 写文件流* - 1. [首先了解一下什么是输入可读流](https://www.jmjc.tech/less/111)* - 方便记忆可读流 通过 管道 流入 可写流。 可读流 可写流* - 2. hasMergeChunk 变量用于记录一下那些已经合并完成了也可以写成数组都行。* - 3. 可以检测可读流的 end 事件表示我这个 chunk 已经流完了然后写一下善后逻辑。* param {*} path* param {*} writeStream* returns*/
let hasMergeChunk {};
function pipeStream(path, writeStream, length) {return new Promise((resolve) {const readStream fse.createReadStream(path);readStream.pipe(writeStream);readStream.on(end, function () {hasMergeChunk[path] finished;fse.unlinkSync(path);resolve();const merging path.split(_)[1];const merged Object.keys(hasMergeChunk).length;console.log(merging: No.${padS(merging)}. progress: [ ${padS(merged)} / ${padS(length)} ]);});});
}/*** 功能合并 chunks* - 1. 首先根据 fileChunksDir 拿到所有 chunks 的文件名* - 2. 然后拼接成 fileAllChunksPaths Array 数组然后一一创建可写流* - 3. fileAllChunksPaths 注意这里需要排序一下不然就是乱的这也是我们创建可写流 srart 位置的基础* - 4. 然后这里通过 pipeStream 函数用 Promise 包装了一下可读流代码需要慢慢读去理解。* - 5. 我们这里的 可写流们是根据 chunks 的不同定义好写入的文件 path* - 以及每个块儿写的开始位置和写入大小每个可写流都是不一样的* -* param {*} param0*/
async function handleMergeChunks({ chunksPath, mergeFilePath, fileName, eachChunkSize }) {hasMergeChunk {};const fileChunksDir ${chunksPath}/${fileName};const fileAllChunksPaths await fse.readdir(fileChunksDir);console.log(fileAllChunksPaths);fileAllChunksPaths.sort((a, b) a.split(_)[1] - b.split(_)[1]);const promiseArray fileAllChunksPaths.map((chunk, index, array) {const eachChunkPath ${fileChunksDir}/${chunk};const writeStream fse.createWriteStream(mergeFilePath, {start: index * eachChunkSize,});return pipeStream(eachChunkPath, writeStream, array.length);});await Promise.all(promiseArray);
}module.exports {handlePostData,handleMergeChunks,
};
