用 Celery 组合链、组和块

我想使用 Celery 作为 URL 抓取器。

我有一个 Url 列表，我必须对每个 URL 执行 HTTP 请求并将结果写入文件中（整个列表使用同一文件）。

我的第一个想法是让 Celery 调用的任务中的这段代码击败所有n分钟 ：

@app.task
def get_urls(self):
    results = [get_url_content.si(
        url=url
    ) for url in urls]

    ch = chain(
        group(*results),
        write_result_on_disk.s()
    )

    return ch()

这段代码工作得很好，但有一个问题：我有一千个 URL 需要抓取，如果其中 1 个 get_url_content 失败，则不会调用 write_result_on_disk 并且我们会丢失所有先前抓取的内容。

我想做的是chunk通过分割 URL 来执行任务，获取结果并将其写入磁盘。例如20个url的内容被写入磁盘上。

请问你有什么想法吗？我尝试过chunks()功能，但没有得到真正有用的结果。

使用 CeleryBeat 执行类似 cron 的任务是一个好主意。

我会尝试捕获你的异常get_url_content微任务。当你抓住它们时，只需归还其他东西即可。这样，您可以在 summarise_task 中评估（例如计数、列出、检查）它们。

如何使用chunks和链条chunks与另一个任务：

步骤 1：将块转换为组：

如中所述http://docs.celeryproject.org/en/latest/userguide/canvas.html#chunks http://docs.celeryproject.org/en/latest/userguide/canvas.html#chunks, .group()转换类型的对象celery.canvas.chunks成组，这是 Celery 中更常见的类型。

第 2 步：链接组和任务

中的“通过组合让你大吃一惊”部分http://docs.celeryproject.org/en/latest/userguide/canvas.html#the-primitives http://docs.celeryproject.org/en/latest/userguide/canvas.html#the-primitives提到：

将一个组与另一个任务链接在一起将自动升级 它是一个和弦

这是包含这两个任务的一些代码以及我通常如何调用它们：

@app.task
def solve_micro_task(arg: str) -> dict:
    ...

@app.task
def summarize(items: List[List[dict]]):
    flat_list = [item for sublist in items for item in sublist]
    for report in flat_list:
        ...

chunk_tasks = solve_micro_task.chunks(<your iterator, e.g., a list>), 10)  # type: celery.canvas.chunks
summarize_task = summarize.s()
chain(chunk_tasks.group(), summarize_task)()