MathModelAgent如何实现任务队列Redis发布订阅机制的深度解析【免费下载链接】MathModelAgent专为数学建模设计的 Agent skills ,自动完成数学建模生成一份完整的可以直接提交的论文。 An Agent Designed for Mathematical Modeling ,Automatically complete mathmodel and generate a complete paper ready for submission.项目地址: https://gitcode.com/GitHub_Trending/ma/MathModelAgentMathModelAgent 是一款专为数学建模场景设计的 Agent 项目它能自动完成选题分析、建模、编码求解到论文写作的全流程。由于一个建模任务往往要运行数小时项目必须解决两个核心问题任务如何排队执行执行进度如何实时推送给前端答案正是本文要拆解的Redis 发布订阅Pub/Sub WebSocket 任务队列机制。一、整体架构一次提交即返回的异步任务传统 Web 服务里用户提交一个耗时任务后HTTP 连接要一直挂着等待结果——这对动辄数小时的建模任务完全不可行。MathModelAgent 采用的是经典的异步任务队列模式入队用户提交任务后端立即生成task_id并创建独立工作目录快速响应接口马上返回{task_id: ..., status: processing}HTTP 连接即刻释放后台执行真正的建模流程被丢进后台通过 Redis 频道持续广播进度实时消费前端通过 WebSocket 订阅该任务的频道逐条接收进度消息。入口逻辑位于 modeling_router.py提交接口的关键动作只有三步await redis_manager.set(ftask_id:{task_id}, task_id) # 任务注册到 Redis background_tasks.add_task(run_modeling_task_async, ...) # 任务入队后台执行 return {task_id: task_id, status: processing} # 立即返回这里有个巧妙的设计task_id:{task_id}这个键写入 Redis 后设置了36000 秒10 小时过期时间它既是任务是否存在的凭证也天然自动清理僵尸任务无需额外维护。二、任务如何执行asyncio 任务注册表任务真正跑起来的地方是run_modeling_task_async。它做了三件关键的事发布任务开始消息到 Redis 频道让前端立刻看到状态变化asyncio.create_task启动建模工作流并设置 5 小时超时asyncio.wait_for将(asyncio.Task, asyncio.Event)注册进全局字典_active_tasks——这正是实现一键停止任务的基础。任务结束时无论成功、取消还是异常都会再发布一条对应的状态消息success/warning/error保证前端状态永远有收尾。三、核心机制Redis 发布订阅频道设计消息发布的核心封装在 redis_manager.py 中每个任务对应一个独立频道频道命名规则task:{task_id}:messagespublish_message方法做了双写——既PUBLISH到 Redis 频道实现实时推送又把消息追加保存到logs/messages/{task_id}.json文件做持久化。为什么需要持久化因为 Redis Pub/Sub 的消息不落地如果用户刷新页面时 WebSocket 还没连上中间的消息就丢了。文件兜底后前端随时可以通过 common_router.py 的GET /messages?task_id...接口拉取完整历史消息实现断点续看。消息体本身由 response.py 中一组 Pydantic 模型定义按msg_type区分为 system系统状态、agent四个 Agent 的输出、tool代码执行结果等类型前端据此渲染不同样式的气泡与 Notebook 单元格。四、WebSocket 订阅端从 Redis 到浏览器的最后一跳ws_router.py 定义了WebSocket /task/{task_id}端点它的工作流程是校验任务检查task_id:{task_id}键是否存在不存在直接关闭连接防止访问非法任务订阅频道调用redis_manager.subscribe_to_task()拿到该任务的 Pub/Sub 订阅句柄消息转发循环以 0.1 秒为节拍轮询pubsub.get_message()解析 JSON 后原样推送给浏览器优雅退出连接关闭时取消订阅并移除连接记录由 ws_manager.py 统一管理活跃连接。还有一个体现细节的小设计任务启动后会有await asyncio.sleep(1)的短暂延迟——这是为了给前端 WebSocket 留出连接窗口避免开始处理这条首条消息因没人订阅而丢失尽管有文件兜底实时性上依然尽力保证。五、前端配套指数退避自动重连浏览器侧的 websocket.ts 实现了TaskWebSocket类它不只是简单建连还内置了一套生产级的重连策略断线后自动重连最多10 次采用指数退避间隔从 1 秒翻倍递增上限 30 秒避免服务端故障时疯狂重试打爆服务手动关闭close()时置位isManualClose标记跳过自动重连逻辑。配合后端的文件持久化 /messages历史接口即使长任务中途断网重连用户刷新页面也能恢复完整进度这正是任务队列体验的最后一公里。六、一键停止任务取消信号的传递建模任务跑偏了怎么办项目通过_active_tasks注册表 asyncio.Event实现协作式取消_, cancel_event _active_tasks[task_id] cancel_event.set() # 仅发信号工作流内部周期性检查并自行优雅退出POST /modeling/{task_id}/cancel接口只负责发信号工作流执行到下一个检查点时感知到事件后自行收尾从而避免粗暴杀进程导致的资源泄漏。七、部署细节一条 Docker Compose 拉起 Redis在 docker-compose.yml 中Redis 以redis:alpine镜像运行并挂载持久化卷后端通过depends_on保证 Redis 先行启动。连接地址由 setting.py 中的两个配置项控制REDIS_URL默认redis://redis:6379/0Docker 网络内服务发现REDIS_MAX_CONNECTIONS连接池上限默认 10适配单机中小并发场景。此外GET /status健康检查接口会实时 ping Redis 并返回后端与缓存服务的运行状态方便运维观测。总结这套任务队列方案值得借鉴的 4 个点设计点实现方式解决的问题异步任务入队FastAPI BackgroundTasks asyncio.Task长任务不阻塞 HTTP进度广播Redis Pub/Sub 按任务隔离频道多任务并发、互不干扰消息双写Pub/Sub 实时推送 JSON 文件持久化断线不丢进度、可回溯优雅取消asyncio.Event 协作式信号随时停止且资源安全回收整体来看MathModelAgent 用Redis 发布订阅 WebSocket 文件持久化三件套以极低的架构复杂度支撑起了多任务并发、实时进度、断线恢复与任务取消四大能力是学习异步任务队列设计的一个非常贴切、可读的开源范例。相关源码集中在 backend/app/services/ 与 backend/app/routers/ 目录配合 docs/tutorial.md 教程可以快速上手本地部署。【免费下载链接】MathModelAgent专为数学建模设计的 Agent skills ,自动完成数学建模生成一份完整的可以直接提交的论文。 An Agent Designed for Mathematical Modeling ,Automatically complete mathmodel and generate a complete paper ready for submission.项目地址: https://gitcode.com/GitHub_Trending/ma/MathModelAgent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 MathModelAgent如何实现任务队列？Redis发布订阅机制的深度解析
MathModelAgent如何实现任务队列Redis发布订阅机制的深度解析【免费下载链接】MathModelAgent专为数学建模设计的 Agent skills ,自动完成数学建模生成一份完整的可以直接提交的论文。 An Agent Designed for Mathematical Modeling ,Automatically complete mathmodel and generate a complete paper ready for submission.项目地址: https://gitcode.com/GitHub_Trending/ma/MathModelAgentMathModelAgent 是一款专为数学建模场景设计的 Agent 项目它能自动完成选题分析、建模、编码求解到论文写作的全流程。由于一个建模任务往往要运行数小时项目必须解决两个核心问题任务如何排队执行执行进度如何实时推送给前端答案正是本文要拆解的Redis 发布订阅Pub/Sub WebSocket 任务队列机制。一、整体架构一次提交即返回的异步任务传统 Web 服务里用户提交一个耗时任务后HTTP 连接要一直挂着等待结果——这对动辄数小时的建模任务完全不可行。MathModelAgent 采用的是经典的异步任务队列模式入队用户提交任务后端立即生成task_id并创建独立工作目录快速响应接口马上返回{task_id: ..., status: processing}HTTP 连接即刻释放后台执行真正的建模流程被丢进后台通过 Redis 频道持续广播进度实时消费前端通过 WebSocket 订阅该任务的频道逐条接收进度消息。入口逻辑位于 modeling_router.py提交接口的关键动作只有三步await redis_manager.set(ftask_id:{task_id}, task_id) # 任务注册到 Redis background_tasks.add_task(run_modeling_task_async, ...) # 任务入队后台执行 return {task_id: task_id, status: processing} # 立即返回这里有个巧妙的设计task_id:{task_id}这个键写入 Redis 后设置了36000 秒10 小时过期时间它既是任务是否存在的凭证也天然自动清理僵尸任务无需额外维护。二、任务如何执行asyncio 任务注册表任务真正跑起来的地方是run_modeling_task_async。它做了三件关键的事发布任务开始消息到 Redis 频道让前端立刻看到状态变化asyncio.create_task启动建模工作流并设置 5 小时超时asyncio.wait_for将(asyncio.Task, asyncio.Event)注册进全局字典_active_tasks——这正是实现一键停止任务的基础。任务结束时无论成功、取消还是异常都会再发布一条对应的状态消息success/warning/error保证前端状态永远有收尾。三、核心机制Redis 发布订阅频道设计消息发布的核心封装在 redis_manager.py 中每个任务对应一个独立频道频道命名规则task:{task_id}:messagespublish_message方法做了双写——既PUBLISH到 Redis 频道实现实时推送又把消息追加保存到logs/messages/{task_id}.json文件做持久化。为什么需要持久化因为 Redis Pub/Sub 的消息不落地如果用户刷新页面时 WebSocket 还没连上中间的消息就丢了。文件兜底后前端随时可以通过 common_router.py 的GET /messages?task_id...接口拉取完整历史消息实现断点续看。消息体本身由 response.py 中一组 Pydantic 模型定义按msg_type区分为 system系统状态、agent四个 Agent 的输出、tool代码执行结果等类型前端据此渲染不同样式的气泡与 Notebook 单元格。四、WebSocket 订阅端从 Redis 到浏览器的最后一跳ws_router.py 定义了WebSocket /task/{task_id}端点它的工作流程是校验任务检查task_id:{task_id}键是否存在不存在直接关闭连接防止访问非法任务订阅频道调用redis_manager.subscribe_to_task()拿到该任务的 Pub/Sub 订阅句柄消息转发循环以 0.1 秒为节拍轮询pubsub.get_message()解析 JSON 后原样推送给浏览器优雅退出连接关闭时取消订阅并移除连接记录由 ws_manager.py 统一管理活跃连接。还有一个体现细节的小设计任务启动后会有await asyncio.sleep(1)的短暂延迟——这是为了给前端 WebSocket 留出连接窗口避免开始处理这条首条消息因没人订阅而丢失尽管有文件兜底实时性上依然尽力保证。五、前端配套指数退避自动重连浏览器侧的 websocket.ts 实现了TaskWebSocket类它不只是简单建连还内置了一套生产级的重连策略断线后自动重连最多10 次采用指数退避间隔从 1 秒翻倍递增上限 30 秒避免服务端故障时疯狂重试打爆服务手动关闭close()时置位isManualClose标记跳过自动重连逻辑。配合后端的文件持久化 /messages历史接口即使长任务中途断网重连用户刷新页面也能恢复完整进度这正是任务队列体验的最后一公里。六、一键停止任务取消信号的传递建模任务跑偏了怎么办项目通过_active_tasks注册表 asyncio.Event实现协作式取消_, cancel_event _active_tasks[task_id] cancel_event.set() # 仅发信号工作流内部周期性检查并自行优雅退出POST /modeling/{task_id}/cancel接口只负责发信号工作流执行到下一个检查点时感知到事件后自行收尾从而避免粗暴杀进程导致的资源泄漏。七、部署细节一条 Docker Compose 拉起 Redis在 docker-compose.yml 中Redis 以redis:alpine镜像运行并挂载持久化卷后端通过depends_on保证 Redis 先行启动。连接地址由 setting.py 中的两个配置项控制REDIS_URL默认redis://redis:6379/0Docker 网络内服务发现REDIS_MAX_CONNECTIONS连接池上限默认 10适配单机中小并发场景。此外GET /status健康检查接口会实时 ping Redis 并返回后端与缓存服务的运行状态方便运维观测。总结这套任务队列方案值得借鉴的 4 个点设计点实现方式解决的问题异步任务入队FastAPI BackgroundTasks asyncio.Task长任务不阻塞 HTTP进度广播Redis Pub/Sub 按任务隔离频道多任务并发、互不干扰消息双写Pub/Sub 实时推送 JSON 文件持久化断线不丢进度、可回溯优雅取消asyncio.Event 协作式信号随时停止且资源安全回收整体来看MathModelAgent 用Redis 发布订阅 WebSocket 文件持久化三件套以极低的架构复杂度支撑起了多任务并发、实时进度、断线恢复与任务取消四大能力是学习异步任务队列设计的一个非常贴切、可读的开源范例。相关源码集中在 backend/app/services/ 与 backend/app/routers/ 目录配合 docs/tutorial.md 教程可以快速上手本地部署。【免费下载链接】MathModelAgent专为数学建模设计的 Agent skills ,自动完成数学建模生成一份完整的可以直接提交的论文。 An Agent Designed for Mathematical Modeling ,Automatically complete mathmodel and generate a complete paper ready for submission.项目地址: https://gitcode.com/GitHub_Trending/ma/MathModelAgent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考