本文目录导读：
我来为你详细讲解文件分片上传、合并和断点续传的实现方案。
整体架构
客户端（浏览器） 服务器
| |
|---- 1. 初始化上传请求 ------>|
|<--- 返回 uploadId ---------|
| |
|---- 2. 分片上传(chunk) ---->|
|---- 3. 上传进度反馈 -------->|
|---- 4. 断点续传查询 -------->|
|<--- 返回已上传分片列表 ----|
| |
|---- 5. 合并分片请求 -------->|
|<--- 返回完整文件URL ------|前端实现
1 核心上传类
class FileUploader {
constructor(options = {}) {
this.chunkSize = options.chunkSize || 2 * 1024 * 1024; // 默认2MB
this.parallelLimit = options.parallelLimit || 3; // 并发数
this.retryCount = options.retryCount || 3; // 重试次数
this.baseUrl = options.baseUrl || '/api/upload';
this.pendingQueue = [];
this.activeCount = 0;
this.fileHash = null;
this.uploadId = null;
}
// 计算文件哈希
async calculateHash(file) {
const chunkSize = 1024 * 1024; // 1MB
const chunks = Math.ceil(file.size / chunkSize);
const spark = new SparkMD5.ArrayBuffer();
for (let i = 0; i < chunks; i++) {
const start = i * chunkSize;
const end = Math.min(start + chunkSize, file.size);
const blob = file.slice(start, end);
const reader = new FileReader();
await new Promise((resolve) => {
reader.onload = (e) => {
spark.append(e.target.result);
this.updateProgress({
type: 'hash',
loaded: i + 1,
total: chunks
});
resolve();
};
reader.readAsArrayBuffer(blob);
});
}
return spark.end();
}
// 初始化上传
async initUpload(file) {
this.fileHash = await this.calculateHash(file);
const response = await axios.post(`${this.baseUrl}/init`, {
fileName: file.name,
fileSize: file.size,
fileHash: this.fileHash,
chunkSize: this.chunkSize
});
this.uploadId = response.data.uploadId;
return response.data;
}
// 切片文件
sliceFile(file) {
const chunks = [];
let start = 0;
let index = 0;
while (start < file.size) {
const end = Math.min(start + this.chunkSize, file.size);
const blob = file.slice(start, end);
chunks.push({
blob,
index,
start,
end,
size: blob.size
});
start = end;
index++;
}
return chunks;
}
// 上传单个分片
async uploadChunk(chunk, retryCount = 0) {
const formData = new FormData();
formData.append('chunk', chunk.blob);
formData.append('index', chunk.index);
formData.append('uploadId', this.uploadId);
formData.append('fileHash', this.fileHash);
try {
const response = await axios.post(`${this.baseUrl}/chunk`, formData, {
headers: {
'Content-Type': 'multipart/form-data'
},
onUploadProgress: (progressEvent) => {
if (this.onChunkProgress) {
this.onChunkProgress(chunk.index, progressEvent);
}
}
});
return response.data;
} catch (error) {
if (retryCount < this.retryCount) {
// 指数退避重试
await new Promise(resolve => setTimeout(resolve, 1000 * Math.pow(2, retryCount)));
return this.uploadChunk(chunk, retryCount + 1);
}
throw error;
}
}
// 获取已上传分片（断点续传）
async getUploadedChunks() {
try {
const response = await axios.get(`${this.baseUrl}/chunks`, {
params: {
uploadId: this.uploadId,
fileHash: this.fileHash
}
});
return response.data.uploadedChunks || [];
} catch (error) {
return [];
}
}
// 控制并发上传
async uploadWithConcurrency(chunks, uploadedChunks = []) {
const uploadPromises = [];
const uploadedIndices = new Set(uploadedChunks);
//过滤已上传的分片
const remainingChunks = chunks.filter(chunk => !uploadedIndices.has(chunk.index));
for (let i = 0; i < this.parallelLimit && i < remainingChunks.length; i++) {
uploadPromises.push(this.processNextChunk(remainingChunks));
}
let completedChunks = uploadedIndices.size;
const totalChunks = chunks.length;
while (uploadPromises.length > 0) {
const completedPromise = await Promise.race(uploadPromises);
completedChunks++;
// 更新进度
if (this.onProgress) {
this.onProgress({
loaded: completedChunks,
total: totalChunks,
percent: (completedChunks / totalChunks) * 100
});
}
// 移除已完成的任务
const index = uploadPromises.indexOf(completedPromise);
uploadPromises.splice(index, 1);
// 添加新的任务
if (remainingChunks.length > 0) {
uploadPromises.push(this.processNextChunk(remainingChunks));
}
}
return completedChunks;
}
// 处理下一个分片
async processNextChunk(remainingChunks) {
const chunk = remainingChunks.shift();
await this.uploadChunk(chunk);
return chunk.index;
}
// 合并分片
async mergeChunks() {
const response = await axios.post(`${this.baseUrl}/merge`, {
uploadId: this.uploadId,
fileHash: this.fileHash,
fileName: this.file.name,
totalChunks: Math.ceil(this.file.size / this.chunkSize)
});
return response.data;
}
// 主上传方法
async upload(file) {
this.file = file;
try {
// 1. 初始化上传
const initData = await this.initUpload(file);
console.log('Upload initialized:', initData);
// 2. 切片
const chunks = this.sliceFile(file);
// 3. 检查已上传的分片（断点续传）
const uploadedChunks = await this.getUploadedChunks();
if (uploadedChunks.length > 0) {
console.log(`Resuming upload, ${uploadedChunks.length} chunks already uploaded`);
}
// 4. 上传分片
await this.uploadWithConcurrency(chunks, uploadedChunks);
// 5. 合并分片
const result = await this.mergeChunks();
console.log('Upload completed:', result);
return result;
} catch (error) {
console.error('Upload failed:', error);
throw error;
}
}
}
2 使用示例
// HTML部分
/*
<div id="upload-container">
<input type="file" id="fileInput" />
<div id="progress-bar">
<div id="progress-fill"></div>
</div>
<span id="progress-text">0%</span>
<button id="uploadBtn">上传</button>
<button id="pauseBtn">暂停</button>
</div>
*/
// JavaScript部分
const uploader = new FileUploader({
chunkSize: 1024 * 1024, // 1MB
parallelLimit: 3,
retryCount: 3
});
// 设置进度回调
uploader.onProgress = (progress) => {
document.getElementById('progress-fill').style.width = `${progress.percent}%`;
document.getElementById('progress-text').textContent = `${Math.round(progress.percent)}%`;
};
// 暂停功能
let isPaused = false;
document.getElementById('pauseBtn').addEventListener('click', () => {
isPaused = !isPaused;
if (isPaused) {
// 取消当前的请求
axios.CancelToken.source().cancel('Upload paused');
}
});
// 上传按钮
document.getElementById('uploadBtn').addEventListener('click', async () => {
const fileInput = document.getElementById('fileInput');
const file = fileInput.files[0];
if (!file) {
alert('请选择文件');
return;
}
try {
const result = await uploader.upload(file);
alert('上传成功！');
} catch (error) {
if (axios.isCancel(error)) {
console.log('Upload paused');
} else {
alert('上传失败：' + error.message);
}
}
});
后端实现（Node.js + Express）
// server.js
const express = require('express');
const multer = require('multer');
const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const app = express();
app.use(express.json());
// 配置存储
const UPLOAD_DIR = path.join(__dirname, 'uploads');
const TEMP_DIR = path.join(__dirname, 'temp');
// 确保目录存在
if (!fs.existsSync(UPLOAD_DIR)) fs.mkdirSync(UPLOAD_DIR, { recursive: true });
if (!fs.existsSync(TEMP_DIR)) fs.mkdirSync(TEMP_DIR, { recursive: true });
// 存储上传任务信息
const uploadTasks = new Map();
// 配置 multer
const storage = multer.diskStorage({
destination: (req, file, cb) => {
const chunkDir = path.join(TEMP_DIR, req.body.fileHash);
if (!fs.existsSync(chunkDir)) {
fs.mkdirSync(chunkDir, { recursive: true });
}
cb(null, chunkDir);
},
filename: (req, file, cb) => {
cb(null, `${req.body.index}`);
}
});
const upload = multer({ storage });
// 初始化上传
app.post('/api/upload/init', (req, res) => {
const { fileName, fileSize, fileHash, chunkSize } = req.body;
// 生成上传ID
const uploadId = crypto.randomUUID();
// 计算总分片数
const totalChunks = Math.ceil(fileSize / chunkSize);
// 保存任务信息
uploadTasks.set(uploadId, {
fileName,
fileSize,
fileHash,
chunkSize,
totalChunks,
uploadedChunks: [],
createdAt: Date.now()
});
res.json({
uploadId,
totalChunks,
chunkSize
});
});
// 上传分片
app.post('/api/upload/chunk', upload.single('chunk'), (req, res) => {
const { uploadId, index, fileHash } = req.body;
// 更新任务信息
const task = uploadTasks.get(uploadId);
if (task) {
if (!task.uploadedChunks.includes(parseInt(index))) {
task.uploadedChunks.push(parseInt(index));
}
}
res.json({
code: 0,
message: 'Chunk uploaded successfully'
});
});
// 获取已上传的分片
app.get('/api/upload/chunks', (req, res) => {
const { uploadId, fileHash } = req.query;
const task = uploadTasks.get(uploadId);
let uploadedChunks = [];
if (task) {
uploadedChunks = task.uploadedChunks;
} else {
// 检查文件是否已存在（秒传）
const filePath = path.join(UPLOAD_DIR, fileHash);
if (fs.existsSync(filePath)) {
uploadedChunks = 'all'; // 文件已存在
}
}
res.json({
uploadedChunks
});
});
// 合并分片
app.post('/api/upload/merge', async (req, res) => {
const { uploadId, fileHash, fileName, totalChunks } = req.body;
const task = uploadTasks.get(uploadId);
if (!task) {
return res.status(404).json({ error: 'Upload task not found' });
}
const chunkDir = path.join(TEMP_DIR, fileHash);
const filePath = path.join(UPLOAD_DIR, fileHash);
try {
// 创建写入流
const writeStream = fs.createWriteStream(filePath);
// 按顺序合并分片
for (let i = 0; i < totalChunks; i++) {
const chunkPath = path.join(chunkDir, `${i}`);
// 检查分片是否存在
if (!fs.existsSync(chunkPath)) {
writeStream.destroy();
fs.unlinkSync(filePath);
throw new Error(`Chunk ${i} not found`);
}
// 读取并写入分片
const chunkBuffer = fs.readFileSync(chunkPath);
writeStream.write(chunkBuffer);
}
// 完成写入
writeStream.end();
// 等待写入完成
await new Promise((resolve, reject) => {
writeStream.on('finish', resolve);
writeStream.on('error', reject);
});
// 清理临时文件
fs.rmSync(chunkDir, { recursive: true, force: true });
// 清理任务信息
uploadTasks.delete(uploadId);
res.json({
code: 0,
message: 'File merged successfully',
data: {
url: `/files/${fileHash}`,
fileName: fileName
}
});
} catch (error) {
res.status(500).json({
code: -1,
message: error.message
});
}
});
// 文件访问
app.get('/files/:hash', (req, res) => {
const filePath = path.join(UPLOAD_DIR, req.params.hash);
if (fs.existsSync(filePath)) {
res.sendFile(filePath);
} else {
res.status(404).json({ error: 'File not found' });
}
});
// 定期清理过期任务
setInterval(() => {
const now = Date.now();
const maxAge = 24 * 60 * 60 * 1000; // 24小时
for (const [uploadId, task] of uploadTasks) {
if (now - task.createdAt > maxAge) {
// 清理临时文件
const chunkDir = path.join(TEMP_DIR, task.fileHash);
if (fs.existsSync(chunkDir)) {
fs.rmSync(chunkDir, { recursive: true, force: true });
}
uploadTasks.delete(uploadId);
}
}
}, 60 * 60 * 1000); // 每小时检查一次
app.listen(3000, () => {
console.log('Server running on port 3000');
});
关键特性说明
1 断点续传实现
- 客户端先查询已上传分片
- 跳过已上传的分片
- 只上传未完成的分片
2 并发控制
- 限制同时上传的分片数量
- 使用 Promise.race 实现并发控制
- 避免过多并发导致服务器压力
3 错误重试
- 失败自动重试
- 使用指数退避策略
- 设置最大重试次数
4 秒传功能
- 通过文件哈希检查
- 如果文件已存在，直接返回结果
- 节省带宽和时间
注意事项
- 安全考虑：添加文件类型验证、大小限制
- 内存管理：使用流式处理大文件
- 进度反馈：实时显示上传进度
- 错误处理：完善错误处理机制
- 清理机制：定期清理过期任务和临时文件
这个方案可以处理大文件上传，支持断点续传,适合在实际生产环境中使用。
