Vue+Node實(shí)現(xiàn)大文件上傳和斷點(diǎn)續(xù)傳

更新時(shí)間：2022年04月20日 11:24:51 作者：我是哪吒

文件上傳在很多項(xiàng)目中都用的到，如果是幾M的很快就傳送完畢，如果是大文件呢？本文將利用Vue+Node實(shí)現(xiàn)大文件上傳和斷點(diǎn)續(xù)傳，感興趣的可以了解一下

源代碼

斷點(diǎn)續(xù)傳、分片上傳、秒傳、重試機(jī)制

文件上傳是開(kāi)發(fā)中的難點(diǎn)，大文件上傳及斷點(diǎn)續(xù)傳難點(diǎn)中的細(xì)節(jié)及核心技術(shù)點(diǎn)。

element-ui 框架的上傳組件，是默認(rèn)基于文件流的。

數(shù)據(jù)格式：form-data；
傳遞的數(shù)據(jù)： file 文件流信息；filename 文件名字

通過(guò) fileRead.readAsDataURL(file) 轉(zhuǎn)為 base64 字符串后，用 encodeURIComponent 編譯再發(fā)送，發(fā)送的數(shù)據(jù)經(jīng)由 qs.stringify 處理，請(qǐng)求頭添加 "Content-Type": "application/x-www-form-urlencoded"

es6文件對(duì)象、ajax 上傳， async await promise 、后臺(tái)文件存儲(chǔ)、流操作等全面的全棧技能的同時(shí)，提升難度到大文件和斷點(diǎn)續(xù)傳。

移動(dòng)時(shí)代圖片成為社交的主流，短視屏?xí)r代鐵定是大文件。

大文件上傳 8M size 1M 8份

前端上傳大文件時(shí)使用 Blob.prototype.slice 將文件切片，并發(fā)上傳多個(gè)切片，最后發(fā)送一個(gè)合并的請(qǐng)求通知服務(wù)端合并切片
服務(wù)端接收切片并存儲(chǔ)，收到合并請(qǐng)求后使用流將切片合并到最終文件
原生 XMLHttpRequest 的 upload.onprogress 對(duì)切片上傳進(jìn)度的監(jiān)聽(tīng)
使用 Vue 計(jì)算屬性根據(jù)每個(gè)切片的進(jìn)度算出整個(gè)文件的上傳進(jìn)度
使用 spark-md5 根據(jù)文件內(nèi)容算出文件 hash
通過(guò) hash 可以判斷服務(wù)端是否已經(jīng)上傳該文件，從而直接提示用戶(hù)上傳成功（秒傳）
通過(guò) XMLHttpRequest 的 abort 方法暫停切片的上傳
上傳前服務(wù)端返回已經(jīng)上傳的切片名，前端跳過(guò)這些切片的上傳

Blob.slice

Blob.slice() 方法用于創(chuàng)建一個(gè)包含源 Blob的指定字節(jié)范圍內(nèi)的數(shù)據(jù)的新 Blob 對(duì)象。

返回值

一個(gè)新的 Blob 對(duì)象，它包含了原始 Blob 對(duì)象的某一個(gè)段的數(shù)據(jù)。

切片

js 在es6 文件對(duì)象file node file stream 有所增強(qiáng)。

任何文件都是二進(jìn)制，分割blob

start, size, offset

http請(qǐng)求可并發(fā) n個(gè)切片并發(fā)上傳速度更快，改善了體驗(yàn)。

前端的切片，讓http并發(fā)帶來(lái)上傳大文件的快感。

file.slice 完成切片， blob 類(lèi)型文件切片， js 二進(jìn)制文件類(lèi)型的 blob協(xié)議
在文件上傳到服務(wù)器之前就可以提前預(yù)覽。

服務(wù)器端

如何將這些切片，合交成一個(gè)，并且能顯示原來(lái)的圖片
stream 流
可讀流，可寫(xiě)流
chunk 都是一個(gè)二進(jìn)制流文件，
Promise.all 來(lái)包裝每個(gè)chunk 的寫(xiě)入
start end fse.createWriteStream
每個(gè)chunk寫(xiě)入先創(chuàng)建可讀流，再pipe給可寫(xiě)流的過(guò)程

思路：以原文件做為文件夾的名字，在上傳blobs到這個(gè)文件夾，前且每個(gè)blob 都以文件-index的命名方式來(lái)存儲(chǔ)

http并發(fā)上傳大文件切片
vue 實(shí)現(xiàn)上傳文件的細(xì)節(jié)

無(wú)論是前端還是后端，傳輸文件，特別是大文件，有可能發(fā)生丟失文件的情況，網(wǎng)速，服務(wù)器超時(shí)，

如何避免丟失呢？

hash，文件名并不是唯一的，不同名的圖片內(nèi)容是一樣，針對(duì)文件內(nèi)容進(jìn)行hash 計(jì)算
hash 前端算一個(gè)，單向
后端拿到內(nèi)容算hash
一樣，
不一樣重傳
html5特性你怎么理解， localStorage ...

Web Workers 優(yōu)化我們的前端性能，將要花大量時(shí)間的，復(fù)雜的，放到一個(gè)新的線(xiàn)程中去計(jì)算

文件上傳通過(guò)hash 計(jì)算，文件沒(méi)有問(wèn)題

es6 哪些特性，你怎么用的

函數(shù)參數(shù)賦默認(rèn)值

給用戶(hù)快速感知，用戶(hù)體驗(yàn)是核心
并發(fā)http 前后端體驗(yàn)，
斷點(diǎn)續(xù)傳

? 上傳 hash abort 恢復(fù)

初始化文件內(nèi)容

yarn init -y

yarn add -g live-server

// web http方式
lastModified: 1644549553742
lastModifiedDate: Fri Feb 11 20xx 11:19:13 GMT+0800 (中國(guó)標(biāo)準(zhǔn)時(shí)間) {}
name: "banner.png"
size: 138424
type: "image/png"
webkitRelativePath: ""j

yarn add multiparty
// 表單文件上傳

$ vue --version
@vue/cli 4.5.13
vue create vue-upload-big-file

$ vue create vue-upload-big-file
? Please pick a preset: (Use arrow keys)
? Please pick a preset: Manually select features
? Check the features needed for your project: (Press <space> to select, <a> to t
? Check the features needed for your project: Choose Vue version, Babel
? Choose a version of Vue.js that you want to start the project with (Use arrow
? Choose a version of Vue.js that you want to start the project with 2.x
? Where do you prefer placing config for Babel, ESLint, etc.? (Use arrow keys)
> In dedicated config files
? Where do you prefer placing config for Babel, ESLint, etc.? In package.json
? Save this as a preset for future projects? (y/N) n

yarn add element-ui

在生成文件切片時(shí)，需要給每個(gè)切片一個(gè)標(biāo)識(shí)作為hash，這里暫時(shí)使用文件名+下標(biāo)，這樣后端可以知道當(dāng)前切片是第幾個(gè)切片，用于之后的合并切片

隨后調(diào)用uploadChunks上傳所有的文件切片，將文件切片，切片hash，以及文件名放入 formData中，再調(diào)用上一步的request函數(shù)返回一個(gè)promise，最后調(diào)用Promise.all并發(fā)上傳所有的切片

hash，文件名，并不是唯一的.

不同名的圖片，內(nèi)容是一樣。針對(duì)文件內(nèi)容進(jìn)行hash計(jì)算

hash 前端算一個(gè)，單向. 內(nèi)容做hash計(jì)算

后端拿到內(nèi)容算hash一樣。不一樣就要重傳。

web workers 優(yōu)化我們的前端性能，將要花大量時(shí)間的，復(fù)雜的，放到一個(gè)新的線(xiàn)程中去計(jì)算，文件上傳通過(guò)hash去計(jì)算，文件沒(méi)有問(wèn)題。

yarn add fs-extra

FormData.append()

發(fā)送數(shù)據(jù)用到了 FormData

formData.append(name, value, filename)，其中 filename 為可選參數(shù)，是傳給服務(wù)器的文件名稱(chēng)，當(dāng)一個(gè) Blob 或 File 被作為第二個(gè)參數(shù)的時(shí)候， Blob 對(duì)象的默認(rèn)文件名是 "blob"。

大文件上傳

將大文件轉(zhuǎn)換為二進(jìn)制流的格式
利用流可以切割的屬性，將二進(jìn)制流切割成多份
組裝和分割塊同等數(shù)量的請(qǐng)求塊，并行或串行的形式發(fā)出請(qǐng)求
再給服務(wù)器端發(fā)出一個(gè)合并的信息

斷點(diǎn)續(xù)傳

為每個(gè)文件切割塊添加不同的標(biāo)識(shí), hash
當(dāng)上傳成功后，記錄上傳成功的標(biāo)識(shí)
當(dāng)我們暫?；蛘甙l(fā)送失敗后，可以重新發(fā)送沒(méi)有上傳成功的切割文件

代碼

<input
  v-if="!changeDisabled"
  type="file"
  :multiple="multiple"
  class="select-file-input"
  :accept="accept"
  @change="handleFileChange"
/>

創(chuàng)建切片

createFileChunk(file, size = chunkSize) {
const fileChunkList = [];
var count = 0;
while (count < file.size) {
  fileChunkList.push({
	file: file.slice(count, count + size)
  });
  count += size;
}
return fileChunkList;
}

并發(fā)及重試

// 為控制請(qǐng)求并發(fā)的Demo
const sendRequest = (urls, max, callback) => {
  let finished = 0;
  const total = urls.length;
  const handler = () => {
    if (urls.length) {
      const url = urls.shift();
      fetch(url)
        .then(() => {
          finished++;
          handler();
        })
        .catch((err) => {
          throw Error(err);
        });
    }

    if (finished >= total) {
      callback();
    }
  };
  // for控制初始并發(fā)
  for (let i = 0; i < max; i++) {
    handler();
  }
};

const urls = Array.from({ length: 10 }, (v, k) => k);

const fetch = function (idx) {
  return new Promise((resolve) => {
    const timeout = parseInt(Math.random() * 1e4);
    console.log('----請(qǐng)求開(kāi)始');
    setTimeout(() => {
      console.log('----請(qǐng)求結(jié)束');
      resolve(idx);
    }, timeout);
  });
};

const max = 4;

const callback = () => {
  console.log('所有請(qǐng)求執(zhí)行完畢');
};

sendRequest(urls, max, callback);

worker處理，性能及速度都會(huì)有很大提升.

// 生成文件 hash（web-worker）
calculateHash(fileChunkList) {
  return new Promise(resolve => {
    this.container.worker = new Worker('./hash.js');
    this.container.worker.postMessage({ fileChunkList });
    this.container.worker.onmessage = e => {
      const { percentage, hash } = e.data;
      if (this.tempFilesArr[fileIndex]) {
        this.tempFilesArr[fileIndex].hashProgress = Number(
          percentage.toFixed(0)
        );
      }

      if (hash) {
        resolve(hash);
      }
    };
  });
}

文件的合并

mergeRequest(data) {
   const obj = {
     md5: data.fileHash,
     fileName: data.name,
     fileChunkNum: data.chunkList.length
   };

   instance.post('fileChunk/merge', obj, 
     {
       timeout: 0
     })
     .then((res) => {
       this.$message.success('上傳成功');
     });
 }

源碼

methods: {
 handleFileChange(e) {
  const [file] = e.target.files;
  if (!file) return;
  Object.assign(this.$data, this.$options.data());
  this.container.file = file;
},
async handleUpload() {}
}

XMLHttpRequest封裝：

request({
  url,
  method = "post",
  data,
  headers = {},
  requestList
}) {
  return new Promise(resolve => {
	const xhr = new XMLHttpRequest();
	xhr.open(method, url);
	Object.keys(headers).forEach(key =>
	  xhr.setRequestHeader(key, headers[key])
	);
	xhr.send(data);
	xhr.onload = e => {
	  resolve({
		data: e.target.response
	  });
	};
  });
}

上傳切片

對(duì)文件進(jìn)行切片
將切片傳輸給服務(wù)端

const SIZE = 10 * 1024 * 1024; // 切片大小

data: () => ({
	container: {
	  file: null
	}，
	data: []
}),

handleFileChange() {},
// 生成文件切片
createFileChunk(file, size = SIZE) {
	const fileChunkList = [];
	let cur = 0;
	while(cur < file.size) {
		fileChunkList.push({ file: file.slice(cur, cur + size) });
		cur += size;
	}
	return fileChunkList;
},
// 上傳切片
async uploadChunks() {
	const requestList = this.data
	 .map(({ chunk, hash }) => {
		 const formData = new FormData();
		 formData.append("chunk", chunk);
		 formData.append("hash", hash);
		 formData.append("filename", this.container.file.name);
		 return { formData };
	 })
	 .map(async ({ formData }) =>
		this.request({
			url: "http://localhost: 3000",
			data: formData
		})
	 );
	 await Promise.all(requestList); // 并發(fā)切片
},
async handleUpload() {
	if (!this.container.file) return;
	const fileChunkList = this.createFileChunk(this.container.file);
	this.data = fileChunkList.map(({file}, index) => ({
		chunk: file,
		hash: this.container.file.name + '-' + index // 文件名 + 數(shù)組下標(biāo)
	}));
	await this.uploadChunks();
}

發(fā)送合并請(qǐng)求

await Promise.all(requestList);

async mergeRequest() {
	await this.reques({
		url: "http://localhost:3000/merge",
		headers: {
			"content-type": "application/json""
		},
		data: JSON.stringify({
			filename: this.container.file.name
		})
	});
},
async handleUpload() {}

http模塊搭建服務(wù)器：

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 === "OPTIONS") {
    res.status = 200;
    res.end();
    return;
  }
});

server.listen(3000, () => console.log("正在監(jiān)聽(tīng) 3000 端口"));

使用 multiparty 包處理前端傳來(lái)的 FormData

在 multiparty.parse 的回調(diào)中， files 參數(shù)保存了 FormData 中文件， fields 參數(shù)保存了 FormData 中非文件的字段

const UPLOAD_DIR = path.resolve(__dirname, "..", "target"); // 大文件存儲(chǔ)目錄

const multipart = new multiparty.Form();
multipart.parse(req. async(err, fields, files) => {
	if (err) {
		return;
	}
	const [chunk] = files.chunk;
	const [hash] = fields.hash;
	const [filename] = fields.filename;
	const chunkDir = path.resolve(UPLOAD_DIR, filename);
	// 切片目錄不存在，創(chuàng)建切片目錄
	if (!fse.existsSync(chunkDir)) {
		await fse.mkdirs(chunkDir);
	}
	// fs-extra 專(zhuān)用方法，類(lèi)似 fs.rename 并且跨平臺(tái)
	// fs-extra 的 rename 方法 windows 平臺(tái)會(huì)有權(quán)限問(wèn)題
	await fse.move(chunk.path, `${chunkDir}/${hash}`);
	res.end("received file chunk");
});

合并切片

// 在接收到前端發(fā)送的合并請(qǐng)求后，服務(wù)端將文件夾下的所有切片進(jìn)行合并
const resolvePost = req =>
 new Promise(resolve => {
	 let chunk = "";
	 req.on("data", data => {
		 chunk += data;
	 });
	 req.on("end", () => {
		 resolve(JSON.parse(chunk));
	 });
 });
 
const pipeStream = (path, writeStream) =>
 new Promise(resolve => {
	 const readStream = fse.createReadStream(path);
	 readStream.on("end", () => {
		 fse.unlinkSync(path);
		 resolve();
	 });
	 readStream.pipe(writeStream);
 });

// 合并切片
const mergeFileChunk = async (filePath, filename, size) => {
	const chunkDir = path.resolve(UPLOAD_DIR, filename);
	const chunkPaths = await fse.readdir(chunkDir);
	// 根據(jù)切片下標(biāo)進(jìn)行排序
	// 否則直接讀取目錄的獲取的順序可能會(huì)錯(cuò)亂
	chunkPaths.sort((a,b)=>a.split("-")[1] - b.split("-")[1]);
	await Promise.all(
		chunkPaths.map((chunkPath, index) =>
			pipeStream(
				path.resolve(chunkDir, chunkPath),
				// 指定位置創(chuàng)建可寫(xiě)流
				fse.createWriteStream(filePath, {
					start: index * size,
					end: (index + 1) * size
				})
			)
		)
	);
	fse.rmdirSync(chunkDir); // 合并后刪除保存切片的目錄
}


if (req.url === '/merge') {
	const data = await resolvePost(req);
	const { filename, size } = data;
	const filePath = path.resolve(UPLOAD_DIR, `${filename}`);
	await mergeFileChunk(filePath, filename);
	res.end(
		JSON.stringify({
			code: 0,
			message: "file merged success"
		})
	)
}

使用 fs.createWriteStream 創(chuàng)建一個(gè)可寫(xiě)流，可寫(xiě)流文件名就是切片文件夾名 + 后綴名組合

將切片通過(guò) fs.createReadStream 創(chuàng)建可讀流，傳輸合并到目標(biāo)文件中

生成hash

// /public/hash.js
self.importScripts("/spark-md5.min.js"); // 導(dǎo)入腳本

// 生成文件 hash
self.onmessage = e => {
  const { fileChunkList } = e.data;
  const spark = new self.SparkMD5.ArrayBuffer();
  let percentage = 0;
  let count = 0;
  const loadNext = index => {
    const reader = new FileReader();
    reader.readAsArrayBuffer(fileChunkList[index].file);
    reader.onload = e => {
      count++;
      spark.append(e.target.result);
      if (count === fileChunkList.length) {
        self.postMessage({
          percentage: 100,
          hash: spark.end()
        });
        self.close();
      } else {
        percentage += 100 / fileChunkList.length;
        self.postMessage({
          percentage
        });
        // 遞歸計(jì)算下一個(gè)切片
        loadNext(count);
      }
    };
  };
  loadNext(0);
};

worker 線(xiàn)程通訊的邏輯

// 生成文件hash
calculateHash(fileChunkList) {
	return new Promise(resolve => {
		// worker屬性
		this.container.worker = new Worker('/hash.js');
		this.container.worker.postMessage({ fileChunkList });
		this.container.worker.onmessage = e => {
			const { percentage, hash } = e.data;
			this.hashPercentage = percentage;
			if (hash) {
				resolve(hash);
			}
		}
	})
}

文件秒傳

async verifyUpload(filename, fileHash) {
	const { data } = await this.request({
		url: "http://localhost:3000/verify",
		headers: {
			"content-type": "application/json"
		},
		data: JSON.stringify({
			filename,
			fileHash
		})
	});
	return JSON.parse(data);
},
async handleUpload() {
	if (!this.container.file) return;
	const fileChunkList = this.createFileChunk(this.container.file);
	this.container.hash = await this.calculateHash(fileChunkList);
	const { shouldUpload } = await this.verifyUpload(
		this.container.file.name,
		this.container.hash
	);
	if(!shouldUpload) {
		this.$message.success("秒傳：上傳成功");
		return;
	}
	this.data = fileChunkList.map(({file}, index) => ({
		fileHash: this.container.hash,
		index,
		hash: this.container.hash + "-"  + index,
		chunk: file,
		percentage: 0
	}));
	await this.uploadChunks();
}

服務(wù)端：

const extractExt = filename =>
	filename.slice(filename.lastIndexOf("."), filename.length); // 提取后綴名

暫停上傳

request({
	url,
	method = "post",
	data,
	headers = {},
	onProgress = e => e,
	requestList
}) {
	return new Promise(resolve => {
		const xhr = new XMLHttpRequest();
		xhr.upload.onprogress = onProgress;
		xhr.open(method, url);
		Object.keys(headers).forEach(key =>
			xhr.setRequestHeader(key, headers[key])
		);
		xhr.send(data);
		xhr.onload = e => {
			// requestList 中只保存正在上傳切片的 xhr
			// 將請(qǐng)求成功的xhr從列表中刪除
			if (requestList) {
				const xhrIndex = requestList.findIndex(item => item === xhr);
				requestList.splice(xhrIndex, 1);
			}
			resolve({
				data: e.targt.response
			});
		};
		// 暴露當(dāng)前xhr給外部
		requestList?.push(xhr);
	})
}

暫停按鈕

 handlePause() {
    this.requestList.forEach(xhr => xhr?.abort());
    this.requestList = [];
}

前端每次上傳前發(fā)送一個(gè)驗(yàn)證的請(qǐng)求，返回兩種結(jié)果

服務(wù)端已存在該文件，不需要再次上傳
服務(wù)端不存在該文件或者已上傳部分文件切片，通知前端進(jìn)行上傳，并把已上傳的文件切片返回給前端

服務(wù)端驗(yàn)證接口

// 返回已經(jīng)上傳切片名列表
const createUploadedList = async fileHash =>
	fse.existsSync(path.resolve(UPLOAD_DIR, fileHash))
	 ? await fse.readdir(path.resolve(UPLOAD_DIR, fileHash))
	 : [];
	 
if (fse.existsSync(filePath)) {
      res.end(
        JSON.stringify({
          shouldUpload: false
        })
      )
    } else {
      res.end(
        JSON.stringify({
          shouldUpload: true,
          uploadedList: await createUploadedList(fileHash)
        })
      )
    }

點(diǎn)擊上傳時(shí)，檢查是否需要上傳和已上傳的切片
點(diǎn)擊暫停后的恢復(fù)上傳，返回已上傳的切片

async handleResume() {
	this.status = Status.uploading;
	const { uploadedList } = await this.verifyUpload(
		this.container.file.name,
		this.container.hash
	)
	await this.uploadChunks(uploadedList)
},

斷點(diǎn)續(xù)傳

服務(wù)器端返回，告知我從那開(kāi)始
瀏覽器端自行處理

緩存處理

在切片上傳的axios成功回調(diào)中，存儲(chǔ)已上傳成功的切片
在切片上傳前，先看下localstorage中是否存在已上傳的切片,并修改uploaded
構(gòu)造切片數(shù)據(jù)時(shí)，過(guò)濾掉uploaded為true的

垃圾文件清理

前端在localstorage設(shè)置緩存時(shí)間，超過(guò)時(shí)間就發(fā)送請(qǐng)求通知后端清理碎片文件，同時(shí)前端也要清理緩存。
前后端都約定好，每個(gè)緩存從生成開(kāi)始，只能存儲(chǔ)12小時(shí)，12小時(shí)后自動(dòng)清理

(時(shí)間差問(wèn)題)