Celery任务基类中是否有任务启动时的初始化函数？

你要的任务启动前执行初始化的需求，Celery其实有现成的钩子和实现方式，下面给你两种靠谱的做法：

方法一：使用官方before_start钩子 #

Celery的Task基类自带before_start方法，它会在任务正式执行之前被调用，正好对应你要的"on_start"场景。直接在你的BaseTask里重写这个方法就行：

class BaseTask(celery.Task):
    def before_start(self, task_id, args, kwargs):
        # 这里写你的初始化逻辑，比如连接资源、加载配置等
        print(f"任务 {task_id} 启动，执行初始化")
        super().before_start(task_id, args, kwargs)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # 原有的失败处理逻辑
        super().on_failure(exc, task_id, args, kwargs, einfo)

    def on_success(self, retval, task_id, args, kwargs):
        # 原有的成功处理逻辑
        super().on_success(retval, task_id, args, kwargs)

方法二：重写__call__方法 #

如果需要更灵活的控制（比如要在任务实例创建后、执行前做初始化），可以重写任务的__call__方法——因为Celery执行任务本质就是调用这个方法：

class BaseTask(celery.Task):
    def __call__(self, *args, **kwargs):
        # 先执行你的初始化逻辑
        print("任务启动前执行初始化")
        # 再调用父类的__call__执行任务本身
        return super().__call__(*args, **kwargs)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        super().on_failure(exc, task_id, args, kwargs, einfo)

    def on_success(self, retval, task_id, args, kwargs):
        super().on_success(retval, task_id, args, kwargs)

注意点 #

• before_start是Celery官方推荐的钩子，专门用于任务启动前的操作，优先级更高；
• 如果你的初始化逻辑需要依赖任务的self实例属性，两种方法都能满足，但__call__更贴近实例调用的流程；
• 不管用哪种方法，记得调用父类的对应方法，避免破坏Celery的原有流程。

内容的提问来源于stack exchange，提问作者Rock Dev Mufasa