如何实现进度提示？Super Resolution异步响应开发指南
1. 引言
1.1 业务场景描述
在图像处理类AI应用中，用户上传低分辨率图片后，系统需要执行耗时的超分辨率重建任务。以基于OpenCV EDSR模型的Super Resolution服务为例，3倍放大一张中等尺寸图像通常需要5-15秒。若无任何反馈机制，用户会误以为系统卡顿或未响应，极大影响使用体验。
当前项目已集成WebUI并实现模型持久化部署，但缺乏实时进度提示功能。本文将围绕这一核心痛点，介绍如何在Flask框架下构建支持异步处理与进度追踪的服务架构，提升交互流畅度和用户体验。
1.2 现有方案的不足
原同步处理流程存在以下问题： - 请求期间阻塞主线程，无法响应其他请求 - 前端长时间无反馈，易引发重复提交 - 无法展示处理进度（如“正在放大：40%”） - 大图处理时可能触发HTTP超时
1.3 解决方案预告
本文提出一种轻量级异步响应方案，结合后台任务队列 + 内存状态存储 + 轮询接口，实现完整的进度提示系统。该方案无需引入Redis、Celery等复杂组件，适用于资源受限的边缘设备或轻量化部署环境。
2. 技术方案选型
2.1 可行性方案对比
| 方案 | 实现复杂度 | 扩展性 | 实时性 | 是否需额外依赖 |
|---|---|---|---|---|
| 同步阻塞处理 | ⭐☆☆☆☆ | 差 | 高 | 无 |
| Flask + Threading + 全局字典 | ⭐⭐⭐☆☆ | 中 | 高 | 无 |
| Flask + Celery + Redis | ⭐⭐⭐⭐☆ | 优 | 高 | 需Redis/Celery |
| WebSocket 推送 | ⭐⭐⭐⭐☆ | 中 | 极高 | 需前端适配 |
结论：对于本镜像场景（单机部署、轻量服务），选择Threading + 全局状态字典是最优解。它不增加外部依赖，开发成本低，且能满足基本进度通知需求。
2.2 为什么选择线程+内存状态管理
- 零依赖：复用现有Flask环境，无需安装新包
- 低延迟：状态读写直接操作内存，无网络开销
- 易维护：代码集中，调试方便
- 适合短任务：超分任务一般<30s，线程生命周期可控
3. 核心实现步骤
3.1 环境准备
确保基础依赖已安装：
pip install opencv-contrib-python flask项目目录结构如下：
/superres_app ├── app.py # Flask主程序 ├── superres.py # 超分处理逻辑 ├── static/upload/ # 用户上传图片 ├── static/output/ # 输出高清图 └── models/EDSR_x3.pb # 模型文件（已持久化）3.2 定义全局状态管理器
创建一个线程安全的状态字典，用于跟踪每个任务的进度：
import threading from datetime import datetime # 全局任务状态字典 task_status = {} # 锁保证线程安全 status_lock = threading.Lock() def update_task_status(task_id, status, progress=0, result_url=None, error=None): with status_lock: task_status[task_id] = { "status": status, "progress": progress, "result_url": result_url, "error": error, "update_time": datetime.now().isoformat() } def get_task_status(task_id): return task_status.get(task_id)3.3 封装超分辨率处理函数
# superres.py import cv2 import os class SuperResolution: def __init__(self, model_path): self.sr = cv2.dnn_superres.DnnSuperResImpl_create() self.sr.readModel(model_path) self.sr.setModel("edsr", 3) # x3 放大 def enhance(self, input_path, output_path, task_id): try: update_task_status(task_id, "processing", 10) img = cv2.imread(input_path) if img is None: raise ValueError("无法读取图像文件") update_task_status(task_id, "processing", 30) # 模拟分阶段处理（实际为一步完成） h, w = img.shape[:2] update_task_status(task_id, "processing", 50) upscaled = self.sr.upsample(img) update_task_status(task_id, "processing", 80) cv2.imwrite(output_path, upscaled) result_url = f"/static/output/{os.path.basename(output_path)}" update_task_status(task_id, "completed", 100, result_url=result_url) except Exception as e: update_task_status(task_id, "failed", 0, error=str(e))3.4 Flask路由实现异步处理
# app.py from flask import Flask, request, jsonify, send_from_directory import uuid import threading import os from superres import SuperResolution app = Flask(__name__) sr_engine = SuperResolution("/root/models/EDSR_x3.pb") @app.route("/upload", methods=["POST"]) def upload(): if 'image' not in request.files: return jsonify({"error": "未上传图片"}), 400 file = request.files['image'] if file.filename == '': return jsonify({"error": "文件名为空"}), 400 # 生成唯一任务ID task_id = str(uuid.uuid4()) filename = f"{task_id}.png" input_path = os.path.join("static/upload", filename) output_path = os.path.join("static/output", filename) file.save(input_path) # 更新初始状态 update_task_status(task_id, "pending", 0) # 启动异步处理线程 thread = threading.Thread( target=sr_engine.enhance, args=(input_path, output_path, task_id) ) thread.start() return jsonify({"task_id": task_id}), 202 @app.route("/status/<task_id>") def status(task_id): status_info = get_task_status(task_id) if not status_info: return jsonify({"error": "任务不存在"}), 404 return jsonify(status_info) @app.route("/static/<folder>/<filename>") def serve_file(folder, filename): return send_from_directory(f"static/{folder}", filename)3.5 前端轮询逻辑实现
<!-- 简化版HTML片段 --> <script> async function startUpload() { const formData = new FormData(document.getElementById("uploadForm")); const res = await fetch("/upload", { method: "POST", body: formData }); const data = await res.json(); if (data.task_id) { pollStatus(data.task_id); } } function pollStatus(taskId) { const interval = setInterval(async () => { const res = await fetch(`/status/${taskId}`); const status = await res.json(); document.getElementById("progress").value = status.progress; document.getElementById("statusText").textContent = `状态: ${status.status} (${status.progress}%)`; if (status.status === "completed") { document.getElementById("resultImg").src = status.result_url; clearInterval(interval); } else if (status.status === "failed") { alert("处理失败: " + status.error); clearInterval(interval); } }, 500); // 每500ms查询一次 } </script>4. 实践问题与优化
4.1 实际遇到的问题及解决方案
问题1：内存泄漏风险
长时间运行可能导致task_status字典无限增长。
解决方案：添加自动清理机制
from collections import OrderedDict # 限制最多保存100个历史任务 MAX_TASKS = 100 task_status = OrderedDict() def update_task_status(task_id, status, progress=0, result_url=None, error=None): with status_lock: if task_id in task_status: task_status.pop(task_id) task_status[task_id] = { ... } if len(task_status) > MAX_TASKS: task_status.popitem(last=False) # 移除最老任务问题2：多线程并发冲突
多个请求同时修改task_status可能引发数据错乱。
解决方案：使用threading.Lock()保护共享状态（已在代码中实现）
问题3：大图处理超时
某些浏览器默认连接超时时间为60秒。
优化措施： - 前端设置更长的fetch超时 - 后端压缩输出图像质量减少I/O时间 - 对超大图自动降采样预处理
4.2 性能优化建议
启用OpenCV并行计算
python cv2.setNumThreads(4) # 利用多核CPU缓存模型实例
避免每次请求重新加载模型（本方案已做到）
限制并发数
python semaphore = threading.Semaphore(2) # 最多同时处理2张图在线程启动前加semaphore.acquire()，结束后释放。静态资源Gzip压缩使用Nginx反向代理时开启gzip，减小传输体积。
5. 总结
5.1 实践经验总结
通过本次改造，我们成功为Super Resolution服务增加了进度提示能力，显著提升了用户体验。关键收获包括： -轻量级异步设计：仅用Python原生线程和字典即可实现任务追踪 -状态机思维：明确定义pending → processing → completed/failed状态流转 -前后端协作模式：前端轮询 + 后端状态暴露构成完整闭环
5.2 最佳实践建议
- 任务ID必须全局唯一：推荐使用UUID避免冲突
- 状态更新频率适中：过频更新无意义，每10%-20%进度更新一次即可
- 错误信息友好化：捕获异常并返回可读提示，便于排查问题
- 定期清理过期任务：防止内存持续增长影响稳定性
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
