FastAPI @repeat_every 如何防止并行 defcheduled_task() 实例答案

【问题标题】：FastAPI @repeat_every how to prevent parallel def scheduled_task() instancesFastAPI @repeat_every 如何防止并行 defcheduled_task() 实例
【发布时间】：2020-07-08 09:25:38
【问题描述】：

我们正在使用fastapi-utils 在后台执行计划任务。如果数据库中有新数据可用，我们会检查所有 5 秒，如果是，我们会处理它（最多需要 5 分钟）

在此期间，协同程序应该被阻塞，以便它只触发一次。

我们注意到我们的数据有时会被处理 3 倍，我们假设调度程序继续运行，即使函数已被触发。

因此我们尝试使用IsRunningQuery 变量来规避它。

我们尝试了一个没有 @repeat_every 的带有 while True 循环的解决方案，以使其在启动时运行一次，但 Azure Webapps 不允许运行它。

@app.on_event("startup") 
@repeat_every(wait_first=True,seconds=int(10))
def scheduled_task() -> None:
    global IsRunningQuery
    global LastCheck
    if IsRunningQuery == False:
        IsRunningQuery = True
        gunicorn_logger.info("status='checkforleads'")
        OurProccessingClass.processDataBaseData() # can take up 5 minutes
        LastCheck=Utils.datetime()
        IsRunningQuery = False

此变体适用于我们的 DEV 环境，但不适用于 Azure

@app.on_event("startup") 
async def scheduled_task() -> None:
    while True:
        gunicorn_logger.info("status='checkforleads'")
        OurProccessingClass.processDataBaseData() # can take up 5 minutes
        time.sleep(int(os.environ["CRM_SLEEP"]))

【问题讨论】：

标签： fastapi

【解决方案1】：

要完成这项任务，您需要一些锁定系统，但要适合您的环境。

例如，当仅运行单个工作程序时，使用单个异步循环，来自异步同步原语的简单 Lock 将是理想的...

但是如果你想引入更多的worker，那么锁的状态将不会在实例之间同步。如果您的工作人员是在同一个系统上生成的，您可以使用文件系统锁（例如来自fnctl 模块的锁），但同样，如果您要引入更多服务器实例，它将不再起作用。

下一步可能是在数据库级别或任何其他能够管理锁或仅将某些任务交付给一个接收者的外部系统上引入锁，但这很快就会变得非常复杂。

这就是为什么，像 celery 这样的系统可以让你安排任务，如果可能的话，系统会注意防止这个任务被多次执行（注意这并不总是可能的，例如执行者可能完成任务但由于某些致命错误或任何其他中断（如断电）而从不更新任务的状态。这就是为什么这些系统可以确保该任务至少运行一次或最多运行一次一次，但永远不能保证两者，只会尽力最大化另一次的机会）。

【讨论】：

我也会为此推荐 Redis。 redis.Lock 可以很容易地跨线程、进程、服务器等使用，只要可以访问 redis url，就可以了！

【解决方案2】：

听起来您应该考虑一个单独的工作进程来执行这个昂贵的数据库调用。这可以是一个简单的 Python 脚本，它会定期调用您的 dB 函数。

您提到您正在使用 Azure Web 堆栈，这是我在 docs/ 中找到的内容 https://docs.microsoft.com/en-us/azure-stack/operator/azure-stack-app-service-add-worker-roles?view=azs-2108&tabs=az

祝你好运，让我们知道你在哪里登陆:)

【讨论】：