如何为基于django-tenant-schemas的多租户Django应用配置django-celery-beat？

我之前也碰到过一模一样的坑！Celery能正常跑但Beat死活起不来，核心原因就是Beat默认只会盯着公共schema（public），但你的周期性任务表都存在各个租户自己的schema里，自然找不到表报错。下面是我亲测有效的解决步骤：

1. 实现租户感知的Beat调度器 #

首先得让Beat能识别并切换租户schema，我们可以自定义一个调度器来遍历所有租户。在项目里新建custom_scheduler.py文件：

from celery.beat import PersistentScheduler
from tenant_schemas.utils import get_tenant_model
from django.db import connection

class TenantAwareScheduler(PersistentScheduler):
    def apply_async(self, entry, producer=None, advance=True, **kwargs):
        # 遍历系统中所有租户
        Tenant = get_tenant_model()
        for tenant in Tenant.objects.all():
            # 切换到当前租户的schema
            connection.set_tenant(tenant)
            # 执行当前租户的周期性任务
            super().apply_async(entry, producer=producer, advance=advance, **kwargs)
        # 任务执行完切回公共schema
        connection.set_schema_to_public()

2. 修改Celery配置指定调度器 #

在项目的celery.py配置文件里，指定使用我们自定义的调度器：

from celery import Celery

app = Celery('your_project_name')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# 配置Beat使用租户感知的调度器
app.conf.beat_scheduler = 'your_project_name.custom_scheduler.TenantAwareScheduler'

3. 确保任务能关联到对应租户 #

你的周期性任务在定义时，要确保能切换到目标租户的环境执行。比如可以把租户schema名称作为参数传入任务，再用租户上下文包裹逻辑：

from celery import shared_task
from tenant_schemas.utils import tenant_context
from your_app.models import Tenant

@shared_task
def tenant_specific_task(tenant_schema_name):
    tenant = Tenant.objects.get(schema_name=tenant_schema_name)
    with tenant_context(tenant):
        # 在这里执行租户专属的任务逻辑
        print(f"执行租户 {tenant.name} 的周期性任务")

4. 启动Beat时指定调度器 #

最后启动Beat的时候，明确指定我们的自定义调度器，命令如下：

celery -A your_project_name beat -l info --scheduler your_project_name.custom_scheduler.TenantAwareScheduler

额外注意事项 #

如果你用django-celery-beat存储任务，记得在创建新租户时自动在租户schema里迁移任务表。可以通过信号来实现：

from django.db.models.signals import post_save
from django.dispatch import receiver
from tenant_schemas.utils import get_tenant_model
from django.core.management import call_command
from django.db import connection

@receiver(post_save, sender=get_tenant_model())
def migrate_tenant_task_tables(sender, instance, created, **kwargs):
    if created:
        # 切换到新租户schema并迁移任务表
        connection.set_tenant(instance)
        call_command('migrate', '--schema=' + instance.schema_name, 'django_celery_beat')
        # 切回公共schema
        connection.set_schema_to_public()

我当时这么改完就成功启动Beat了，你可以照着试试！

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