Python Celery + FastAPI + Vue 全栈异步任务实战

本文将手把手带你搭建FastAPI后端 API Celery异步任务队列 Redis消息中间件 / 结果存储 Vue前端的全栈异步项目实现异步任务提交、任务状态查询、前端实时查看进度的完整功能适合处理耗时操作文件导出、数据处理、邮件发送、批量数据计算、AI 模型推理等大中型业务场景。在现代 Web 应用开发中同步处理耗时任务会直接导致接口超时、前端页面卡死、服务器并发能力急剧下降而基于 Celery 的异步架构能完美解决这些问题。本教程从环境搭建、代码编写、服务启动到前后端联调全流程讲解代码可直接复制运行零基础也能快速掌握全栈异步开发核心技能。一、项目架构总览本项目采用前后端分离架构整体分为四层结构每层职责清晰、低耦合高内聚符合企业级项目开发规范。1.1 技术栈分工前端Vue3 Axios Element Plus负责任务提交界面渲染、异步请求发送、任务状态轮询、进度条实时展示为用户提供流畅的交互体验后端FastAPI高性能 Python Web 框架基于 Starlette 和 Pydantic 构建提供 RESTful API 接口处理前端请求、响应数据、跨域配置支持异步请求处理异步任务核心Celery分布式任务队列专门处理耗时、非实时的后台任务支持任务进度更新、异常捕获、结果返回中间件存储Redis高性能内存数据库同时承担两个核心角色Broker消息代理存储 Celery 的待执行任务队列Backend结果存储保存 Celery 任务的执行状态、进度数据和最终返回结果1.2 完整项目执行流程前端 Vue 页面用户输入任务信息点击提交按钮通过 Axios 发送请求到 FastAPI 后端接口FastAPI 接收请求后不执行耗时操作直接将任务投递到 Celery 任务队列立即返回唯一任务 ID 给前端前端获取任务 ID 后启动定时器轮询 FastAPI 的任务状态查询接口Celery Worker 监听 Redis 任务队列自动拉取任务并在后台异步执行任务执行过程中Celery 实时更新进度信息到 Redis前端通过轮询获取最新任务状态和进度实时渲染进度条和状态文字任务执行完成 / 失败后Celery 将最终结果存入 Redis前端获取结果后停止轮询并展示整个流程实现了请求与执行分离后端接口毫秒级响应前端无阻塞系统并发能力大幅提升。二、环境准备2.1 安装依赖软件本项目依赖三大基础环境必须提前安装配置完成Python 3.9 及以上版本推荐使用 Python3.9/3.10兼容性最佳官网下载https://www.python.org/downloads/安装时勾选Add Python to PATH配置环境变量。Redis 6.0核心中间件Redis 是本项目的核心依赖Celery 必须依赖 Broker 才能运行本项目统一使用 Redis 作为 Broker 和 Backend。Windows 系统推荐使用第三方维护版本下载地址https://github.com/tporadowski/redis/releases下载后解压直接运行redis-server.exe即可启动Linux 系统sudo apt update sudo apt install redis-serverMac 系统brew install redis启动后默认端口6379无密码本地开发环境。Node.js 16 及以上版本用于运行 Vue3 前端项目官网下载https://nodejs.org/安装完成后打开命令行执行node -v和npm -v验证是否安装成功。2.2 创建标准项目目录结构遵循企业级项目规范创建清晰的目录结构方便后续维护和扩展plaintextasync_project/ ├── backend/ # 后端服务目录FastAPI Celery │ ├── main.py # FastAPI主入口文件接口定义 │ ├── celery_app.py # Celery核心配置文件 │ ├── tasks.py # 异步任务定义文件 │ └── requirements.txt # Python依赖清单 ├── frontend/ # 前端目录Vue3项目 │ └── ...(vue自动生成的项目文件) └── start.sh # 可选Linux/Mac一键启动所有服务脚本2.3 后端 Python 依赖安装进入backend目录创建requirements.txt文件写入项目所需的所有 Python 依赖txtfastapi0.104.1 uvicorn0.24.0 celery5.3.6 redis5.0.1 python-multipart0.0.6 cors-middleware0.0.1 eventlet0.33.3安装命令打开命令行执行以下指令一键安装所有依赖bash运行cd backend pip install -r requirements.txtfastapiWeb 框架核心uvicornASGI 服务器用于运行 FastAPIcelery异步任务队列redisRedis 客户端连接 Redis 服务python-multipart处理文件上传等表单数据cors-middleware解决前后端跨域问题eventletWindows 系统运行 Celery 的必备协程库三、后端核心代码实现后端是整个项目的核心分为Celery 配置、异步任务定义、FastAPI 接口开发三部分代码逻辑清晰模块化设计。3.1 Celery 配置celery_app.py该文件用于初始化 Celery 实例配置 Redis 连接、时区、序列化方式等核心参数是 Celery 的入口文件。python运行from celery import Celery # 1. 配置Redis连接地址本地默认配置无密码 # 格式redis://:密码主机地址:端口/数据库号 REDIS_URL redis://localhost:6379/0 # 2. 创建Celery应用实例 celery Celery( async_tasks, # 任务模块名称自定义即可 brokerREDIS_URL, # 指定BrokerRedis存储任务队列 backendREDIS_URL, # 指定BackendRedis存储任务结果 # 时区配置解决时间异常问题 timezoneAsia/Shanghai, enable_utcFalse, # 序列化配置统一使用json格式保证前后端数据兼容 task_serializerjson, result_serializerjson, accept_content[json], # 任务过期时间任务结果保存1小时自动清理 result_expires3600, )核心配置说明broker必须配置Celery 的任务队列存储位置backend可选用于存储任务执行结果和进度本项目必须配置timezone统一时区避免任务时间异常result_expires自动清理过期任务结果节省 Redis 内存3.2 异步任务定义tasks.py该文件用于编写所有耗时异步任务支持进度实时更新、异常捕获处理是 Celery 的核心功能实现。python运行from celery_app import celery import time from celery.exceptions import Ignore # 定义耗时异步任务bindTrue表示绑定任务实例可使用self更新状态 celery.task(bindTrue, nametasks.long_time_task) def long_time_task(self, task_name: str): 耗时异步任务模拟企业级耗时操作 selfCelery任务上下文对象用于更新任务进度和状态 task_name前端传入的任务名称用于个性化展示 # 定义总进度步数模拟10秒的耗时操作 total_steps 10 try: # 循环模拟任务执行过程 for i in range(1, total_steps 1): # 模拟耗时文件处理、数据计算、接口调用等 time.sleep(1) # 核心代码更新任务进度前端可实时获取 self.update_state( statePROGRESS, # 状态执行中 meta{ current: i, # 当前进度 total: total_steps, # 总进度 status: f正在处理{task_name}进度{i*10}% } ) # 任务执行完成返回标准化结果 return { code: 200, state: 执行完成, status: SUCCESS, result: f任务【{task_name}】执行完成, progress: 100 } # 异常处理任务执行失败时捕获错误信息 except Exception as e: self.update_state( stateFAILURE, meta{ exc_type: str(type(e).__name__), # 异常类型 exc_message: str(e) # 异常信息 } ) # 忽略默认异常处理返回自定义错误信息 raise Ignore()任务功能说明支持实时进度更新每执行一步更新一次进度信息标准化异常处理错误信息清晰返回给前端支持自定义任务参数适配不同业务场景可扩展添加任务重试、超时控制、日志记录等功能3.3 FastAPI 接口开发main.py该文件是后端 API 入口提供任务提交、任务状态查询、健康检查三大接口并解决前后端跨域问题。python运行from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from celery_app import celery from tasks import long_time_task from celery.result import AsyncResult # 1. 创建FastAPI应用实例 app FastAPI( titleFastAPICelery异步任务项目, description全栈异步任务实战接口文档, version1.0.0 ) # 2. 核心配置跨域中间件解决Vue前端访问后端的跨域问题 app.add_middleware( CORSMiddleware, allow_origins[*], # 开发环境允许所有域名生产环境替换为前端地址 allow_credentialsTrue, allow_methods[*], # 允许所有请求方式 allow_headers[*], # 允许所有请求头 ) # 3. 接口1提交异步任务前端调用 app.post(/api/submit_task, summary提交耗时异步任务) def submit_task(task_name: str): 接收前端请求投递任务到Celery队列 :param task_name: 任务名称前端传入参数 :return: 标准化响应 唯一任务ID # delay()核心方法将任务异步投递到队列非阻塞执行 task long_time_task.delay(task_name) # 返回任务ID前端通过该ID查询状态 return { code: 200, msg: 任务已成功提交至异步队列后台处理中, task_id: task.id } # 4. 接口2根据任务ID查询任务状态和进度 app.get(/api/task_status/{task_id}, summary查询异步任务状态) def get_task_status(task_id: str): 前端轮询调用获取任务实时状态 :param task_id: 提交任务时返回的唯一ID :return: 任务状态、进度、结果/错误信息 # 根据任务ID获取任务执行结果 task_result AsyncResult(task_id, appcelery) # 分状态处理任务结果 if task_result.state PENDING: # 状态1任务排队中未开始执行 response { state: 等待执行, status: 任务排队中..., progress: 0 } elif task_result.state PROGRESS: # 状态2任务执行中返回实时进度 response { state: 执行中, progress: int(task_result.info[current] / task_result.info[total] * 100), status: task_result.info[status] } elif task_result.state SUCCESS: # 状态3任务执行完成返回结果 response task_result.result elif task_result.state FAILURE: # 状态4任务执行失败返回错误信息 response { state: 执行失败, error: task_result.info[exc_message], progress: 0 } else: # 其他未知状态 response { state: task_result.state, status: 未知状态请检查任务是否存在, progress: 0 } return response # 5. 接口3健康检查测试服务是否正常运行 app.get(/, summary服务健康检查) def health_check(): return { code: 200, message: FastAPICelery异步服务运行正常, docs: 接口文档地址http://localhost:8000/docs }接口说明所有接口返回标准化 JSON 数据前端易于解析自动生成接口文档访问http://localhost:8000/docs可在线测试接口完整的状态判断覆盖任务所有生命周期跨域配置保证 Vue 前端可以正常访问四、启动后端服务后端需要同时启动 3 个服务启动顺序不能颠倒缺一不可Redis → FastAPI → Celery Worker。4.1 启动 Redis 服务Redis 是基础服务必须第一个启动bash运行# Windows系统进入Redis安装目录 redis-server.exe # Linux/Mac系统 redis-server启动成功后命令行会显示Ready to accept connections表示 Redis 正常运行。4.2 启动 FastAPI 服务FastAPI 提供 API 接口使用 uvicorn 运行bash运行cd backend uvicorn main:app --host 0.0.0.0 --port 8000 --reload参数说明--host 0.0.0.0允许所有 IP 访问--port 8000服务端口号--reload开发模式代码修改后自动重启启动成功后访问服务地址http://localhost:8000接口文档http://localhost:8000/docs4.3 启动 Celery Worker 服务Celery Worker 是任务执行者监听 Redis 队列必须单独启动bash运行cd backend # Windows系统必须加-P eventlet否则无法运行 celery -A celery_app worker --loglevelinfo -P eventlet # Linux/Mac系统 celery -A celery_app worker --loglevelinfo启动成功标志命令行出现ready字样表示 Celery 已就绪等待接收任务。五、Vue 前端实现前端采用 Vue3 Vite 构建使用 Element Plus 组件库快速搭建界面Axios 发送请求实现任务提交和进度实时展示。5.1 创建 Vue3 项目bash运行# 进入项目根目录创建Vue项目 npm create vitelatest frontend -- --template vue cd frontend # 安装核心依赖 npm install axios element-plusaxios发送 HTTP 请求与后端交互element-plusVue3 UI 组件库提供进度条、按钮、输入框等组件5.2 前端核心代码src/App.vue替换App.vue文件全部内容实现完整功能vuetemplate div classasync-task-container h2FastAPICelery 全栈异步任务演示/h2 !-- 任务提交表单区域 -- div classtask-form el-input v-modeltaskName placeholder请输入任务名称 stylewidth: 300px; margin-right: 10px / el-button typeprimary clicksubmitTask :loadingsubmitting sizedefault 提交异步任务 /el-button /div !-- 任务进度展示区域有任务ID时显示 -- div v-iftaskId classtask-progress-card h3任务状态{{ taskInfo.state }}/h3 !-- 进度条组件 -- el-progress :percentagetaskInfo.progress statusprimary stylemargin: 20px 0 / !-- 状态文字展示 -- div classstatus-text {{ taskInfo.status || taskInfo.result || taskInfo.error }} /div /div /div /template script setup import { ref, onMounted, onUnmounted } from vue import axios from axios import { ElMessage } from element-plus import ElementPlus from element-plus import element-plus/dist/index.css // 配置Axios请求基础地址 const request axios.create({ baseURL: http://localhost:8000/api, timeout: 10000 // 请求超时时间 }) // 响应式数据定义 const taskName ref(测试异步任务) const submitting ref(false) // 提交加载状态 const taskId ref() // 任务ID const taskInfo ref({ state: , progress: 0, status: }) // 任务信息 let timer null // 轮询定时器 /** * 1. 提交异步任务到后端 */ const submitTask async () { // 表单验证 if (!taskName.value.trim()) { ElMessage.warning(请输入任务名称) return } submitting.value true try { // 发送POST请求提交任务 const res await request.post(/submit_task, null, { params: { task_name: taskName.value } }) // 保存任务ID taskId.value res.data.task_id ElMessage.success(任务提交成功正在后台处理...) // 启动轮询查询任务状态 startQueryTaskStatus() } catch (err) { ElMessage.error(任务提交失败请检查后端服务是否启动) console.error(提交任务错误, err) } finally { submitting.value false } } /** * 2. 查询任务状态轮询执行 */ const queryTaskStatus async () { if (!taskId.value) return try { const res await request.get(/task_status/${taskId.value}) taskInfo.value res.data // 任务完成或失败时停止轮询 if (taskInfo.value.state 执行完成 || taskInfo.value.state 执行失败) { clearInterval(timer) timer null if (taskInfo.value.state 执行完成) { ElMessage.success(taskInfo.value.result) } else { ElMessage.error(任务执行失败) } } } catch (err) { ElMessage.error(查询任务状态失败) clearInterval(timer) timer null } } /** * 3. 启动轮询每1秒查询一次状态 */ const startQueryTaskStatus () { // 清除已有定时器避免重复轮询 if (timer) clearInterval(timer) timer setInterval(queryTaskStatus, 1000) } // 页面销毁时清除定时器防止内存泄漏 onUnmounted(() { if (timer) clearInterval(timer) }) /script style scoped .async-task-container { max-width: 800px; margin: 80px auto; padding: 0 20px; text-align: center; font-family: Microsoft YaHei, sans-serif; } .task-form { margin: 40px 0; display: flex; justify-content: center; align-items: center; } .task-progress-card { margin-top: 40px; padding: 30px; border: 1px solid #e4e7ed; border-radius: 12px; background: #fafafa; box-shadow: 0 2px 12px rgba(0, 0, 0, 0.08); } .status-text { font-size: 16px; color: #303133; min-height: 30px; } /style前端功能说明简洁美观的界面支持任务名称输入、提交按钮加载状态提示提升用户体验实时进度条展示直观显示任务执行进度自动轮询状态任务完成 / 失败后自动停止轮询完善的错误提示和表单验证5.3 启动 Vue 前端服务bash运行cd frontend npm run dev启动成功后默认访问地址http://localhost:5173六、项目运行完整演示按照以下步骤操作即可体验全栈异步任务流程启动顺序启动 Redis → 启动 FastAPI → 启动 Celery Worker → 启动 Vue 前端打开前端页面http://localhost:5173输入任务名称如数据导出任务点击提交异步任务按钮前端立即提示提交成功页面实时显示任务状态和进度条从 0% 逐步增长到 100%10 秒后任务执行完成进度条显示 100%弹出成功提示若任务执行异常页面会显示错误信息轮询自动停止整个过程中前端页面无任何卡顿后端接口无阻塞完美实现异步处理效果。七、核心知识点总结7.1 Celery 三大核心角色Broker任务队列存储本项目使用 Redis负责接收、存储待执行任务Worker任务执行者独立进程持续监听 Broker拉取任务执行Backend结果存储本项目使用 Redis保存任务进度、状态、结果7.2 核心技术点任务进度更新通过self.update_state()实现任务执行过程中的实时进度同步异步解耦FastAPI 只负责接收请求和返回结果耗时任务交给 Celery 执行前后端交互前端轮询查询状态替代长连接降低服务器压力跨域处理FastAPI 配置 CORSMiddleware解决前后端分离跨域问题任务唯一标识每个任务分配唯一 ID精准查询任务状态7.3 异步架构优势接口响应极快耗时任务后台执行接口毫秒级响应前端无阻塞用户提交任务后可继续操作页面无需等待高并发支持支持多 Worker 并行处理任务提升系统处理能力异常可控任务异常不会影响主接口单独捕获处理适用场景广文件导出、数据计算、邮件发送、AI 推理、批量处理等八、生产环境优化建议本项目为开发环境版本部署到生产环境需要进行以下优化提升系统稳定性、安全性和性能8.1 安全优化Redis 设置密码修改 Redis 配置文件添加密码Celery 连接时携带密码限制跨域域名FastAPI 跨域配置不使用*指定前端正式域名接口鉴权添加 Token/JWT 认证仅授权用户可提交任务8.2 性能优化Celery 多 Worker启动多个 Worker 进程celery -A celery_app worker -c 88 个并发FastAPI 托管使用 GunicornUvicorn 托管 FastAPI支持多进程前端优化使用 WebSocket 替代轮询实现服务端主动推送减少无效请求Redis 持久化开启 Redis 持久化防止服务重启丢失任务数据8.3 运维优化日志记录添加任务执行日志方便问题排查任务监控集成 Flower 监控 Celery 任务实时查看任务状态自动重启使用 Supervisor 托管 Celery 和 FastAPI服务崩溃自动重启超时控制为异步任务设置最大执行时间避免死循环任务占用资源九、常见问题排查9.1 Celery 启动失败Windows 系统必须加-P eventlet参数否则无法启动检查 Redis 是否启动连接地址是否正确检查 Python 版本是否兼容9.2 任务状态一直为 PENDINGCelery Worker 未启动任务名称不匹配检查celery.task的 name 参数Redis 连接失败无法获取任务9.3 前端无法访问后端接口检查 FastAPI 是否启动检查跨域配置是否正确检查请求地址是否正确9.4 任务进度不更新检查任务中self.update_state()代码是否正确检查 Redis 是否正常存储进度数据总结本项目完整实现了FastAPICeleryVueRedis全栈异步任务开发流程从环境搭建、代码编写、服务启动到前后端联调全覆盖是企业级异步开发的标准实践方案。开箱即用所有代码可直接复制运行无需额外修改适合新手快速上手功能完整实现任务提交、实时进度、状态查询、异常处理全流程架构标准前后端分离模块化设计易于扩展和维护场景通用可直接改造用于文件导出、数据处理、邮件发送等实际业务核心价值彻底解决耗时任务阻塞前端、接口超时、并发能力低等痛点项目启动顺序必须牢记Redis → FastAPI → Celery Worker → Vue 前端四个服务缺一不可。掌握本套技术栈将大幅提升你的全栈开发能力轻松应对大中型 Web 项目的异步需求。