在 Celery 链中并行运行结果列表答案

【问题标题】：Run list of results in parallel in Celery chain在 Celery 链中并行运行结果列表
【发布时间】：2020-01-16 10:41:46
【问题描述】：

这是我要实现的逻辑：

1. Get a list of URLs by scraping a home page
2. Get, scrape and store a sublist of URLs in parallel by visiting each link in above list

我首先尝试创建一个主任务，它首先从主页抓取所有 URL，然后在 for 循环中获取子 URL：

@task
def master_task():
    urls = scrape_list_of_urls()
    job = group([scrape_url_and_save.s(url) for url in urls])  # scrape_url takes around 200ms each URL, and there are thousands of URLs. Hence I want it to run in parallel
    result = job.apply_async()
    result.join()
@task
def scrape_url_and_save(url):
    save_to_db(contents_of_url_by_scraping)
...
def scrape(request):  # In Django
    master_task.delay()
    ...

但这会导致错误：

RuntimeError：永远不要在任务中调用 result.get()！

我在 Django 应用程序中使用 Celery 4。 master_task 必须是一项任务，因为我不希望用户在抓取主页时等待。我不确定我的代码逻辑是否正确。更好的逻辑将不胜感激。

【问题讨论】：

标签： python celery python-3.7 django-celery

【解决方案1】：

有类似的 SO question 可以解决运行时错误的问题。 Link to the question
有什么原因，为什么这么复杂？我认为有这么多任务过于复杂。也许我错了，但是你列表中的第 3 点是多余的，不需要一次将数据存储到数据库中。

所以如果我是你，我会做一个master_task，它会为每个url 生成scrape_url.delay(url)，它会执行scraping 工作并将结果保存到数据库中。这将导致在队列中产生许多任务，然后并行处理它们将取决于您的工作人员数量。

【讨论】：

我检查了那个问题，但这并不是我真正想要使用的。这更像是一种黑客攻击而不是正确的代码。 2. 嗯，可以，我可以跳出第3步，直接在第2步存储数据，这个可以忽略。我希望有一个使用链或其他东西的解决方案，因为有数千个 URL，这可能会导致性能非常缓慢或死锁。
看起来您希望获得一些魔法。如果您的任务可能会卡住并导致死锁，那么您对等待任务结果的期望是什么？我认为跳过第 3 步，让第 2 步保存数据是合适的。
啊我不是这个意思，我是因为这个意思：docs.celeryproject.org/en/latest/userguide/… 既然有上千个URL，如果worker pool用完就会导致死锁。跨度>
当然，为什么不像现在这样链式运行任务并让它们保存数据？

【解决方案2】：

这是实现所需工作流程的方法 - 您的主任务应返回一组子任务，例如

@app.task(bind=True)
def master_task(self):
    urls = scrape_list_of_urls()
    job = group((self.app.signature('tasks_module.scrape_url_and_save', (url,)) for url in urls))  
    # run scrape_url_and_save in parallel e.g. using gevent/eventlet worker pool
    return job.delay()

【讨论】：