芹菜任务，运行更多的任务

I am using celerybeat to kick off a primary task that kicks of a number of secondary tasks. I have both tasks written already.

我正在使用celerybeat来启动一个主要任务，该任务启动了一些次要任务。我已经写好了两个任务。

Is there a way to easily do this? Does Celery allow for tasks to be run from within tasks?

有什么简单的方法吗?芹菜允许任务在任务内部运行吗?

My example:

我的例子:

@task
def compute(users=None):
    if users is None:
        users = User.objects.all()

    tasks = []
    for user in users:
        tasks.append(compute_for_user.subtask((user.id,)))

    job = TaskSet(tasks)
    job.apply_async() # raises a IOError: Socket closed

@task
def compute_for_user(user_id):
    #do some stuff

compute gets called from celerybeat, but causes an IOError when it tries to run apply_async. Any ideas?

compute从celerybeat调用，但在尝试运行apply_async时导致IOError。什么好主意吗?

3 个解决方案

#1

To answer your opening questions: As of version 2.0, Celery provides an easy way to start tasks from other tasks. What you are calling "secondary tasks" are what it calls "subtasks". See the documentation for Sets of tasks, Subtasks and Callbacks, which @Paperino was kind enough to link to.

要回答您的开场白问题:在2.0版本中，芹菜提供了从其他任务开始任务的简单方法。你所说的“次要任务”就是它所说的“子任务”。请参阅任务集、子任务和回调的文档，@Paperino可以将其链接到该文档中。

For version 3.0, Celery changed to using groups for this, and other, types of behavior.

在3.0版本中，芹菜改为使用组来进行这种行为，以及其他类型的行为。

Your code shows that you are already familiar with this interface. Your actual question seems to be, "Why am I getting a 'Socket Closed' IOError when I try to run my set of subtasks?" I don't think anyone can answer that, because you have not provided enough information about your program. Your excerpt cannot be run as-is, so we cannot examine the problem you're having for ourselves. Please post the stacktrace provided with the IOError, and with any luck, someone that can help you with your crasher will come along.

您的代码表明您已经熟悉这个接口。您的实际问题似乎是:“为什么我在尝试运行一组子任务时，会得到一个‘套接字关闭’IOError ?”我认为没有人能回答这个问题，因为你没有提供关于你的程序的足够信息。您的摘录不能按原样运行，因此我们不能检查您为我们自己所拥有的问题。请把IOError提供的stacktrace贴出来，如果幸运的话，可以帮助你处理你的碎纸机的人会出现。

#2

You can use something like this (Support in 3.0 )

您可以使用类似的东西(在3.0中支持)

g = group(compute_for_user.s(user.id) for user in users)
g.apply_async()

#3

And since version 3.0 'TaskSet' isn't the term anymore... Groups, Chains and Chords as a special type of subtask is the new thing, see http://docs.celeryproject.org/en/3.1/whatsnew-3.0.html#group-chord-chain-are-now-subtasks

既然3.0版的TaskSet已经不是这个词了……组、链和和弦作为一种特殊类型的子任务是新事物，请参见http://docs.celeryproject.org/en/3.1/whatsnew3.0.html #group-chord- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

#1