celery - 通过send_task在celery中链接两个远程任务

Question

在 celery 中，我想将远程 worker 的两个任务链接在一起。有人可以告诉我如何在send_task中指定它吗？还是有其他方法可以调用远程任务？

BR

Answer 1

我试图做同样的事情。我找不到纯粹按其名称生成任务的任何内置功能，但是添加这样的工具并不难:

from celery import Task as BaseTask

class Task(BaseTask):
    def __init__(self, name, *args, **kwargs):
        super(BaseTask, self).__init__(*args, **kwargs)
        self.name = name

通过此类，您可以执行以下操作:

(
    Task('worker.hello').s('world') | 
    Task('messaging.email-results').s(email_address='user@company.com')
)()

或者，或者:

app.send_task(
    'worker.hello', ['world'], 
    chain=[
        Task('messaging.email-results').s(email_address='user@company.com')
    ]
)

编辑:

忽略之上，我意识到实现此目的的正确方法是使用 Signature类(如下面的@Chrismit所述):

from celery import Signature

(
    Signature('worker.hello', args=['world']) | 
    Signature('messaging.email-results', kwargs={'email_address':'user@company.org'})
)()

或者，或者:

from celery import Signature

app.send_task(
    'worker.hello', ['world'], 
    chain=[
        Signature('messaging.email-results', kwargs={'email_address': 'user@company.com'})
    ]
)

重要说明:链中第一个任务之后的所有任务实际上都不会排定，直到工作人员处理之前的任务为止(这很有意义，因为在运行前一个任务之前，我们不知道后续任务的输入) 。 随后的任务在工作程序的代码库中安排。 出于这个原因，您需要确保满足以下条件之一:

您的每个工作人员都知道task_routes，因此可以将后续任务放置在适当的队列中(例如，在我的示例中，应该知道以messaging.*开头的任务应该放在'messaging'队列中)

创建链时，您已将正确的queue编码到每个Signature类中。 Celery已经有了从任务名称中派生队列名称的工具，可以依靠该名称:

def get_queue_name(task_name):
    return app.amqp.router.route({}, task_name)['queue'].name

(
    Signature('worker.hello', args=['world']) | 
    Signature(
        'messaging.email-results',
        kwargs={'email_address':'user@company.org'},
        queue=get_queue_name('messaging.email-results')  # If typing the task name twice annoys you, you could subclass Signature to do this automatically
    )
)()

(我认为这是最干净的解决方案，因为它允许工作人员彼此不了解)

所有任务都在默认队列中执行。如果您没有在工作程序上声明任何task_routes且任务签名中未指定queue，则Celery将在工作程序的default_queue中安排该任务。除非定制，否则为'celery'。我强烈建议您不要这样做，因为它不是很明确，也不允许进行过多的队列管理，但是尽管如此，它还是一个选择。