我每15分钟运行一次celerybeat调度程序，需要从API获取数据（速率限制= 300请求/min max）并将结果存储到数据库中。我想同时并行获取URL，但要遵守速率限制。如果任何工作进程在此失败，我不想重试，因为我将在15分钟内再次ping。关于如何在celery中实现这一点，有什么建议吗？
@celery.task(bind=True)
def fetch_store(self):
start = time()
return c.chain(c.group(emap.s() for _ in range(2000)), ereduce.s(start)).apply_async()
@celery.task(rate_limit='300/m')
def fetch():
#... requests data from external API
return data
@celery.task
def store(numbers, start):
end = time()
logger.info("Received" + numbers + " " + (end - start)/1000 + "seconds")
1条答案
按热度按时间vc9ivgsu1#
我通常定义一个自定义的
Task子类，并将max_retries设置为0（不是None，这会使它永远重试）：你也可以像这样在一行中做装饰器：
有关详细信息，请参阅文档。