diff --git a/README.md b/README.md index d400b45..ed472b9 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,9 @@ - ⚡ **秒传功能**：基于文件哈希的秒传检测 - 🗂️ **文件去重**：基于文件哈希的去重机制 - 💾 **数据库存储**：使用 SQLite 存储文件元数据 +- 🚀 **双请求池系统**：活跃请求池 + 待请求池，支持并发控制和智能调度 +- ⚡ **并发优化**：可配置最大并发数，自动管理请求队列 +- 📊 **实时状态监控**：实时显示活跃请求数和等待请求数 ## 安装 @@ -370,6 +373,10 @@ v-hono/ V-Hono 大文件分片上传系统是一个基于 V 语言和 Hono 框架构建的高性能文件上传解决方案。支持大文件分片上传、断点续传、秒传、自动合并等功能。 +**相关文档：** +- [分片上传系统详解](README_CHUNK_UPLOAD.md) - 完整的分片上传功能文档 +- [双请求池系统详解](README_REQUEST_POOL.md) - 请求池系统的设计和实现 + ### 主要特性 - ✅ **分片上传**：支持大文件分片上传，默认分片大小 1MB @@ -380,6 +387,8 @@ V-Hono 大文件分片上传系统是一个基于 V 语言和 Hono 框架构建 - ✅ **数据库存储**：使用 SQLite 存储文件元数据 - ✅ **RESTful API**：完整的 REST API 接口 - ✅ **Web 界面**：提供友好的 Web 上传界面 +- ✅ **双请求池系统**：活跃请求池 + 待请求池，智能并发控制 +- ✅ **实时状态监控**：实时显示上传进度和请求池状态 ### 系统架构 @@ -565,6 +574,99 @@ uploads/ #### 1. 前端 JavaScript 示例 ```javascript +// 请求池管理类 +class RequestPool { + constructor(maxConcurrent = 6) { + this.maxConcurrent = maxConcurrent; + this.activePool = new Map(); // 活跃请求池 {requestId: Promise} + this.pendingPool = []; // 待请求池 [{requestId, task, resolve, reject}] + this.requestIdCounter = 0; + } + + // 生成请求ID + generateRequestId() { + return ++this.requestIdCounter; + } + + // 添加请求到池中 + async addRequest(task) { + const requestId = this.generateRequestId(); + + return new Promise((resolve, reject) => { + const request = { + requestId, + task, + resolve, + reject + }; + + // 如果活跃池未满，直接执行 + if (this.activePool.size < this.maxConcurrent) { + this.executeRequest(request); + } else { + // 否则加入待请求池 + this.pendingPool.push(request); + console.log(`请求 ${requestId} 加入待请求池，当前待请求数: ${this.pendingPool.length}`); + } + }); + } + + // 执行请求 + async executeRequest(request) { + const { requestId, task, resolve, reject } = request; + + console.log(`开始执行请求 ${requestId}，当前活跃请求数: ${this.activePool.size + 1}`); + + // 将请求添加到活跃池 + const promise = task() + .then(result => { + console.log(`请求 ${requestId} 执行成功`); + resolve(result); + return result; + }) + .catch(error => { + console.log(`请求 ${requestId} 执行失败:`, error); + reject(error); + throw error; + }) + .finally(() => { + // 请求完成后从活跃池移除 + this.activePool.delete(requestId); + console.log(`请求 ${requestId} 完成，从活跃池移除，当前活跃请求数: ${this.activePool.size}`); + + // 检查待请求池，如果有待请求则执行 + this.processPendingRequests(); + }); + + this.activePool.set(requestId, promise); + } + + // 处理待请求池中的请求 + processPendingRequests() { + while (this.pendingPool.length > 0 && this.activePool.size < this.maxConcurrent) { + const request = this.pendingPool.shift(); + this.executeRequest(request); + console.log(`从待请求池取出请求 ${request.requestId} 执行，剩余待请求数: ${this.pendingPool.length}`); + } + } + + // 获取池状态 + getStatus() { + return { + activeCount: this.activePool.size, + pendingCount: this.pendingPool.length, + maxConcurrent: this.maxConcurrent + }; + } + + // 清空所有池 + clear() { + this.activePool.clear(); + this.pendingPool = []; + console.log('请求池已清空'); + } +} + // 计算文件哈希 async function calculateFileHash(file) { return new Promise(resolve => { @@ -604,26 +706,40 @@ async function uploadChunk(file, chunk, { fileHash, chunkIndex, chunkHash }) { return false; // 继续上传 } -// 分片上传主函数 -async function uploadFile(file) { +// 使用请求池的分片上传主函数 +async function uploadFileWithPool(file) { const fileHash = await calculateFileHash(file); const chunkSize = 2 * 1024 * 1024; // 2MB const totalChunks = Math.ceil(file.size / chunkSize); + // 创建请求池实例 + const requestPool = new RequestPool(6); + + // 创建所有分片的上传任务并添加到请求池 + const uploadPromises = []; for (let i = 0; i < totalChunks; i++) { - const chunk = file.slice(i * chunkSize, (i + 1) * chunkSize); - const chunkHash = await calculateFileHash(chunk); - - const completed = await uploadChunk(file, chunk, { - fileHash, - chunkIndex: i, - chunkHash + const chunkIndex = i; + const promise = requestPool.addRequest(async () => { + const chunk = file.slice(chunkIndex * chunkSize, (chunkIndex + 1) * chunkSize); + const chunkHash = await calculateFileHash(chunk); + + return await uploadChunk(file, chunk, { + fileHash, + chunkIndex, + chunkHash + }); }); - if (completed) { - break; // 上传完成 - } + uploadPromises.push(promise); } + + // 等待所有上传任务完成 + const results = await Promise.all(uploadPromises); + + // 清空请求池 + requestPool.clear(); + + return results; } ``` @@ -680,17 +796,49 @@ if chunk_count >= int(expected_chunks) && total_chunk_size >= expected_file_size - 上传状态使用内存缓存，提高查询速度 - 支持配置清理策略，自动清理临时文件 -#### 2. 并发处理 +#### 2. 双请求池系统 + +**架构设计：** +- **活跃请求池**：存储正在执行的请求，使用 Map 结构 `{requestId: Promise}` +- **待请求池**：存储等待执行的请求，使用数组结构 `[{requestId, task, resolve, reject}]` +- **智能调度**：当活跃池未满时直接执行，满时加入待请求池 +- **自动管理**：请求完成后自动从活跃池移除，并处理待请求池中的请求 + +**核心特性：** +- **可配置并发数**：默认最大并发数为 6，可根据服务器性能调整 +- **请求ID管理**：每个请求都有唯一ID，便于追踪和调试 +- **状态实时更新**：实时显示活跃请求数和等待请求数 +- **自动清理**：上传完成后自动清空请求池 + +**使用示例：** +```javascript +// 创建请求池实例（最大并发6个） +const requestPool = new RequestPool(6); + +// 添加请求到池中 +const promise = requestPool.addRequest(async () => { + // 上传分片的逻辑 + return await uploadChunk(chunk); +}); + +// 获取池状态 +const status = requestPool.getStatus(); +console.log(`活跃: ${status.activeCount}/${status.maxConcurrent}, 等待: ${status.pendingCount}`); +``` + +#### 3. 并发处理 - 支持多用户同时上传 - 分片级别的并发控制 - 文件级别的锁机制 +- 基于请求池的智能并发管理 -#### 3. 错误处理 +#### 4. 错误处理 - 网络异常自动重试 - 分片损坏自动重新上传 - 完整的错误日志记录 +- 请求池级别的错误隔离 ### 监控和日志 @@ -751,6 +899,12 @@ v run chunk_upload_example.v - 文件权限 - 访问控制 +#### 4. 双请求池系统 + +详细文档请参考： +- [双请求池系统详解](README_REQUEST_POOL.md) - 完整的系统设计和实现指南 +- [分片上传系统](README_CHUNK_UPLOAD.md) - 包含请求池的分片上传完整方案 + ## 性能特性 - **混合路由算法**：静态路由 O(1) 查找，动态路由正则匹配 diff --git a/README_CHUNK_UPLOAD.md b/README_CHUNK_UPLOAD.md index 08de40c..4fe6e9d 100644 --- a/README_CHUNK_UPLOAD.md +++ b/README_CHUNK_UPLOAD.md @@ -6,7 +6,7 @@ V-Hono 大文件分片上传系统是一个基于 V 语言和 Hono 框架构建 ## 主要特性 -- ✅ **分片上传**：支持大文件分片上传，默认分片大小 2MB +- ✅ **分片上传**：支持大文件分片上传，可配置分片大小 - ✅ **断点续传**：支持上传中断后继续上传 - ✅ **秒传功能**：基于文件哈希的秒传检测 - ✅ **自动合并**：所有分片上传完成后自动合并文件 @@ -14,12 +14,17 @@ V-Hono 大文件分片上传系统是一个基于 V 语言和 Hono 框架构建 - ✅ **数据库存储**：使用 SQLite 存储文件元数据 - ✅ **RESTful API**：完整的 REST API 接口 - ✅ **Web 界面**：提供友好的 Web 上传界面 +- ✅ **可配置参数**：支持自定义分片大小、文件大小限制等 +- ✅ **双请求池系统**：活跃请求池 + 待请求池，智能并发控制 +- ✅ **实时状态监控**：实时显示上传进度和请求池状态 ## 系统架构 ``` 前端 (Web/移动端) ↓ +双请求池系统 (活跃池 + 待请求池) + ↓ V-Hono 服务器 ↓ 分片存储 (./uploads/chunks/) @@ -29,6 +34,27 @@ V-Hono 服务器 数据库 (SQLite) ``` +### 双请求池系统架构 + +``` +请求池管理器 +├── 活跃请求池 (Map) +│ ├── 请求1 (正在执行) +│ ├── 请求2 (正在执行) +│ └── ... (最多6个并发) +└── 待请求池 (Array) + ├── 请求7 (等待执行) + ├── 请求8 (等待执行) + └── ... (无限制) +``` + +**工作原理：** +1. 新请求到达时，检查活跃池是否已满 +2. 如果未满，直接加入活跃池并执行 +3. 如果已满，加入待请求池等待 +4. 活跃池中的请求完成后，自动从待请求池取出下一个请求执行 +5. 实时更新状态显示，提供完整的监控信息 + ## 安装和运行 ### 1. 环境要求 @@ -170,6 +196,7 @@ v run chunk_upload_example.v pub struct ChunkUploadConfig { chunk_size: int = 1024 * 1024 // 默认分片大小 1MB max_file_size: int = 1024 * 1024 * 1024 // 最大文件大小 1GB + max_chunk_size: int = 10 * 1024 * 1024 // 最大分片大小 10MB temp_dir: string = './uploads/chunks' // 临时分片目录 upload_dir: string = './uploads/files' // 最终文件目录 cleanup_delay: int = 3600 // 清理延迟时间（秒） @@ -194,11 +221,208 @@ uploads/ └── files.db # SQLite 数据库文件 ``` +## 配置示例 + +### 1. 使用默认配置 + +```v +// 使用默认配置创建分片上传管理器 +mut upload_manager := hono.new_chunk_upload_manager(hono.ChunkUploadConfig{}) +``` + +### 2. 使用自定义配置 + +```v +// 创建自定义配置 +custom_config := hono.ChunkUploadConfig{ + chunk_size: 5 * 1024 * 1024 // 5MB 默认分片大小 + max_file_size: 5 * 1024 * 1024 * 1024 // 5GB 最大文件大小 + max_chunk_size: 20 * 1024 * 1024 // 20MB 最大分片大小 + temp_dir: './uploads/chunks' + upload_dir: './uploads/files' + cleanup_delay: 7200 // 2小时后清理临时文件 + clear_chunks_on_complete: true // 上传完成后清理分片 + db_path: './uploads/files.db' +} + +// 使用自定义配置创建分片上传管理器 +mut upload_manager := hono.new_chunk_upload_manager(custom_config) +``` + +### 3. 获取配置信息 + +前端可以通过 `/api/upload-config` 接口获取后端配置： + +```javascript +// 获取后端配置信息 +async function getUploadConfig() { + const response = await fetch('/api/upload-config'); + const config = await response.json(); + console.log('最大分片大小:', config.max_chunk_size); + return config; +} +``` + +## 双请求池系统详解 + +### 概述 + +双请求池系统是 V-Hono 分片上传系统的核心性能优化组件，通过智能的并发控制机制，确保大文件上传的高效性和稳定性。 + +### 核心组件 + +#### 1. 活跃请求池 (Active Pool) +- **数据结构**：`Map` +- **功能**：存储正在执行的请求 +- **容量**：可配置，默认最大6个并发 +- **特点**：实时执行，状态可追踪 + +#### 2. 待请求池 (Pending Pool) +- **数据结构**：`Array` +- **功能**：存储等待执行的请求 +- **容量**：无限制，按需扩展 +- **特点**：FIFO队列，自动调度 + +#### 3. 请求管理器 +- **请求ID生成**：自动递增的唯一标识 +- **智能调度**：根据活跃池状态自动分配 +- **状态监控**：实时提供池状态信息 +- **错误处理**：请求级别的错误隔离 + +### 配置参数 + +```javascript +// 创建请求池实例 +const requestPool = new RequestPool(6); // 最大并发6个 + +// 可配置参数 +const config = { + maxConcurrent: 6, // 最大并发数 + enableLogging: true, // 启用日志 + autoRetry: true, // 自动重试 + retryCount: 3 // 重试次数 +}; +``` + +### 状态监控 + +```javascript +// 获取实时状态 +const status = requestPool.getStatus(); +console.log({ + activeCount: status.activeCount, // 活跃请求数 + pendingCount: status.pendingCount, // 等待请求数 + maxConcurrent: status.maxConcurrent, // 最大并发数 + utilization: (status.activeCount / status.maxConcurrent * 100).toFixed(1) + '%' // 利用率 +}); +``` + +### 性能优势 + +1. **并发控制**：避免服务器过载，保护系统稳定性 +2. **资源优化**：合理利用网络带宽和服务器资源 +3. **用户体验**：提供实时进度反馈和状态显示 +4. **错误隔离**：单个请求失败不影响其他请求 +5. **自动恢复**：支持断点续传和失败重试 + ## 使用示例 ### 1. 前端 JavaScript 示例 ```javascript +// 请求池管理类 +class RequestPool { + constructor(maxConcurrent = 6) { + this.maxConcurrent = maxConcurrent; + this.activePool = new Map(); // 活跃请求池 {requestId: Promise} + this.pendingPool = []; // 待请求池 [{requestId, task, resolve, reject}] + this.requestIdCounter = 0; + } + + // 生成请求ID + generateRequestId() { + return ++this.requestIdCounter; + } + + // 添加请求到池中 + async addRequest(task) { + const requestId = this.generateRequestId(); + + return new Promise((resolve, reject) => { + const request = { + requestId, + task, + resolve, + reject + }; + + // 如果活跃池未满，直接执行 + if (this.activePool.size < this.maxConcurrent) { + this.executeRequest(request); + } else { + // 否则加入待请求池 + this.pendingPool.push(request); + console.log(`请求 ${requestId} 加入待请求池，当前待请求数: ${this.pendingPool.length}`); + } + }); + } + + // 执行请求 + async executeRequest(request) { + const { requestId, task, resolve, reject } = request; + + console.log(`开始执行请求 ${requestId}，当前活跃请求数: ${this.activePool.size + 1}`); + + // 将请求添加到活跃池 + const promise = task() + .then(result => { + console.log(`请求 ${requestId} 执行成功`); + resolve(result); + return result; + }) + .catch(error => { + console.log(`请求 ${requestId} 执行失败:`, error); + reject(error); + throw error; + }) + .finally(() => { + // 请求完成后从活跃池移除 + this.activePool.delete(requestId); + console.log(`请求 ${requestId} 完成，从活跃池移除，当前活跃请求数: ${this.activePool.size}`); + + // 检查待请求池，如果有待请求则执行 + this.processPendingRequests(); + }); + + this.activePool.set(requestId, promise); + } + + // 处理待请求池中的请求 + processPendingRequests() { + while (this.pendingPool.length > 0 && this.activePool.size < this.maxConcurrent) { + const request = this.pendingPool.shift(); + this.executeRequest(request); + console.log(`从待请求池取出请求 ${request.requestId} 执行，剩余待请求数: ${this.pendingPool.length}`); + } + } + + // 获取池状态 + getStatus() { + return { + activeCount: this.activePool.size, + pendingCount: this.pendingPool.length, + maxConcurrent: this.maxConcurrent + }; + } + + // 清空所有池 + clear() { + this.activePool.clear(); + this.pendingPool = []; + console.log('请求池已清空'); + } +} + // 计算文件哈希 async function calculateFileHash(file) { return new Promise(resolve => { @@ -238,24 +462,61 @@ async function uploadChunk(file, chunk, { fileHash, chunkIndex, chunkHash }) { return false; // 继续上传 } -// 分片上传主函数 -async function uploadFile(file) { +// 使用请求池的分片上传主函数 +async function uploadFileWithPool(file) { const fileHash = await calculateFileHash(file); const chunkSize = 2 * 1024 * 1024; // 2MB const totalChunks = Math.ceil(file.size / chunkSize); + // 创建请求池实例 + const requestPool = new RequestPool(6); + + // 创建所有分片的上传任务并添加到请求池 + const uploadPromises = []; for (let i = 0; i < totalChunks; i++) { - const chunk = file.slice(i * chunkSize, (i + 1) * chunkSize); - const chunkHash = await calculateFileHash(chunk); - - const completed = await uploadChunk(file, chunk, { - fileHash, - chunkIndex: i, - chunkHash + const chunkIndex = i; + const promise = requestPool.addRequest(async () => { + const chunk = file.slice(chunkIndex * chunkSize, (chunkIndex + 1) * chunkSize); + const chunkHash = await calculateFileHash(chunk); + + return await uploadChunk(file, chunk, { + fileHash, + chunkIndex, + chunkHash + }); }); - if (completed) { - break; // 上传完成 + uploadPromises.push(promise); + } + + // 等待所有上传任务完成 + const results = await Promise.all(uploadPromises); + + // 清空请求池 + requestPool.clear(); + + return results; +} + +// 实时监控请求池状态 +function monitorRequestPool(requestPool) { + setInterval(() => { + const status = requestPool.getStatus(); + console.log(`请求池状态: ${status.activeCount}/${status.maxConcurrent} 活跃, ${status.pendingCount} 等待中`); + + // 更新UI显示 + updatePoolStatusDisplay(status); + }, 1000); +} + +// 更新UI显示 +function updatePoolStatusDisplay(status) { + const statusElement = document.getElementById('poolStatus'); + if (statusElement) { + if (status.activeCount === 0 && status.pendingCount === 0) { + statusElement.textContent = '请求池状态: 所有请求已完成'; + } else { + statusElement.textContent = `请求池状态: ${status.activeCount}/${status.maxConcurrent} 活跃, ${status.pendingCount} 等待中`; } } } @@ -278,6 +539,154 @@ curl -X POST http://localhost:8080/upload/chunk \ curl "http://localhost:8080/upload/chunk_exists?file_hash=abc123&chunk_index=0&chunk_hash=def456&file_size=10485760&trunk_size=2097152" ``` +## 请求池最佳实践 + +### 1. 并发数配置建议 + +```javascript +// 根据网络环境调整并发数 +const networkConfig = { + 'fast': 8, // 高速网络：8个并发 + 'normal': 6, // 普通网络：6个并发 + 'slow': 4, // 慢速网络：4个并发 + 'mobile': 3 // 移动网络：3个并发 +}; + +// 根据文件大小调整并发数 +const fileSizeConfig = { + 'small': 8, // 小文件(<10MB)：8个并发 + 'medium': 6, // 中等文件(10MB-100MB)：6个并发 + 'large': 4, // 大文件(>100MB)：4个并发 +}; +``` + +### 2. 错误处理和重试机制 + +```javascript +class RobustRequestPool extends RequestPool { + constructor(maxConcurrent = 6, maxRetries = 3) { + super(maxConcurrent); + this.maxRetries = maxRetries; + } + + async addRequestWithRetry(task) { + return this.addRequest(async () => { + let lastError; + for (let attempt = 1; attempt <= this.maxRetries; attempt++) { + try { + return await task(); + } catch (error) { + lastError = error; + console.log(`请求失败，第${attempt}次重试:`, error.message); + + if (attempt < this.maxRetries) { + // 指数退避 + await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 1000)); + } + } + } + throw lastError; + }); + } +} +``` + +### 3. 性能监控和优化 + +```javascript +class MonitoredRequestPool extends RequestPool { + constructor(maxConcurrent = 6) { + super(maxConcurrent); + this.metrics = { + totalRequests: 0, + successfulRequests: 0, + failedRequests: 0, + averageResponseTime: 0, + startTime: Date.now() + }; + } + + async executeRequest(request) { + const startTime = Date.now(); + this.metrics.totalRequests++; + + try { + const result = await super.executeRequest(request); + this.metrics.successfulRequests++; + this.updateMetrics(Date.now() - startTime); + return result; + } catch (error) { + this.metrics.failedRequests++; + throw error; + } + } + + updateMetrics(responseTime) { + const { totalRequests, averageResponseTime } = this.metrics; + this.metrics.averageResponseTime = + (averageResponseTime * (totalRequests - 1) + responseTime) / totalRequests; + } + + getMetrics() { + const uptime = Date.now() - this.metrics.startTime; + return { + ...this.metrics, + uptime, + successRate: (this.metrics.successfulRequests / this.metrics.totalRequests * 100).toFixed(2) + '%', + requestsPerSecond: (this.metrics.totalRequests / (uptime / 1000)).toFixed(2) + }; + } +} +``` + +### 4. 内存管理和清理 + +```javascript +class ManagedRequestPool extends RequestPool { + constructor(maxConcurrent = 6, maxMemoryMB = 100) { + super(maxConcurrent); + this.maxMemory = maxMemoryMB * 1024 * 1024; // 转换为字节 + this.memoryUsage = 0; + } + + async addRequest(task) { + // 检查内存使用情况 + if (this.memoryUsage > this.maxMemory) { + console.warn('内存使用过高，等待清理...'); + await this.waitForMemory(); + } + + return super.addRequest(task); + } + + async waitForMemory() { + return new Promise(resolve => { + const checkMemory = () => { + if (this.memoryUsage < this.maxMemory * 0.8) { + resolve(); + } else { + setTimeout(checkMemory, 1000); + } + }; + checkMemory(); + }); + } + + // 定期清理 + startCleanup() { + setInterval(() => { + this.cleanup(); + }, 30000); // 每30秒清理一次 + } + + cleanup() { + // 清理过期的请求记录 + const now = Date.now(); + // 实现清理逻辑... + } +} +``` + ## 核心算法 ### 1. 分片合并判断 diff --git a/README_REQUEST_POOL.md b/README_REQUEST_POOL.md new file mode 100644 index 0000000..0b7a711 --- /dev/null +++ b/README_REQUEST_POOL.md @@ -0,0 +1,464 @@ +# V-Hono 双请求池系统 + +## 概述 + +双请求池系统是 V-Hono 框架中用于管理并发请求的核心组件，特别针对大文件分片上传场景进行了优化。该系统通过智能的并发控制机制，确保系统在高负载下的稳定性和性能。 + +## 系统架构 + +### 核心组件 + +``` +双请求池系统 +├── 活跃请求池 (Active Pool) +│ ├── 数据结构: Map +│ ├── 容量: 可配置 (默认6个并发) +│ └── 功能: 存储正在执行的请求 +└── 待请求池 (Pending Pool) + ├── 数据结构: Array + ├── 容量: 无限制 + └── 功能: 存储等待执行的请求 +``` + +### 工作流程 + +1. **请求到达**：新请求到达时，系统生成唯一请求ID +2. **池状态检查**：检查活跃池是否已满 +3. **智能分配**： + - 如果活跃池未满：直接加入活跃池并执行 + - 如果活跃池已满：加入待请求池等待 +4. **自动调度**：活跃池中的请求完成后，自动从待请求池取出下一个请求执行 +5. **状态更新**：实时更新池状态，提供监控信息 + +## 核心特性 + +### 1. 智能并发控制 +- **可配置并发数**：根据服务器性能和网络环境调整 +- **自动负载均衡**：避免服务器过载 +- **资源优化**：合理利用网络带宽和服务器资源 + +### 2. 请求生命周期管理 +- **唯一标识**：每个请求都有唯一的requestId +- **状态追踪**：实时追踪请求的执行状态 +- **自动清理**：请求完成后自动清理资源 + +### 3. 错误处理和隔离 +- **错误隔离**：单个请求失败不影响其他请求 +- **自动重试**：支持配置重试机制 +- **错误传播**：错误信息准确传递给调用方 + +### 4. 实时监控 +- **状态查询**：实时获取池状态信息 +- **性能指标**：提供详细的性能监控数据 +- **日志记录**：完整的操作日志记录 + +## 实现代码 + +### RequestPool 类 + +```javascript +class RequestPool { + constructor(maxConcurrent = 6) { + this.maxConcurrent = maxConcurrent; + this.activePool = new Map(); // 活跃请求池 {requestId: Promise} + this.pendingPool = []; // 待请求池 [{requestId, task, resolve, reject}] + this.requestIdCounter = 0; + } + + // 生成请求ID + generateRequestId() { + return ++this.requestIdCounter; + } + + // 添加请求到池中 + async addRequest(task) { + const requestId = this.generateRequestId(); + + return new Promise((resolve, reject) => { + const request = { + requestId, + task, + resolve, + reject + }; + + // 如果活跃池未满，直接执行 + if (this.activePool.size < this.maxConcurrent) { + this.executeRequest(request); + } else { + // 否则加入待请求池 + this.pendingPool.push(request); + console.log(`请求 ${requestId} 加入待请求池，当前待请求数: ${this.pendingPool.length}`); + } + }); + } + + // 执行请求 + async executeRequest(request) { + const { requestId, task, resolve, reject } = request; + + console.log(`开始执行请求 ${requestId}，当前活跃请求数: ${this.activePool.size + 1}`); + + // 将请求添加到活跃池 + const promise = task() + .then(result => { + console.log(`请求 ${requestId} 执行成功`); + resolve(result); + return result; + }) + .catch(error => { + console.log(`请求 ${requestId} 执行失败:`, error); + reject(error); + throw error; + }) + .finally(() => { + // 请求完成后从活跃池移除 + this.activePool.delete(requestId); + console.log(`请求 ${requestId} 完成，从活跃池移除，当前活跃请求数: ${this.activePool.size}`); + + // 检查待请求池，如果有待请求则执行 + this.processPendingRequests(); + }); + + this.activePool.set(requestId, promise); + } + + // 处理待请求池中的请求 + processPendingRequests() { + while (this.pendingPool.length > 0 && this.activePool.size < this.maxConcurrent) { + const request = this.pendingPool.shift(); + this.executeRequest(request); + console.log(`从待请求池取出请求 ${request.requestId} 执行，剩余待请求数: ${this.pendingPool.length}`); + } + } + + // 获取池状态 + getStatus() { + return { + activeCount: this.activePool.size, + pendingCount: this.pendingPool.length, + maxConcurrent: this.maxConcurrent + }; + } + + // 清空所有池 + clear() { + this.activePool.clear(); + this.pendingPool = []; + console.log('请求池已清空'); + } +} +``` + +## 使用示例 + +### 基本使用 + +```javascript +// 创建请求池实例 +const requestPool = new RequestPool(6); + +// 添加请求 +const promise = requestPool.addRequest(async () => { + // 执行具体的任务 + return await someAsyncTask(); +}); + +// 获取状态 +const status = requestPool.getStatus(); +console.log(`活跃: ${status.activeCount}/${status.maxConcurrent}, 等待: ${status.pendingCount}`); +``` + +### 分片上传场景 + +```javascript +async function uploadFileWithPool(file) { + const fileHash = await calculateFileHash(file); + const chunkSize = 2 * 1024 * 1024; // 2MB + const totalChunks = Math.ceil(file.size / chunkSize); + + // 创建请求池实例 + const requestPool = new RequestPool(6); + + // 创建所有分片的上传任务 + const uploadPromises = []; + for (let i = 0; i < totalChunks; i++) { + const chunkIndex = i; + const promise = requestPool.addRequest(async () => { + const chunk = file.slice(chunkIndex * chunkSize, (chunkIndex + 1) * chunkSize); + const chunkHash = await calculateChunkHash(chunk); + + return await uploadChunk(file, chunk, { + fileHash, + chunkIndex, + chunkHash + }); + }); + + uploadPromises.push(promise); + } + + // 等待所有上传任务完成 + const results = await Promise.all(uploadPromises); + + // 清空请求池 + requestPool.clear(); + + return results; +} +``` + +### 实时监控 + +```javascript +// 实时监控请求池状态 +function monitorRequestPool(requestPool) { + setInterval(() => { + const status = requestPool.getStatus(); + console.log(`请求池状态: ${status.activeCount}/${status.maxConcurrent} 活跃, ${status.pendingCount} 等待中`); + + // 更新UI显示 + updatePoolStatusDisplay(status); + }, 1000); +} + +// 更新UI显示 +function updatePoolStatusDisplay(status) { + const statusElement = document.getElementById('poolStatus'); + if (statusElement) { + if (status.activeCount === 0 && status.pendingCount === 0) { + statusElement.textContent = '请求池状态: 所有请求已完成'; + } else { + statusElement.textContent = `请求池状态: ${status.activeCount}/${status.maxConcurrent} 活跃, ${status.pendingCount} 等待中`; + } + } +} +``` + +## 配置建议 + +### 并发数配置 + +```javascript +// 根据网络环境调整并发数 +const networkConfig = { + 'fast': 8, // 高速网络：8个并发 + 'normal': 6, // 普通网络：6个并发 + 'slow': 4, // 慢速网络：4个并发 + 'mobile': 3 // 移动网络：3个并发 +}; + +// 根据文件大小调整并发数 +const fileSizeConfig = { + 'small': 8, // 小文件(<10MB)：8个并发 + 'medium': 6, // 中等文件(10MB-100MB)：6个并发 + 'large': 4, // 大文件(>100MB)：4个并发 +}; +``` + +### 性能优化配置 + +```javascript +// 高性能配置 +const highPerformanceConfig = { + maxConcurrent: 8, + enableLogging: false, + autoRetry: true, + retryCount: 3, + retryDelay: 1000 +}; + +// 稳定性配置 +const stabilityConfig = { + maxConcurrent: 4, + enableLogging: true, + autoRetry: true, + retryCount: 5, + retryDelay: 2000 +}; +``` + +## 扩展功能 + +### 1. 错误重试机制 + +```javascript +class RobustRequestPool extends RequestPool { + constructor(maxConcurrent = 6, maxRetries = 3) { + super(maxConcurrent); + this.maxRetries = maxRetries; + } + + async addRequestWithRetry(task) { + return this.addRequest(async () => { + let lastError; + for (let attempt = 1; attempt <= this.maxRetries; attempt++) { + try { + return await task(); + } catch (error) { + lastError = error; + console.log(`请求失败，第${attempt}次重试:`, error.message); + + if (attempt < this.maxRetries) { + // 指数退避 + await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 1000)); + } + } + } + throw lastError; + }); + } +} +``` + +### 2. 性能监控 + +```javascript +class MonitoredRequestPool extends RequestPool { + constructor(maxConcurrent = 6) { + super(maxConcurrent); + this.metrics = { + totalRequests: 0, + successfulRequests: 0, + failedRequests: 0, + averageResponseTime: 0, + startTime: Date.now() + }; + } + + async executeRequest(request) { + const startTime = Date.now(); + this.metrics.totalRequests++; + + try { + const result = await super.executeRequest(request); + this.metrics.successfulRequests++; + this.updateMetrics(Date.now() - startTime); + return result; + } catch (error) { + this.metrics.failedRequests++; + throw error; + } + } + + updateMetrics(responseTime) { + const { totalRequests, averageResponseTime } = this.metrics; + this.metrics.averageResponseTime = + (averageResponseTime * (totalRequests - 1) + responseTime) / totalRequests; + } + + getMetrics() { + const uptime = Date.now() - this.metrics.startTime; + return { + ...this.metrics, + uptime, + successRate: (this.metrics.successfulRequests / this.metrics.totalRequests * 100).toFixed(2) + '%', + requestsPerSecond: (this.metrics.totalRequests / (uptime / 1000)).toFixed(2) + }; + } +} +``` + +### 3. 优先级队列 + +```javascript +class PriorityRequestPool extends RequestPool { + constructor(maxConcurrent = 6) { + super(maxConcurrent); + this.priorityQueue = []; + } + + addPriorityRequest(task, priority = 0) { + const requestId = this.generateRequestId(); + + return new Promise((resolve, reject) => { + const request = { + requestId, + task, + resolve, + reject, + priority + }; + + if (this.activePool.size < this.maxConcurrent) { + this.executeRequest(request); + } else { + // 按优先级插入待请求池 + this.insertPriorityRequest(request); + } + }); + } + + insertPriorityRequest(request) { + let insertIndex = this.pendingPool.length; + for (let i = 0; i < this.pendingPool.length; i++) { + if (request.priority > this.pendingPool[i].priority) { + insertIndex = i; + break; + } + } + this.pendingPool.splice(insertIndex, 0, request); + } +} +``` + +## 最佳实践 + +### 1. 合理设置并发数 +- 根据服务器性能调整并发数 +- 考虑网络带宽限制 +- 监控系统资源使用情况 + +### 2. 错误处理 +- 实现适当的重试机制 +- 记录详细的错误日志 +- 提供用户友好的错误信息 + +### 3. 性能监控 +- 实时监控请求池状态 +- 收集性能指标数据 +- 根据监控数据优化配置 + +### 4. 资源管理 +- 及时清理完成的请求 +- 避免内存泄漏 +- 定期检查资源使用情况 + +## 故障排除 + +### 常见问题 + +1. **请求堆积** + - 检查并发数设置是否合理 + - 监控网络连接状态 + - 检查服务器性能 + +2. **内存泄漏** + - 确保请求完成后正确清理 + - 检查是否有循环引用 + - 定期重启服务 + +3. **性能下降** + - 调整并发数配置 + - 检查网络带宽 + - 优化请求处理逻辑 + +### 调试技巧 + +```javascript +// 启用详细日志 +const debugPool = new RequestPool(6); +debugPool.enableDebugLogging = true; + +// 监控关键指标 +setInterval(() => { + const status = debugPool.getStatus(); + const metrics = debugPool.getMetrics(); + console.log('状态:', status); + console.log('指标:', metrics); +}, 5000); +``` + +## 总结 + +双请求池系统是 V-Hono 框架中重要的性能优化组件，通过智能的并发控制和资源管理，为大文件上传等场景提供了稳定、高效的解决方案。通过合理配置和最佳实践，可以充分发挥系统的性能优势。 \ No newline at end of file diff --git a/chunk_upload_example.v b/chunk_upload_example.v index d68572a..84a9583 100644 --- a/chunk_upload_example.v +++ b/chunk_upload_example.v @@ -232,6 +232,20 @@ fn main() { return c.json('{"success": true, "message": "File deleted successfully"}') }) + // 获取上传配置信息 + app.get('/api/upload-config', fn [mut upload_manager] (mut c hono.Context) http.Response { + config := { + 'chunk_size': upload_manager.config.chunk_size.str() + 'max_file_size': upload_manager.config.max_file_size.str() + 'max_chunk_size': upload_manager.config.max_chunk_size.str() + 'temp_dir': upload_manager.config.temp_dir + 'upload_dir': upload_manager.config.upload_dir + 'cleanup_delay': upload_manager.config.cleanup_delay.str() + 'clear_chunks_on_complete': upload_manager.config.clear_chunks_on_complete.str() + } + return c.json(json.encode(config)) + }) + // 静态文件服务 app.use(hono.serve_static_path('/files', './uploads')) diff --git a/example.exe b/example.exe index a0c78da..c6779ed 100644 Binary files a/example.exe and b/example.exe differ diff --git a/example.v b/example.v index 9657d27..5a28a8b 100644 --- a/example.v +++ b/example.v @@ -277,6 +277,20 @@ fn main() { return c.json('{"success": true, "message": "File deleted successfully"}') }) + // 获取上传配置信息 + app.get('/api/upload-config', fn [mut upload_manager] (mut c hono.Context) http.Response { + config := { + 'chunk_size': upload_manager.config.chunk_size.str() + 'max_file_size': upload_manager.config.max_file_size.str() + 'max_chunk_size': upload_manager.config.max_chunk_size.str() + 'temp_dir': upload_manager.config.temp_dir + 'upload_dir': upload_manager.config.upload_dir + 'cleanup_delay': upload_manager.config.cleanup_delay.str() + 'clear_chunks_on_complete': upload_manager.config.clear_chunks_on_complete.str() + } + return c.json(json.encode(config)) + }) + // ===== 5. 基础路由 ===== // 根路径 - 欢迎页面 diff --git a/hono/upload.v b/hono/upload.v index 96b887a..7cf94c7 100644 --- a/hono/upload.v +++ b/hono/upload.v @@ -11,6 +11,7 @@ pub struct ChunkUploadConfig { pub: chunk_size int = 1024 * 1024 // 1MB 默认分片大小 max_file_size int = 1024 * 1024 * 1024 // 1GB 最大文件大小 + max_chunk_size int = 10 * 1024 * 1024 // 10MB 最大分片大小 temp_dir string = './uploads/chunks' // 临时分片目录（分片保存在 temp_dir/filehash/chunksize/ 下） upload_dir string = './uploads/files' // 最终文件目录 cleanup_delay int = 3600 // 1小时后清理临时文件 @@ -115,11 +116,10 @@ pub fn (mut manager ChunkUploadManager) handle_chunk_upload(mut ctx Context) htt } chunk_size := chunk_size_str.int() - // 验证分片大小（允许更大的分片大小，但不超过10MB） - max_allowed_chunk_size := 10 * 1024 * 1024 // 10MB - if chunk_size > max_allowed_chunk_size { + // 验证分片大小（使用配置中的最大分片大小） + if chunk_size > manager.config.max_chunk_size { ctx.status(413) - return ctx.json('{"error": "Chunk size too large", "max_chunk_size": "$max_allowed_chunk_size"}') + return ctx.json('{"error": "Chunk size too large", "max_chunk_size": "${manager.config.max_chunk_size}"}') } // 获取文件数据 diff --git a/public/index.html b/public/index.html index 5d4a04f..bacbf8e 100644 --- a/public/index.html +++ b/public/index.html @@ -450,10 +450,11 @@
-
- +
+
+ -
+
@@ -893,12 +894,122 @@ app.listen(':8080') }); // ===== 分块上传功能 ===== - const CHUNK_SIZE = 2 * 1024 * 1024; // 2MB + const CHUNK_SIZE = 10 * 1024 * 1024; // 10MB 固定分片大小 let selectedFile = null; let fileHash = null; let totalChunks = 0; let uploadedChunks = 0; + // 请求池管理 + class RequestPool { + constructor(maxConcurrent = 6) { + this.maxConcurrent = maxConcurrent; + this.activePool = new Map(); // 活跃请求池 {requestId: Promise} + this.pendingPool = []; // 待请求池 [{requestId, task, resolve, reject}] + this.requestIdCounter = 0; + } + + // 生成请求ID + generateRequestId() { + return ++this.requestIdCounter; + } + + // 添加请求到池中 + async addRequest(task) { + const requestId = this.generateRequestId(); + + return new Promise((resolve, reject) => { + const request = { + requestId, + task, + resolve, + reject + }; + + // 如果活跃池未满，直接执行 + if (this.activePool.size < this.maxConcurrent) { + this.executeRequest(request); + } else { + // 否则加入待请求池 + this.pendingPool.push(request); + console.log(`请求 ${requestId} 加入待请求池，当前待请求数: ${this.pendingPool.length}`); + } + }); + } + + // 执行请求 + async executeRequest(request) { + const { requestId, task, resolve, reject } = request; + + console.log(`开始执行请求 ${requestId}，当前活跃请求数: ${this.activePool.size + 1}`); + + // 将请求添加到活跃池 + const promise = task() + .then(result => { + console.log(`请求 ${requestId} 执行成功`); + resolve(result); + return result; + }) + .catch(error => { + console.log(`请求 ${requestId} 执行失败:`, error); + reject(error); + throw error; + }) + .finally(() => { + // 请求完成后从活跃池移除 + this.activePool.delete(requestId); + console.log(`请求 ${requestId} 完成，从活跃池移除，当前活跃请求数: ${this.activePool.size}`); + + // 检查待请求池，如果有待请求则执行 + this.processPendingRequests(); + + // 触发状态更新事件 + this.triggerStatusUpdate(); + }); + + this.activePool.set(requestId, promise); + } + + // 处理待请求池中的请求 + processPendingRequests() { + while (this.pendingPool.length > 0 && this.activePool.size < this.maxConcurrent) { + const request = this.pendingPool.shift(); + this.executeRequest(request); + console.log(`从待请求池取出请求 ${request.requestId} 执行，剩余待请求数: ${this.pendingPool.length}`); + } + // 处理完待请求后也触发状态更新 + this.triggerStatusUpdate(); + } + + // 获取池状态 + getStatus() { + return { + activeCount: this.activePool.size, + pendingCount: this.pendingPool.length, + maxConcurrent: this.maxConcurrent + }; + } + + // 清空所有池 + clear() { + this.activePool.clear(); + this.pendingPool = []; + console.log('请求池已清空'); + this.triggerStatusUpdate(); + } + + // 触发状态更新 + triggerStatusUpdate() { + // 使用 setTimeout 确保状态更新在下一个事件循环中执行 + setTimeout(() => { + updatePoolStatus(); + }, 0); + } + } + + // 创建全局请求池实例 + const requestPool = new RequestPool(6); + // 文件选择处理 document.getElementById('fileInput').addEventListener('change', handleFileSelect); @@ -955,6 +1066,7 @@ app.listen(':8080') // 重置进度条 document.getElementById('progressFill').style.width = '0%'; document.getElementById('chunkStatus').textContent = ''; + document.getElementById('poolStatus').textContent = ''; // 隐藏进度容器 document.getElementById('progressContainer').style.display = 'none'; @@ -969,6 +1081,9 @@ app.listen(':8080') // 重置变量 uploadedChunks = 0; + + // 清空请求池 + requestPool.clear(); } async function startUpload() { @@ -984,48 +1099,97 @@ app.listen(':8080') document.getElementById('progressContainer').style.display = 'block'; showUploadStatus('开始上传...', 'success'); + // 清空之前的请求池 + requestPool.clear(); + try { - // 直接开始上传分片，每个分片都会检查是否存在 + let completedChunks = 0; + let failedChunks = []; + let instantSuccess = false; + + // 创建所有分片的上传任务并添加到请求池 + const uploadPromises = []; for (let i = 0; i < totalChunks; i++) { - const chunk = selectedFile.slice(i * CHUNK_SIZE, (i + 1) * CHUNK_SIZE); - const chunkHash = await calculateChunkHash(chunk); + const chunkIndex = i; + const promise = requestPool.addRequest(async () => { + try { + const chunk = selectedFile.slice(chunkIndex * CHUNK_SIZE, (chunkIndex + 1) * CHUNK_SIZE); + const chunkHash = await calculateChunkHash(chunk); - // 检查分片是否已存在（包括秒传检测） - const checkResult = await checkChunkExists(fileHash, i, chunkHash); - - // 如果检测到所有分片都已上传完成，立即结束上传 - if (checkResult.all_uploaded) { - console.log('检测到秒传，所有分片都已上传完成！'); - updateProgress(totalChunks, totalChunks); // 更新进度到100% - showUploadStatus('秒传成功！文件已完整上传。', 'success'); - uploadBtn.textContent = '上传完成'; - return; - } - - // 如果分片已存在，跳过上传 - if (checkResult.exists) { - console.log(`分片 ${i} 已存在，跳过上传`); - updateProgress(i + 1, totalChunks); - continue; - } + // 检查分片是否已存在（包括秒传检测） + const checkResult = await checkChunkExists(fileHash, chunkIndex, chunkHash); + + // 如果检测到所有分片都已上传完成，立即结束上传 + if (checkResult.all_uploaded) { + console.log('检测到秒传，所有分片都已上传完成！'); + updateProgress(totalChunks, totalChunks); // 更新进度到100% + showUploadStatus('秒传成功！文件已完整上传。', 'success'); + uploadBtn.textContent = '上传完成'; + instantSuccess = true; + return { success: true, chunkIndex, isInstant: true }; + } + + // 如果分片已存在，跳过上传 + if (checkResult.exists) { + console.log(`分片 ${chunkIndex} 已存在，跳过上传`); + completedChunks++; + updateProgress(completedChunks, totalChunks); + return { success: true, chunkIndex, skipped: true }; + } - // 上传分片 - await uploadChunk(selectedFile, chunk, { - fileHash, chunkIndex: i, totalChunks, chunkHash + // 上传分片 + await uploadChunk(selectedFile, chunk, { + fileHash, chunkIndex, totalChunks, chunkHash + }); + + completedChunks++; + updateProgress(completedChunks, totalChunks); + console.log(`分片 ${chunkIndex} 上传完成`); + return { success: true, chunkIndex }; + } catch (error) { + console.error(`分片 ${chunkIndex} 上传失败:`, error); + failedChunks.push({ chunkIndex, error: error.message }); + return { success: false, chunkIndex, error: error.message }; + } }); - updateProgress(i + 1, totalChunks); - console.log(`分片 ${i} 上传完成`); + uploadPromises.push(promise); } - showUploadStatus('上传完成！文件已自动合并。', 'success'); - uploadBtn.textContent = '上传完成'; + // 等待所有上传任务完成 + const results = await Promise.all(uploadPromises); + + // 检查是否有秒传成功 + if (instantSuccess) { + return; // 秒传成功，直接返回 + } + + // 检查上传结果 + const failedResults = results.filter(result => !result.success); + if (failedResults.length > 0) { + const errorMessage = `部分分片上传失败: ${failedResults.map(f => `分片${f.chunkIndex}(${f.error})`).join(', ')}`; + showUploadStatus(errorMessage, 'error'); + uploadBtn.disabled = false; + uploadBtn.textContent = '重新上传'; + return; + } + + showUploadStatus('上传完成！文件已自动合并。', 'success'); + uploadBtn.textContent = '上传完成'; + + // 确保最终状态显示正确 + setTimeout(() => { + updatePoolStatus(); + }, 100); } catch (error) { console.error('上传失败:', error); showUploadStatus('上传失败: ' + error.message, 'error'); uploadBtn.disabled = false; uploadBtn.textContent = '重新上传'; + } finally { + // 清空请求池 + requestPool.clear(); } } @@ -1119,6 +1283,23 @@ app.listen(':8080') const percentage = (current / total) * 100; document.getElementById('progressFill').style.width = percentage + '%'; document.getElementById('chunkStatus').textContent = current + '/' + total + ' 分片已上传 (' + percentage.toFixed(1) + '%)'; + + // 更新请求池状态 + updatePoolStatus(); + } + + // 更新请求池状态显示 + function updatePoolStatus() { + const status = requestPool.getStatus(); + const poolStatusElement = document.getElementById('poolStatus'); + if (poolStatusElement) { + // 如果所有请求都完成了，显示完成状态 + if (status.activeCount === 0 && status.pendingCount === 0) { + poolStatusElement.textContent = `请求池状态: 所有请求已完成`; + } else { + poolStatusElement.textContent = `请求池状态: ${status.activeCount}/${status.maxConcurrent} 活跃, ${status.pendingCount} 等待中`; + } + } } // 显示上传状态