基于Flask SocketIO的异步任务日志实时推送及线程管控方案咨询

原有实现的核心问题 #

你当前使用的死循环轮询方案确实不合理，核心缺陷如下：

• 永久阻塞SocketIO事件处理线程，即使客户端断开连接、任务结束，线程也不会释放，会快速耗尽gunicorn的worker资源
• 固定2秒轮询的延迟不可控，要么延迟高要么数据库查询压力大
• 没有退出机制，会一直占用数据库连接执行无效查询操作

你提出的后台守护线程方案完全可行，落地时需要注意以下实现细节： #

1. 线程生命周期管理 #

• 用线程安全的全局字典存储run_id + 会话sid对应的线程对象和停止标志位，避免同一任务同一会话重复启动线程
• 客户端disconnect事件触发时，将对应线程的停止标志位设为True，线程下次循环检测到标志位就主动退出

# 全局存储结构示例，生产环境可加读写锁保证线程安全
thread_map = {}
lock = threading.Lock()

2. 日志推送逻辑优化 #

• 后台线程每次查询完日志后，先判断任务是否已结束：可在任务表新增status字段，标记任务为运行中/成功/失败，只要任务非运行中且本次查询没有新日志，就主动退出线程
• 轮询间隔可做自适应调整：有新日志时缩短间隔到500ms，连续3次无新日志时拉长到3s，平衡推送延迟和数据库压力
• 推送日志时用socketio.emit指定to=sid参数，确保只推送给对应会话的客户端，避免消息串发

def poll_logs(run_id, sid, stop_event):
    created_at = None
    idle_count = 0
    start_time = datetime.now()
    while not stop_event.is_set():
        # 先判断任务是否已结束
        task_status = task_repo.get_status(run_id)
        logs = run_log_repo.get_logs(run_id, created_at)
        if logs:
            idle_count = 0
            for e in logs:
                socketio.emit('exec-logs', f'run={e["run_id"]}, msg={e["message"]}', to=sid)
                created_at = e['created_at']
        else:
            idle_count += 1
            # 任务结束且无新日志，直接退出
            if task_status != 'running' and idle_count >= 2:
                break
            # 兜底超时：总运行超过15分钟（超过任务最大10分钟的预设时长）强制退出
            if (datetime.now() - start_time).total_seconds() > 15 * 60:
                break
        # 自适应休眠
        time.sleep(0.5 if idle_count < 3 else 3)
    # 退出后清理全局存储
    with lock:
        if (run_id, sid) in thread_map:
            del thread_map[(run_id, sid)]

3. 事件绑定逻辑调整 #

不要在connect事件直接启动线程，应该等客户端发送log-for-run事件携带run_id后再启动，避免客户端连接后不查看日志也浪费线程资源

@socketio.on('log-for-run')
def handle_log_subscribe(run_id):
    sid = request.sid
    stop_event = threading.Event()
    thread = threading.Thread(target=poll_logs, args=(run_id, sid, stop_event), daemon=True)
    with lock:
        # 同一会话同一任务避免重复启动
        if (run_id, sid) not in thread_map:
            thread_map[(run_id, sid)] = (thread, stop_event)
            thread.start()

@socketio.on('disconnect')
def handle_disconnect():
    sid = request.sid
    with lock:
        # 清理当前会话对应的所有线程
        to_delete = [key for key in thread_map if key[1] == sid]
        for key in to_delete:
            _, stop_event = thread_map[key]
            stop_event.set()
            del thread_map[key]

扩展优化建议 #

如果后续要进一步降低数据库查询压力，可以加一层内存消息队列做日志中转：任务生成日志时，除了写入数据库，同时发一条消息到Redis的对应Channel，后台线程订阅对应run_id的Channel，有新日志直接推，没有就阻塞等待，完全避免轮询数据库的开销。

内容的提问来源于stack exchange，提问作者Rag