如何覆盖芹菜任务的后端答案

【问题标题】：How do I override the backend for celery tasks如何覆盖芹菜任务的后端
【发布时间】：2015-07-18 02:14:24
【问题描述】：

我们使用 Redis 作为我们的结果后端。然而，对于一项任务，我们想覆盖它以使用 RabbitMQ。

用于此任务的结果存储后端。默认为 CELERY_RESULT_BACKEND 设置

所以我假设我们可以将Task.backend 设置为与CELERY_RESULT_BACKEND 接受的格式相同的字符串。

所以我试试这个：

celeryconfig.py

CELERY_RESULT_BACKEND = "redis://redis-host:7777"

tasks.py

@app.task(backend='amqp://guest@localhost/tasks-stg')
def my_task(params):
    ...

但是工人失败了：

[2015-05-07 13:33:49,264: ERROR/Worker-1] Process Worker-1
Traceback (most recent call last):
  File "/project/python2.7_x64/lib/python2.7/site-packages/billiard/process.py", line 292, in _bootstrap
    self.run()
  File "/project/python2.7_x64/lib/python2.7/site-packages/billiard/pool.py", line 286, in run
    self.after_fork()
  File "/project/python2.7_x64/lib/python2.7/site-packages/billiard/pool.py", line 389, in after_fork
    self.initializer(*self.initargs)
  File "/project/python2.7_x64/lib/python2.7/site-packages/celery/concurrency/prefork.py", line 81, in process_initializer
    app=app)
  File "/project/python2.7_x64/lib/python2.7/site-packages/celery/app/trace.py", line 178, in build_tracer
    store_result = backend.store_result
AttributeError: 'str' object has no attribute 'store_result'

【问题讨论】：

标签： python celery celery-task

【解决方案1】：

文档不正确。 Task.backend 实际上是来自celery.backends 的后端类的一个实例。在这种情况下，要覆盖任务类，我必须这样做：

from celery.backends.amqp import AMQPBackend

@app.task(backend=AMQPBackend(app, url='amqp://guest@localhost/tasks-stg'))
def my_task(params):
    ...

但是工作人员继续使用默认类，并且似乎没有提供覆盖它的方法。

【讨论】：

这确实对我有用，尽管回答的结束语。

【解决方案2】：

扩展和更新原来接受的答案，遇到类似问题的人可能会对List of Options available in Celery task decorator 感兴趣。请参阅Task.backend 部分：

用于此任务的结果存储后端。 celery.backends 中一个后端类的实例。默认为app.backend

另外，AMQP result backend has been removed 从 5.0 版本开始，您可以改用 celery.backends.rpc.RPCBackend。

例如：

from celery.backends.rpc import RPCBackend as CeleryRpcBackend
_rpc_backend = CeleryRpcBackend(app=your_celery_app, \
        exchange=RPC_REPLY_EXCHANGE_DEFAULT_NAME, \
        exchange_type=RPC_EXCHANGE_DEFAULT_TYPE )

@your_celery_app.task(backend=_rpc_backend, ... OTHER_ARGS ... )
def your_task_function(**kwargs):
    DO_SOMETHING ...

【讨论】：