Django Celery无法从指定数据库Schema获取数据求助

重启supervisorctl管理的Celery Worker数小时后，会间歇性出现无法从指定Schema的可用表中获取数据的问题，所有配置的三个RabbitMQ队列均存在该问题。

相关代码：

with schema_context("public"):
    try:
        client = Client.objects.get(subdomain=hostname)
    except ObjectDoesNotExist:
        client = Client.objects.using("paid").get(subdomain=hostname)

依赖版本：

• celery==4.4.7
• django-celery==3.3.0
• django-tenant-users==0.3.12

1. 重置数据库连接避免上下文残留 #

Celery Worker长期运行会复用数据库连接，schema上下文可能在连接中残留导致后续请求出错。在任务执行前强制关闭所有连接：

from django.db import connections

def your_task(hostname):
    # 重置所有数据库连接
    for conn in connections.all():
        conn.close()
    with schema_context("public"):
        try:
            client = Client.objects.get(subdomain=hostname)
        except ObjectDoesNotExist:
            client = Client.objects.using("paid").get(subdomain=hostname)

2. 确保任务前后schema状态正确 #

给Celery任务添加明确的schema上下文管理，任务结束后强制切回默认schema：

from celery import task
from django_tenants.utils import schema_context, get_public_schema_name

@task(bind=True)
def your_task(self, hostname):
    client = None
    with schema_context(get_public_schema_name()):
        try:
            client = Client.objects.get(subdomain=hostname)
        except ObjectDoesNotExist:
            with schema_context("paid"):
                client = Client.objects.get(subdomain=hostname)
    # 强制切回public schema，避免影响后续任务
    with schema_context(get_public_schema_name()):
        pass

3. 配置Worker自动重启 #

通过supervisor设置Celery Worker的--max-tasks-per-child参数，让Worker执行一定数量任务后自动重启，彻底避免连接复用带来的问题：

[program:celery-worker]
command=celery -A your_project worker -Q queue1,queue2,queue3 --max-tasks-per-child 1000 --loglevel=info
user=your_user
directory=/path/to/your/project
stdout_logfile=/var/log/celery/worker.log
stderr_logfile=/var/log/celery/worker.err.log
autostart=true
autorestart=true

4. 验证schema上下文管理器的正确性 #

手动测试schema_context是否在退出时正确重置schema，避免因上下文管理器失效导致的问题：

from django_tenants.utils import get_current_schema_name, schema_context, get_public_schema_name

# 测试上下文切换
with schema_context("public"):
    assert get_current_schema_name() == "public"
# 验证退出后是否回到默认schema
assert get_current_schema_name() == get_public_schema_name()

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