今天来看一个实用的技术方案：如何通过75-SKill实现用户进度的实时查询功能。这个方案的核心价值在于能够快速构建一个稳定可靠的进度监控系统，特别适合需要实时反馈用户操作进度的应用场景。
75-SKill是一个专注于任务进度监控的技术框架，它提供了完整的API接口和监控机制，能够帮助开发者快速实现用户进度的实时追踪。无论是批量数据处理、文件上传下载，还是复杂的计算任务，都可以通过这个方案来提供实时的进度反馈。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 技术架构 | 基于WebSocket的实时通信 + RESTful API查询 |
| 进度精度 | 支持百分比进度和详细步骤状态 |
| 并发支持 | 多用户同时查询，支持负载均衡 |
| 数据存储 | Redis缓存实时进度 + 数据库持久化 |
| 部署方式 | Docker容器化部署，支持一键启动 |
| 监控维度 | 任务状态、耗时、错误信息、完成百分比 |
2. 适用场景与使用边界
这个方案特别适合以下场景：
- 文件上传/下载进度监控
- 批量数据处理任务跟踪
- AI模型训练进度展示
- 数据导出任务状态查询
- 长时间运行的计算任务监控
使用边界方面需要注意：
- 进度数据需要业务系统主动推送更新
- 实时性依赖于WebSocket连接的稳定性
- 历史进度数据需要定期清理避免存储压力
- 高并发场景需要合理配置Redis和数据库资源
3. 环境准备与前置条件
在开始部署之前，需要确保环境满足以下要求：
系统环境要求：
- 操作系统：Linux Ubuntu 18.04+ / CentOS 7+ / Windows Server 2016+
- 内存：至少4GB可用内存
- 磁盘空间：至少10GB可用空间
- 网络：开放WebSocket端口（默认8080）和API端口（默认8000）
软件依赖：
- Docker 20.10+
- Docker Compose 1.29+
- Redis 6.0+
- MySQL 8.0 或 PostgreSQL 13+
4. 安装部署与启动方式
4.1 Docker快速部署
创建docker-compose.yml配置文件：
version: '3.8' services: redis: image: redis:6.2-alpine ports: - "6379:6379" volumes: - redis_data:/data mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: your_password MYSQL_DATABASE: progress_db ports: - "3306:3306" volumes: - mysql_data:/var/lib/mysql progress-api: image: progress-monitor:latest ports: - "8000:8000" environment: - REDIS_URL=redis://redis:6379 - DATABASE_URL=mysql://root:your_password@mysql:3306/progress_db depends_on: - redis - mysql progress-ws: image: progress-websocket:latest ports: - "8080:8080" environment: - REDIS_URL=redis://redis:6379 depends_on: - redis启动服务：
# 下载镜像并启动服务 docker-compose up -d # 查看服务状态 docker-compose ps # 查看日志 docker-compose logs -f progress-api4.2 手动部署方式
如果需要手动部署，可以按照以下步骤：
# 克隆项目代码 git clone https://github.com/example/75-skill-progress.git cd 75-skill-progress # 安装Python依赖 pip install -r requirements.txt # 配置环境变量 export REDIS_URL="redis://localhost:6379" export DATABASE_URL="mysql://user:pass@localhost/progress_db" # 启动API服务 python app/api_server.py --port 8000 --host 0.0.0.0 # 启动WebSocket服务 python app/websocket_server.py --port 8080 --host 0.0.0.05. 功能测试与效果验证
5.1 创建进度任务测试
首先测试创建进度任务的功能：
import requests import json # 创建进度任务 def create_progress_task(user_id, task_type, total_steps): url = "http://localhost:8000/api/progress/tasks" payload = { "user_id": user_id, "task_type": task_type, "total_steps": total_steps, "description": "测试任务进度监控" } response = requests.post(url, json=payload) if response.status_code == 201: task_data = response.json() print(f"任务创建成功: {task_data['task_id']}") return task_data['task_id'] else: print(f"任务创建失败: {response.text}") return None # 测试创建任务 task_id = create_progress_task("user123", "file_upload", 100)5.2 更新进度状态测试
模拟任务进度更新：
def update_progress(task_id, current_step, status="processing", message=""): url = f"http://localhost:8000/api/progress/tasks/{task_id}" payload = { "current_step": current_step, "status": status, "message": message } response = requests.patch(url, json=payload) if response.status_code == 200: print(f"进度更新成功: {current_step}%") else: print(f"进度更新失败: {response.text}") # 模拟进度更新 for step in range(0, 101, 10): update_progress(task_id, step, "processing", f"当前进度: {step}%")5.3 实时进度查询测试
通过WebSocket实时查询进度：
import asyncio import websockets async def realtime_progress_monitor(task_id): uri = f"ws://localhost:8080/ws/progress/{task_id}" async with websockets.connect(uri) as websocket: print("WebSocket连接建立成功，开始监听进度...") try: while True: message = await websocket.recv() progress_data = json.loads(message) print(f"实时进度: {progress_data['progress']}% - {progress_data['message']}") if progress_data['status'] == 'completed': print("任务完成!") break except websockets.exceptions.ConnectionClosed: print("WebSocket连接已关闭") # 运行实时监控 asyncio.get_event_loop().run_until_complete(realtime_progress_monitor(task_id))6. 接口API与批量任务
6.1 完整的REST API接口
75-SKill提供了一套完整的REST API用于进度管理：
创建任务接口：
POST /api/progress/tasks Content-Type: application/json { "user_id": "string", "task_type": "string", "total_steps": 100, "description": "string" }查询任务进度：
GET /api/progress/tasks/{task_id}批量查询用户任务：
GET /api/progress/users/{user_id}/tasks更新任务进度：
PATCH /api/progress/tasks/{task_id} Content-Type: application/json { "current_step": 50, "status": "processing", "message": "正在处理中" }6.2 批量任务进度监控
对于批量任务，可以使用以下方案：
import concurrent.futures from typing import List class BatchProgressMonitor: def __init__(self, api_base_url: str): self.api_base_url = api_base_url self.tasks = [] def create_batch_tasks(self, user_id: str, task_count: int) -> List[str]: """创建批量任务""" task_ids = [] for i in range(task_count): task_data = { "user_id": user_id, "task_type": "batch_processing", "total_steps": 100, "description": f"批量任务 {i+1}" } response = requests.post( f"{self.api_base_url}/api/progress/tasks", json=task_data ) if response.status_code == 201: task_id = response.json()['task_id'] task_ids.append(task_id) self.tasks.append({ 'task_id': task_id, 'current_progress': 0 }) return task_ids def update_batch_progress(self, task_id: str, progress: int): """更新批量任务进度""" update_url = f"{self.api_base_url}/api/progress/tasks/{task_id}" payload = { "current_step": progress, "status": "processing" if progress < 100 else "completed" } requests.patch(update_url, json=payload) def monitor_batch_progress(self, user_id: str): """监控批量任务总体进度""" url = f"{self.api_base_url}/api/progress/users/{user_id}/tasks" response = requests.get(url) if response.status_code == 200: tasks = response.json()['tasks'] total_progress = sum(task['current_step'] for task in tasks) / len(tasks) print(f"批量任务总体进度: {total_progress:.1f}%") return total_progress7. 资源占用与性能观察
7.1 系统资源监控
在实际部署中，需要关注以下资源指标：
内存占用观察：
# 监控Docker容器内存使用 docker stats progress-api progress-ws redis mysql # 监控系统内存使用 free -h网络连接监控：
# 查看WebSocket连接数 netstat -an | grep 8080 | wc -l # 查看API接口连接数 netstat -an | grep 8000 | wc -l7.2 性能优化建议
根据实际测试，提供以下优化建议：
- Redis配置优化：
# redis.conf 重要配置 maxmemory 1gb maxmemory-policy allkeys-lru save 900 1 save 300 10- 数据库索引优化：
-- 为进度查询添加索引 CREATE INDEX idx_user_task ON progress_tasks(user_id, created_at); CREATE INDEX idx_task_status ON progress_tasks(status);- WebSocket连接管理：
# 设置合理的心跳间隔 websocket_server = WebSocketServer( ping_interval=30, # 30秒心跳 ping_timeout=10, # 10秒超时 max_connections=1000 # 最大连接数 )8. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| WebSocket连接失败 | 端口被占用或防火墙阻止 | 检查端口8080是否开放 | 修改端口或配置防火墙 |
| 进度更新无响应 | Redis连接失败 | 检查Redis服务状态 | 重启Redis或检查连接配置 |
| API接口返回404 | 服务未正常启动 | 查看容器日志 | 重新部署服务 |
| 进度数据丢失 | 数据库连接异常 | 检查数据库连接状态 | 修复数据库连接配置 |
| 实时推送延迟 | 网络带宽不足 | 监控网络流量 | 优化网络配置或升级带宽 |
8.1 详细排查步骤
WebSocket连接问题排查：
# 检查端口监听状态 netstat -tlnp | grep 8080 # 测试WebSocket连接 curl -i -N -H "Connection: Upgrade" -H "Upgrade: websocket" \ -H "Host: localhost:8080" -H "Origin: http://localhost" \ http://localhost:8080/ws/progress/testRedis连接问题排查：
# 测试Redis连接 redis-cli -h localhost -p 6379 ping # 查看Redis内存使用 redis-cli info memory数据库连接问题排查：
# 测试MySQL连接 mysql -h localhost -P 3306 -u root -p -e "SELECT 1;" # 检查数据库表状态 mysql -u root -p -e "USE progress_db; SHOW TABLES;"9. 最佳实践与使用建议
9.1 进度更新策略
合理的进度更新策略可以提升用户体验：
class ProgressUpdateStrategy: def __init__(self, min_interval=1.0, max_updates=100): self.min_interval = min_interval # 最小更新间隔(秒) self.max_updates = max_updates # 最大更新次数 self.last_update_time = 0 def should_update(self, current_progress, total_steps): """判断是否应该更新进度""" current_time = time.time() time_elapsed = current_time - self.last_update_time # 进度完成时强制更新 if current_progress >= total_steps: return True # 基于时间和进度的智能更新 progress_percentage = current_progress / total_steps expected_updates = min(self.max_updates, total_steps) target_interval = total_steps / expected_updates return (time_elapsed >= self.min_interval and current_progress % target_interval < 1)9.2 错误处理与重试机制
健壮的错误处理是保证系统稳定性的关键：
import time from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry def create_retry_session(retries=3, backoff_factor=0.3): """创建带重试机制的HTTP会话""" session = requests.Session() retry_strategy = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session def safe_progress_update(session, task_id, progress_data): """安全的进度更新方法""" max_retries = 3 for attempt in range(max_retries): try: response = session.patch( f"http://localhost:8000/api/progress/tasks/{task_id}", json=progress_data, timeout=5 ) response.raise_for_status() return True except requests.exceptions.RequestException as e: if attempt == max_retries - 1: print(f"进度更新失败: {e}") return False time.sleep(2 ** attempt) # 指数退避10. 扩展功能与高级用法
10.1 进度预估算法
基于历史数据的智能进度预估：
class ProgressEstimator: def __init__(self): self.history = [] def add_sample(self, elapsed_time, progress): """添加进度样本""" self.history.append((elapsed_time, progress)) # 保持最近100个样本 if len(self.history) > 100: self.history.pop(0) def estimate_remaining_time(self, current_progress, current_elapsed): """预估剩余时间""" if not self.history or current_progress == 0: return None # 使用加权平均计算平均速度 total_weight = 0 weighted_speed = 0 for i, (time, progress) in enumerate(self.history): weight = 1.0 / (len(self.history) - i) # 近期样本权重更高 if progress > 0: speed = progress / time weighted_speed += speed * weight total_weight += weight if total_weight > 0: avg_speed = weighted_speed / total_weight remaining_progress = 100 - current_progress return remaining_progress / avg_speed if avg_speed > 0 else None return None10.2 多维度进度监控
支持复杂任务的多维度进度跟踪：
class MultiDimensionalProgress: def __init__(self, task_id, dimensions): self.task_id = task_id self.dimensions = dimensions self.progress_data = {dim: 0 for dim in dimensions} def update_dimension(self, dimension, progress): """更新单个维度进度""" if dimension in self.dimensions: self.progress_data[dimension] = progress self._update_overall_progress() def _update_overall_progress(self): """计算整体进度""" total_progress = sum(self.progress_data.values()) / len(self.dimensions) # 更新到进度系统 update_progress(self.task_id, total_progress, message=f"多维度进度: {self.progress_data}") def get_progress_summary(self): """获取进度摘要""" return { 'overall': sum(self.progress_data.values()) / len(self.dimensions), 'details': self.progress_data }75-SKill实时进度查询方案提供了一个完整的技术栈来解决用户进度监控的需求。通过合理的架构设计和优化配置，这个方案可以支撑从中小型项目到大型企业级应用的各种场景。关键在于根据实际业务需求调整更新频率、数据存储策略和监控粒度，确保系统既实时又稳定。
