重写Celery Task类以实现任务间变量共享

1. 任务间复用对象的方法 #

Celery 每个任务实例会独立初始化，所以直接用 Task 实例属性无法跨任务共享对象。根据 worker 运行模式，有两种可行方向：

• 单进程 worker 场景：用 Task 类变量存储对象，同一进程内所有任务实例共享类变量副本：

  class Handler(Task):
      # 类变量，单进程内所有任务共享
      shared_base_obj = None
  
      @property
      def base_handler(self):
          if self.shared_base_obj is None:
              self.shared_base_obj = cls1()
          return self.shared_base_obj
  

• 多进程/分布式场景：用外部共享存储（如 Redis、Memcached），将对象序列化后存入，跨任务、跨进程访问；也可借助 multiprocessing.Manager 实现进程间内存共享，但复杂度较高。

2. 重写 Task 的 run() 方法及 Celery 5.X 注册 #

完全可以重写 run() 方法，这是自定义任务行为的常规操作，Celery 5.X 支持两种注册方式：

方式一：装饰器指定基类 #

class CustomTask(Task):
    def run(self, *args, **kwargs):
        # 自定义前置逻辑，比如初始化实例对象
        if not hasattr(self, 'base_obj'):
            self.base_obj = cls1()
        # 调用任务实际逻辑
        return self._execute_task(*args, **kwargs)

    def _execute_task(self, *args, **kwargs):
        # 子类/任务函数需实现此方法
        pass

# 注册任务时绑定基类
@app.task(base=CustomTask)
def test123():
    from celery import current_task
    current_task.base_obj.some_method()

@app.task(base=CustomTask)
def test456():
    from celery import current_task
    current_task.base_obj.another_method()

方式二：继承基类后注册 #

class Test123Task(CustomTask):
    name = 'test123'

    def _execute_task(self):
        self.base_obj.some_method()

class Test456Task(CustomTask):
    name = 'test456'

    def _execute_task(self):
        self.base_obj.another_method()

# 手动注册任务
app.tasks.register(Test123Task())
app.tasks.register(Test456Task())

3. 序列化之外的替代方案 #

• Worker 启动时初始化全局对象：借助 worker_init 信号，在每个 worker 进程启动时初始化一次对象，任务直接访问全局变量：

  from celery.signals import worker_init
  
  shared_obj = None
  
  @worker_init.connect
  def init_worker(**kwargs):
      global shared_obj
      shared_obj = cls1()
  
  @app.task
  def test123():
      shared_obj.some_method()
  
  @app.task
  def test456():
      shared_obj.another_method()
  

• 单例模式+锁：单进程场景用单例类管理对象；多进程场景配合文件锁/Redis 锁，避免并发初始化冲突。

修正后的代码示例 #

针对你原代码中「直接给函数对象赋值而非 Task 实例属性」的问题，修正后的单进程可用版本：

from celery import Task, Celery

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

class cls1:
    def method(self):
        print("cls1 method called")

class Handler(Task):
    shared_base_obj = None

    @property
    def base_handler(self):
        if Handler.shared_base_obj is None:
            Handler.shared_base_obj = cls1()
        return Handler.shared_base_obj

@app.task(base=Handler)
def test123():
    from celery import current_task
    current_task.base_handler.method()

@app.task(base=Handler)
def test456():
    from celery import current_task
    current_task.base_handler.method()

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