Celery基于类的任务已在Worker注册，调用时提示NotRegistered

环境与Celery配置 #

Celery版本：celery==5.3.6
配置代码：

from celery import Celery
from settings.config import settings

celery_app = Celery(
    broker=settings.RABBITMQ_URL,
    backend="rpc://",
)

celery_app.config_from_object(settings.CELERY_SETTINGS_MODULE)
celery_app.autodiscover_tasks(["app.tasks"], force=True)

自定义AsyncTask实现 #

在app.tasks模块中定义继承自celery.Task的异步任务类：

from celery import Task
from abc import abstractmethod

class AsyncTask(Task):
    def run(self, *args, **kwargs):
        coro = self._run(*args, **kwargs)
        try:
            result = self._thread_isolated_worker(coro)
            return result
        except (DeprecatedTimeoutError, TimeoutError):
            raise TimeoutError
    
    @abstractmethod
    async def task(self, *args, **kwargs):
        # 任务逻辑实现
        pass

async_task = celery_app.register_task(AsyncTask())

调用脚本与报错信息 #

执行以下调用脚本时：

from app.tasks import async_task
async_task.delay().get()

抛出错误：

celery.exceptions.NotRegistered: 'async_task'

注：使用@shared_task装饰器注册的任务可正常运行，Worker启动时能看到该自定义任务已注册。

问题原因与解决方案 #

问题根源 #

Celery注册任务时，默认会使用类名作为任务标识（这里是AsyncTask），但调用时使用的是变量名async_task，两者不匹配，导致Worker无法找到对应的任务。

解决方法 #

方法1：注册实例时显式指定任务名称

创建AsyncTask实例时，通过name参数指定与变量名一致的任务标识：

async_task = celery_app.register_task(AsyncTask(name="async_task"))

方法2：在Task类中定义name属性

直接在自定义Task类中声明name属性，固定任务标识：

class AsyncTask(Task):
    name = "async_task"  # 显式指定任务名称
    
    def run(self, *args, **kwargs):
        # 原有代码
        pass
    
    @abstractmethod
    async def task(self, *args, **kwargs):
        pass

async_task = celery_app.register_task(AsyncTask())

额外验证点

• 确认Worker启动时，autodiscover_tasks确实扫描到了包含async_task注册代码的模块
• 调用时导入的async_task必须是已注册到celery_app的实例，而非未注册的类对象

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