守护程序化的 Celery 工作线程在手动启动工作线程时在 1 个任务上引发错误，而工作线程不会启动

Question

我正在尝试解决我在应用程序中处理的一个不方便的问题。我设置了一个 Celery 守护进程，与手动启动工作线程时的平稳过程相比，它在链中的 1 个任务上的行为似乎有所不同。

此任务尝试从 URL 列表中下载图片并保存它们，但守护程序在某个时候会抛出一些"TimeLimitExceeded"。我绝对可以手动运行工作线程(例如在screen内(，但是我失去了守护程序及其日志的灵活性......

为了解决这个问题，我用(bind=True(设置了任务，并实现了一次try/except 如果发生此特定错误，据说会重试任务(参见Celery文档(。

@celery.task(bind=True)
def fetch_img(self, datasetId):
    list_imgs = retrieve_imgs(datasetId)    # list of pair url + new filepath
    total = len(list_imgs)
    for p in range(len(list_imgs)):
        url = list_imgs[p][0]
        filepath = list_imgs[p][1]
        filename = os.path.basename(filepath)
        try:
            fetch_img = fetchUrl(url, filepath)
            if fetch_img[0] is True:            # download picture
                # message
                mesPic_part1 = 'n' + "# Url '" + url + "' successfully fetched"
                mesPic_part2 = 'n' + "--> File saved as '.../" + datasetId + '/' + filename + "'"
                list_parts = [mesPic_part1, mesPic_part2]
                downloaded += 1
            else:
                # get error message if download failed
                list_parts = [fetch_img[1]]
            # Message(s)
            for m in list_parts:
                log_message_line(m)
        except TimeLimitExceeded as exc:
            raise self.retry(countdown=60, exc=exc)
    return datasetId

但它并没有改善...当守护程序出现问题时，日志会给出：

[2019-07-23 18:44:24,691: ERROR/MainProcess] Task handler raised error: TimeLimitExceeded(300.0,)
Traceback (most recent call last):
  File "/opt/some/path/app/venv/lib/python3.6/site-packages/billiard/pool.py", line 658, in on_hard_timeout
    raise TimeLimitExceeded(job._timeout)
billiard.exceptions.TimeLimitExceeded: TimeLimitExceeded(300.0,)
[2019-07-23 18:44:24,694: ERROR/MainProcess] Hard time limit (300.0s) exceeded for application.core.celery.etl.task_etl_fetchImg.fetch_img[5cdce7d5-6ab2-425b-a1dd-5d847e3d403e]

你有过类似的事情吗？如果您碰巧有一些提示，我会很高兴...提前谢谢你！

Answer 1

您会收到 TimeLimitExceeded 异常，因为您的任务执行时间超过 300 秒。 300 秒是任务可以运行的默认时间。

您可以通过执行此操作来增加任务的time_limit(以秒为单位(。

@celery.task(time_limit=20)
def mytask():
    do_something()

您还可以使用以下方法为配置中的任务设置时间限制：

CELERY_ANNOTATIONS = {'module.mytask': {'time_limit': 20}}

Answer 2

更好的方法

from celery import shared_task
@shared_task(soft_time_limit=1200,time_limit=1300)

这给了足够的时间来运行你的任务，软限制在time_limit之前清理任务