5

在 Celery 中，我正在运行一个主任务，它为从查询中获得的每个项目运行一个子任务。子任务应该并行运行。在 UI 上，我有一个进度条，显示总共完成了多少子任务。我正在更新主要任务状态以将信息提供给进度条。我的问题是主要任务在将所有子任务推送到代理后立即结束，因此我无法再更新他的状态。我希望主要任务可以等到所有子任务完成。是否可以？还有其他解决方案吗？这是我的伪代码（真实代码不使用全局 ;-)）。

total = 0
done = 0

@task(ignore_result=True)
def copy_media(path):
    global total, done
    copy_media.update_state(state=STARTED, meta={'total': total, 'done': done})
    documents = Document.objects.all()
    total = documents.count()
    copy_media.update_state(state=STARTED, meta={'total': total, 'done': done})
    for document in documents:
        process_doc.delay(document, path, copy_media)

@task(ignore_result=True)
def process_doc(document, path, copy_media):
    global total, done
    # Do some stuff
    done += 1
    copy_media.update_state(state=STARTED, meta={'total': total, 'done': done})
4

3 回答 3

3

我找到了一种使用TaskSet. 但我并不完全满意，因为我不能忽略子任务的结果。如果我忽略process_doc任务的结果results.ready()总是返回False，results.completed_count()总是返回 0，等等。这是代码：

@task(ignore_result=True)
def copy_media(path):
    copy_media.update_state(state=STARTED, meta={'total': total, 'done': done})
    documents = Document.objects.all()
    total = documents.count()
    copy_media.update_state(state=STARTED, meta={'total': total, 'done': done})
    job = TaskSet(tasks=[process_doc.subtask((document, path))
                         for document in documents])
    results = job.apply_async()
    doc_name = ''
    while not results.ready():
        done = results.completed_count()
        if done:
            last = done - 1
            for idx in xrange(last, -1, -1):
                if results[idx].ready():
                    doc_name = results[idx].result
                    break
        copy_media.update_state(state=STARTED, meta={'total': total, 'done': done, 'doc-name': doc_name})
        time.sleep(0.25)

@task()
def process_doc(document, path):
    # Do some stuff
    return document
于 2012-04-06T21:30:37.510 回答
0

您可以使用 memcached 支持的缓存来存储完整任务的数量。甚至cache.inrc在 django 缓存 API 中用于原子增量，以确保计数的并发更新不会搞砸。

此外，保持主任务运行直到所有子任务完成是个坏主意，因为你基本上会长时间阻止一个芹菜工人。如果 celery 与一个工作进程一起运行，这将导致永无止境的锁定。

于 2012-04-06T21:37:15.647 回答
-1

我不知道您正在运行哪个版本的 celery，但您可以查看Group子任务（3.0 中的新功能）。

于 2012-10-10T13:45:53.033 回答