如何在Celery任务中可靠使用Tortoise ORM？

你遇到的问题本质是Celery的同步Worker模型与异步Tortoise ORM的事件循环、连接池冲突导致的，完全不需要放弃，按以下步骤可以解决：

核心思路 #

每个Celery任务单独初始化、销毁Tortoise ORM，避免Worker进程复用带来的事件循环残留、连接池共享冲突。

1. 封装ORM初始化/销毁逻辑 #

写独立的异步函数负责ORM的启动和关闭，确保任务前后清理资源：

from tortoise import Tortoise

async def init_tortoise():
    await Tortoise.init(
        db_url="postgres://user:pass@db:5432/dbname",
        modules={"models": ["your_app.models"]},
        # 根据并发量调整连接池参数，避免连接耗尽
        pool_size=10,
        max_overflow=20
    )
    # 可选：首次运行时生成表结构，生产环境建议提前迁移
    # await Tortoise.generate_schemas()

async def close_tortoise():
    await Tortoise.close_connections()

2. 封装Celery任务的异步执行器 #

为每个任务创建独立的事件循环，包含完整的ORM生命周期：

import asyncio
from your_module import init_tortoise, close_tortoise

def run_celery_async_task(coro, *args, **kwargs):
    # 为每个任务创建全新的事件循环，避免复用冲突
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        # 先初始化ORM
        loop.run_until_complete(init_tortoise())
        # 执行目标异步任务
        result = loop.run_until_complete(coro(*args, **kwargs))
        return result
    finally:
        # 必须清理：关闭ORM连接和事件循环
        loop.run_until_complete(close_tortoise())
        loop.close()

3. 编写Celery任务 #

将数据库逻辑写成异步函数，在Celery同步任务中调用封装好的执行器：

from celery import Celery

app = Celery('tasks', broker='redis://redis:6379/0')

# 异步数据库任务逻辑
async def async_update_user_status(user_id):
    from your_app.models import User
    user = await User.get(id=user_id)
    user.status = "processed"
    await user.save()
    return user.id

# Celery同步任务
@app.task
def process_user(user_id):
    return run_celery_async_task(async_update_user_status, user_id)

4. 关键注意事项 #

• 绝对不要在全局作用域初始化Tortoise，否则Worker进程复用时会引发连接池冲突。
• 调整Celery的--concurrency参数时，要确保Tortoise的pool_size * concurrency不超过PostgreSQL的max_connections（默认100），避免连接耗尽。
• 如果仍出现asyncpg连接错误，检查数据库服务是否稳定，或者增加pool_recycle参数（比如设置为300，自动回收闲置连接）。

备选方案：改用异步任务队列 #

如果上述方案仍有问题，可以考虑替换Celery为Arq——基于asyncio的异步任务队列，和Tortoise ORM天然兼容，无需处理同步转异步的麻烦，但需要调整项目的任务队列架构。

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