WebSocket实时通信实现：监控长任务进度更新状态
在构建现代AI系统时，一个常见的痛点是——用户点击“开始”后，页面仿佛卡住，进度条不动，日志无输出，只能干等几分钟甚至几十分钟。这种“黑盒式”的交互体验，不仅让用户焦虑，也让开发者难以快速定位问题。
尤其是在语音合成、图像生成这类计算密集型任务中，批量处理成百上千个文件已是常态。如何让这个过程变得透明、可控、可追踪？答案就是：用 WebSocket 实现长任务的实时状态推送。
我们以GLM-TTS项目中的批量语音合成为例。用户上传一份包含数百条文本和参考音频的任务列表，系统需要依次完成TTS推理、保存音频、最终打包下载。整个流程可能持续数十分钟。若没有实时反馈机制，一旦某个环节出错，用户只能等到全部结束才发现失败。
这时候，传统的HTTP轮询就显得力不从心了：频繁请求服务器消耗资源，延迟高，还容易漏掉关键中间状态。而WebSocket 提供了一条“专属通道”，让服务端可以随时把当前进度、日志信息、错误警告主动“喊出来”，前端即时响应并更新UI。
它的工作方式很像直播间弹幕——连接建立一次，之后双方都能自由发言。客户端不再需要反复发问“好了吗？”；服务器可以在任何时候回答：“第35个文件正在处理”、“警告：参考音频缺失”、“任务已完成，点击下载”。
握手即通，双向通行
WebSocket 的核心优势在于协议升级机制。初始仍走HTTP，但通过特殊的Upgrade头部，将连接从“一次性交易”切换为“长期对话”。一旦握手成功（返回101 Switching Protocols），这条TCP通道就会被保留下来，支持文本或二进制帧的全双工传输。
这意味着：
- 客户端可以发送控制指令，比如启动任务、暂停、取消；
- 服务端能主动推送结构化消息，如{type: 'progress', value: 72.3, current: 'file_078.wav'}；
- 即使网络短暂波动，也能通过心跳机制维持连接稳定性。
相比定时轮询那种“每隔3秒问一次”的笨办法，WebSocket 资源开销更低，响应更及时。尤其在高频小数据量推送场景下，节省的带宽和CPU可达90%以上。
如何集成到 Flask 应用？
在 GLM-TTS 中，我们使用Flask-SocketIO快速搭建 WebSocket 服务。它封装了底层细节，提供简洁的事件驱动接口：
from flask import Flask from flask_socketio import SocketIO, emit import threading app = Flask(__name__) socketio = SocketIO(app, cors_allowed_origins="*") def long_running_task(task_id, file_count): for i in range(file_count): # 模拟处理耗时 time.sleep(1.5) progress = (i + 1) / file_count * 100 log_msg = f"正在处理第 {i+1} 个音频文件..." # 主动推送到前端 socketio.emit('progress_update', { 'task_id': task_id, 'progress': round(progress, 2), 'status': 'running', 'log': log_msg, 'current_file': f'batch_{i+1}.wav' }) # 任务完成通知 socketio.emit('task_complete', { 'task_id': task_id, 'status': 'success', 'output_dir': '/@outputs/batch/', 'download_url': '/download?task=batch_20250405' }) @socketio.on('start_batch_tts') def handle_start(data): task_id = data['task_id'] count = data['file_count'] thread = threading.Thread(target=long_running_task, args=(task_id, count)) thread.start()这段代码的关键点在于：
- 使用独立线程执行耗时任务，避免阻塞主线程；
- 在处理过程中调用socketio.emit()向所有监听客户端广播状态；
- 前端只需注册对应事件监听器即可动态刷新进度条、日志面板等组件。
这正是实现“实时可视化监控”的核心技术路径。
而在任务管理层面，我们采用 JSONL（JSON Lines）格式来组织批量任务。每行一个独立JSON对象，无需逗号分隔，便于流式读取与增量解析：
{"prompt_text": "你好，我是张老师", "prompt_audio": "voices/zhanglaoshi.wav", "input_text": "今天我们要学习语音合成技术", "output_name": "lesson_01"} {"prompt_text": "早上好", "prompt_audio": "voices/wangxiaomei.wav", "input_text": "欢迎收听每日新闻播报", "output_name": "news_daily"}这种设计带来了几个好处：
- 文件可大可小，支持逐行加载而不必一次性载入内存；
- 易于脚本生成，适合自动化流水线集成；
- 出错时能精确定位到具体行号，便于重试或修复；
- 输出命名可控，方便后期归档管理。
当用户上传该文件并点击“开始合成”时，前端通过 WebSocket 发送start_batch_tts事件，携带任务ID与总数量。后端接收到后立即启动后台线程，进入逐条推理流程，并持续回传状态。
整个系统的协作关系如下：
[前端 Web UI] │ ├── HTTP ──→ [Flask App] ←──┐ │ │ └── WebSocket ◄─→ [SocketIO Server] │ [批量推理引擎] │ [TTS 模型推理 pipeline] │ [音频写入 @outputs/]HTTP负责页面渲染与静态资源加载，WebSocket专司状态同步。两者分工明确，各司其职。
实际运行中，用户可以看到清晰的进度展示：
- 实时滚动的日志窗口，显示每一步操作；
- 动态增长的进度条，反映整体完成度；
- 当前处理文件名、已成功/失败数量统计；
- 最终生成ZIP包的下载链接自动弹出。
更重要的是，当某项任务失败时，系统会立即上报错误原因，例如：
[ERROR] Audio file not found: voices/zhanglaoshi.wav无需登录服务器查看日志，用户就能知道是路径配置错误还是文件遗漏，极大提升了调试效率。
当然，在落地过程中也需要考虑一些工程细节：
连接断开怎么办？
WebSocket 并非永不中断。浏览器刷新、网络切换、设备休眠都可能导致连接丢失。为此，我们需要：
- 前端启用自动重连机制；
- 服务端缓存最近任务状态（可用 Redis 存储）；
- 新连接建立后，先查询当前进度并补发最新状态，实现“断点续传”式体验。
如何防止资源耗尽？
TTS 模型通常依赖GPU，显存占用巨大。如果多个批量任务并发执行，极易引发 OOM（内存溢出）。因此必须：
- 限制单次最大任务数（如≤500条）；
- 每个任务完成后调用torch.cuda.empty_cache()释放显存；
- 提供手动“清理显存”按钮，供紧急情况使用；
- 支持串行或有限并行调度，避免资源争抢。
日志要不要分级？
当然要。不是所有消息都需要用户关注。我们将日志分为三级：
-INFO：常规提示，如“开始处理第X个文件”；
-WARN：非致命问题，如“参考文本为空，将使用默认提示”；
-ERROR：导致任务失败的问题，必须干预。
前端可根据用户偏好选择显示级别，避免信息过载。
安全性如何保障？
尽管方便，但开放 WebSocket 也带来潜在风险：
- 必须验证身份，可通过 JWT Token 在握手阶段完成认证；
- 禁止危险路径跳转，如../或绝对路径写入；
- 对输入文件路径做白名单校验，防止目录遍历攻击；
- 设置合理的超时与速率限制，防滥用。
这套架构的价值远不止于语音合成。任何涉及“长时间运行 + 多步骤反馈”的AI任务，都可以借鉴这一模式：
- Stable Diffusion 批量绘图：实时预览每张图像生成进度；
- 视频自动剪辑与配音：显示转码、识别、合成各阶段耗时；
- PDF转有声读物：逐页朗读并记录中断位置；
- 模型训练任务监控：推送 loss 曲线、epoch 进度、GPU利用率。
只要用户需要“看见过程”，WebSocket 就是最合适的工具。
最终的目标，是让复杂的技术隐形于流畅的体验之下。用户不必关心背后有多少线程、是否用了GPU、模型有多大——他们只需要知道：“我的任务正在进行，一切正常”。
而这，正是通过 WebSocket 实现长任务状态监控的核心意义：把不可见的计算过程，变成可视、可感、可信的交互体验。
这种高度集成的设计思路，正引领着智能服务系统向更可靠、更高效的方向演进。
