Celery 任务 prefork 多进程启动时，如何通过自定义类的__call__函数设置属性

# pool 并发方式
# worker_param = ['celery', '-A', 'app.celery', 'worker', '--pool=solo', '-l', 'INFO']
worker_param = ['celery', '-A', 'app.celery', 'worker', '--pool=prefork', '-l', 'INFO']
subprocess.Popen(worker_param)

# task 类
class AloneTask(celery.Task):
    def __init__(self, *args, **kwargs):
        super(AloneTask, self).__init__()

    def __call__(self, *args, **kwargs):
        print(f">>>self1={self}")
        self._sid = kwargs.get('sid', '1')

    def my_func(self):
        #  [ pool=solo ] ⇒self1==self2
        #  [ pool=prefork ] ⇒self1 ！=self2
        print(f">>>self2={self}")
        return self._sid

# 调用 call_my_task.delay("100")
@celery.task(base=AloneTask)
def call_my_task(sid):
    #  [ pool=solo ] ⇒OK
    #  [ pool=prefork ] ⇒AttributeError: 'call_my_task' object has no attribute '_sid'
    print(call_my_task.my_func())

第 1 条附言 · 2022-02-17 08:45:20 +08:00

解决方案：通过一个全局变量来重置任务函数

g_task = None

def __call__(self, *args, **kwargs):
        print(f">>>self1={self}")
        self._sid = kwargs.get('sid', '1')
        global g_task
        g_task = self

def call_my_task(sid)
    call_my_task = g_task
    print(call_my_task.my_func())

2 条回复 • 2022-05-01 09:21:04 +08:00

sunhk25

2022-02-16 18:53:45 +08:00

以上代码是在 Windows 下执行的，Celery 从版本 4 之后虽然可以使用，但是不太维护。
在 Ubuntu 上测试了一下，上面的代码没有问题。

opengo

2022-05-01 09:21:04 +08:00

@celery.task(bind=True, base=AloneTask)
def call_my_task(self, sid):
print(self._sid)