在任务执行期间从 Celery worker 中捕获失败的任务

Question

我目前在 celery（与 RabbitMQ）中使用Task.delay() 来执行异步任务。任务分配给 8 个工作人员，在这些任务中，我正在执行数据库插入操作。

有时一个任务会失败，并且数据库插入不会发生。这可能是由于超时或 JSON 解码错误造成的。我想捕获执行失败。

以下是相关代码：

views.py

def celery_url_filtering(request):
    for each_data in dataset:
        #each_data is a Json object
        res = result.delay(each_data)
    while(res.status == 'PENDING'):
         pass
    return JsonResponse({'ok':'Success'})

tasks.py

@app.task
def result(dataeach_data):
    # Parse each_data  and do data insertion here
    return "Something"

如何在列表中捕获失败的执行？

Answer 1

据我所知，您的担忧是您想查看“PENDING”或“FAILURE”任务并重试/应用一些应用程序逻辑。

如果是这样，您可以根据需要让 cron 以固定的时间表、每天/每小时等方式运行。此 cron 作业可以根据您的计划捕获在最后一天/小时内失败的任务。

您可以使用django-celery-beat 设置 cron 作业，使用django-celery-results 使用 Django ORM 存储 celery 任务结果。

例如，你可以有一个像这样的 celery 任务

from celery import shared_task
from django_celery_results.models import TaskResult

**tasks.py**
@shared_task(name="failed_task_cron", bind=True)
def failed_task_cron(self, **kwargs):
    """
    Celery task to run on daily schedule to do something with the failed tasks
    """
    tasks = TaskResult.objects.filter(status='FAILURE')
    # tasks is a queryset of all the failed tasks
    # Perform application logic

您可以在 celery 设置中像这样为上述任务设置 cron

from celery.schedules import crontab

# ...

CELERY_BEAT_SCHEDULE = {
    "failed_task_cron": {
        "task": "path/to/tasks.failed_task_cron",
        "schedule": crontab(hour=1)           # Runs every hour
    }
}