如何调用基于类方法的Celery任务？

你遇到的问题本质是Celery的@app.task装饰器直接装饰实例方法时，没办法正确处理Python实例方法的self参数绑定逻辑。当你调用prs.foo.delay(data)时，Celery把你传入的data当成了方法的第一个参数self，导致实际需要的data参数缺失，所以报错「Missing argument data」。

下面给你几个可行的解决方案，你可以根据自己的需求选择：

方案一：把任务逻辑抽成独立的函数（最推荐） #

这种方式最简单，也避免了实例序列化的问题：

from celery import Celery

app = Celery('tasks', broker='your_broker_url')

class Parser:
    def __init__(self):
        self.OM = Omni()  # 初始化认证对象
    def foo(self, data):
        self.OM.parse(data)

# 把任务定义成独立的Celery任务函数
@app.task
def foo_task(data):
    # 每次任务执行时创建Parser实例
    prs = Parser()
    prs.foo(data)

调用的时候直接用：

foo_task.delay(data)

如果你的Omni认证是可以每次任务都重新初始化的，这个方案最稳妥。如果需要复用认证状态，可以考虑在Celery worker启动时初始化Parser实例（比如用app.on_configure或者worker_init信号）。

方案二：让Parser继承Celery的Task类 #

如果你需要保留实例的状态（比如复用OM的认证），可以让你的Parser类继承Celery的Task基类：

from celery import Celery, Task

app = Celery('tasks', broker='your_broker_url')

class Parser(Task):
    # 设置任务名称，可选
    name = 'parser.foo'
    
    def __init__(self):
        # 初始化只会在worker启动时执行一次，复用OM实例
        self.OM = Omni()
    
    def run(self, data):
        # run方法是Celery任务的执行入口
        self.OM.parse(data)

# 注册任务到Celery app
foo_task = Parser()
app.tasks.register(foo_task)

调用方式：

foo_task.delay(data)

这个方案的好处是OM对象只会在worker启动时初始化一次，不用每次任务都重新认证，适合需要持久化连接或状态的场景。

方案三：不推荐的实例序列化方式 #

如果你一定要直接调用实例方法，需要把实例本身传给Celery任务，但这要求你的Parser和Omni对象是可序列化的（比如用pickle），而且存在安全风险（不建议在生产环境使用）：

@app.task
def foo_task(parser, data):
    parser.foo(data)

# 调用时传入实例和数据
prs = Parser()
foo_task.delay(prs, data)

同时需要在Celery配置中允许pickle序列化：

app.conf.update(
    task_serializer='pickle',
    accept_content=['pickle'],
    result_serializer='pickle',
)

再次提醒，这种方式因为pickle的安全问题，不推荐在生产环境使用。

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