因为Flask默认单线程同步模型，耗时操作会阻塞worker进程，导致请求排队、超时（504）或白屏；必须将任务剥离HTTP生命周期，交由Celery+Redis异步处理。
为什么直接在Flask路由里跑耗时任务会卡住请求
Flask默认是单线程同步模型，time.sleep(10)、大文件处理、调用外部API等操作会阻塞整个worker进程，导致后续请求排队甚至超时。用户看到的是白屏或504，不是“任务在后台跑”。必须把耗时逻辑剥离出HTTP请求生命周期。
Celery需要独立的Broker，Redis比RabbitMQ更轻量易配
Broker是Celery任务队列的中间人，没它任务发不出去也收不回来。本地开发用Redis最省事：docker run -p 6379:6379 -d redis:alpine 启一个容器就行。RabbitMQ配置项多、端口杂，新手容易卡在ConnectionRefusedError。确认Broker可用再启动Celery worker，否则celery -A tasks worker --loglevel=info会报Connection to broker failed。
- 检查
CELERY_BROKER_URL = "redis://localhost:6379/0"是否和Redis容器端口一致 - 不要用
localhost在Docker Compose里——改用服务名如redis://redis:6379/0 - Celery 5.0+默认禁用pickle，必须显式设
task_serializer = "json"，否则TypeError: Object of type xxx is not JSON serializable
Flask上下文不能自动穿透到Celery任务里
Celery worker是独立Python进程，current_app、db这些Flask全局对象在任务函数里直接用会报RuntimeError: Working outside of application context。正确做法是在任务函数内重新初始化依赖，而不是传入实例：
# tasks.py
from celery import Celery
from your_app import create_app, db # 假设你用工厂函数
<p>celery = Celery("tasks")
celery.conf.broker_url = "redis://localhost:6379/0"</p><p>@celery.task
def send_email_async(user_id):
app = create_app() # 新建应用上下文
with app.app_context():
user = db.session.get(User, user_id) # 现在db可用</p><h1>发邮件逻辑...</h1><p></p><div class="aritcle_card flexRow artxards">
<div class="artcardd flexRow">
<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill10420" title="Flask Wrap"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20d79f23518.jpg" alt="Flask Wrap" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a rel="nofollow" href="/xiazai/skill10420" title="Flask Wrap" class="overflowclass">Flask Wrap</a>
<p class="overflowclass">火焰围攻 授权: q15040409-creator.功能概述火焰围攻 授权: q15040409-creator.是一项面向实际任务的技能，主要用于Description.；Flask Web. - Python.；核心要点HTTP?；.Version.1.0.tags. wrap.github.automation.。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。使用与执行使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要</p>
</div>
<a rel="nofollow" href="/xiazai/skill10420" title="Flask Wrap" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
</a>
</div>
</div>
- 别试图把
db对象作为参数传进@celery.task函数——序列化失败 - 如果用SQLAlchemy，每个Celery任务必须自己
db.session.close()，否则连接泄漏 -
异步任务里不要依赖
request对象，HTTP上下文已销毁
前端怎么知道Celery任务进度
Celery本身不提供实时进度推送，得靠轮询+任务ID。关键三步：路由返回任务ID → Celery任务更新update_state → 前端用fetch查状态：
# views.py
@bp.route("/start-task")
def start_task():
task = long_running_task.delay()
return {"task_id": task.id}
<h1>tasks.py</h1><p>@celery.task(bind=True)
def long_running_task(self):
for i in range(100):
time.sleep(0.1)
self.update_state(state="PROGRESS", meta={"progress": i})
</p>
前端用/task-status?task_id=xxx查时，后端要捕获celery.exceptions.TimeoutError和celery.exceptions.TaskRevokedError，否则查一个不存在的任务会500。另外，result.ready()为True不代表成功——得看result.successful()，因为任务可能抛异常但状态仍是SUCCESS。
任务结果默认只存几秒，要查历史记录得配result_expires，比如celery.conf.result_expires = 3600（1小时）。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
