FastAPI玩转Redis数据库，异步任务如何不哭？

📝 摘要：还在为FastAPI异步任务里管理Redis连接和数据库会话而头疼？本文将从一个真实踩坑案例出发，带你深入浅出地理解如何优雅地在后台任务中操作外部服务。我们会手写一个生产级别的代码组织方案，帮你彻底告别连接泄露和资源耗尽的问题，让你的异步任务跑得又快又稳。 今天咱们聊的这个话题，可以说是每个用FastAPI做生产项目的同学，几乎都会遇到的一道坎——异步任务里怎么安全、高效地调用Redis和数据库？ 你可能会问：“这有啥难的？直接丢到 BackgroundTasks 里，然后正常调用不就行了？”哎，如果真这么简单，我这篇文章就不会诞生了。听我给你讲个故事。 🎯 1. 一个让人抓狂的“小问题” 之前有个项目，需要在上传用户头像后，异步生成几种不同尺寸的缩略图，并把处理结果和状态存到MySQL，同时把用户ID和任务ID塞到Redis里做状态追踪。一切看起来都很美好，代码也跑通了。 但上线后，噩梦开始了！应用跑了一两天，就开始随机报错，有的说MySQL连接已关闭，有的说Redis连接数超限。 最离谱的是，有时候任务执行到一半，数据库连接突然断开了，导致部分数据写入失败，状态成了“薛定谔的完成”。 后来debug了好几天，才发现问题的根源：我把数据库和Redis的会话（Session/Connection）直接传递到了异步任务函数里，但生命周期完全错乱了！ 🔍 2. 问题到底出在哪？ FastAPI的BackgroundTasks虽然用起来简单，但它本质上是在响应返回之后，在同一个进程中“偷偷”执行的一个函数。 问题是，你在请求生命周期内创建的数据库会话（比如通过依赖项注入的db: Session），在请求结束后，通常会被框架自动关闭。但你的后台任务还在用这个已经被关闭的会话，不出错才怪！ Redis连接也是类似，如果你把连接池里“借”出来的连接直接传进去，一旦主请求结束，连接被归还或关闭，后台任务再用的时候，就会直接GG。 这里我要特别强调一点：千万不要在异步任务里，直接复用请求生命周期内的资源对象！这是新手最容易踩的坑，也是我当初血泪教训的核心。 ⚙️ 3. 核心原理：各管各的，生命周期要分离 那正确的姿势是什么呢？核心思想就是“谁用谁创建，用完自己关”。 异步任务函数内部，不应该依赖外部传递进来的“活”连接，而是应该拥有自己独立的资源管理逻辑。 具体来说，我们需要在异步任务函数内部，重新创建所需的资源（比如新的数据库Session，新的Redis连接），并在任务执行完毕后，确保这些资源被正确关闭或归还。 这听起来像是个体力活，但其实我们可以通过一些好的代码组织模式，让它变得优雅且可维护。 🛠️ 4. 实战：生产级别的组织方案 好，理论说完了，咱们直接上代码。 📁 第一步：目录结构 project/ ├── app/ │ ├── api/ # 路由层 │ ├── core/ # 核心配置（数据库、Redis等） │ │ ├── database.py │ │ └── redis_client.py │ ├── models/ # 数据库模型 │ ├── schemas/ # Pydantic模型 │ └── tasks/ # 异步任务模块！✨ │ ├── __init__.py │ ├── user_tasks.py │ └── worker.py # 任务入口 └── ... 💾 第二步：核心资源管理（重点！） 在tasks/worker.py里，我们定义一个基类或辅助函数，专门负责在每个任务中，初始化和管理这些资源。