运行更多任务的 Celery 任务

Question

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

有没有办法轻松做到这一点？ Celery 是否允许从任务内部运行任务？

我的示例：

@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 从 celerybeat 调用，但在尝试运行 apply_async 时导致 IOError。有什么想法吗？

Answer 1

回答您的开场问题：从版本 2.0 开始，Celery 提供了一种从其他任务启动任务的简单方法。您所说的“次要任务”就是所谓的“子任务”。请参阅任务、子任务和回调集的文档，其中@ Paperino 很友善地提供了链接。

对于 3.0 版，Celery 更改为使用 组 来实现此行为和其他类型的行为。

您的代码表明您已经熟悉此界面。您的实际问题似乎是，“当我尝试运行一组子任务时，为什么会出现‘Socket Closed’IOError？”我认为没有人可以回答这个问题，因为您没有提供有关您的程序的足够信息。您的摘录无法按原样运行，因此我们无法亲自检查您遇到的问题。请发布随 IOError 提供的堆栈跟踪，如果幸运的话，将会有人帮助您解决崩溃问题。

Answer 2

你可以使用这样的东西（3.0中支持）

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

Answer 3

由于 3.0 版“任务集”不再是这个术语...组、链和和弦作为一种特殊类型的子任务是新事物，请参阅 http://docs.celeryproject.org/en/3.1/whatsnew-3.0.html#group-chord-chain-are-now-subtasks

Answer 4

对于提到的 IOError，虽然这里的信息不足以说明导致它的原因，但我的猜测是您尝试在任务函数内部建立连接，因此每当调用任务时，就会建立一个新连接。如果任务被调用一千次，就会有一千个连接。这将淹没系统套接字管理器，并且 IOError 是它的抱怨。