在FastAPI中,@repeat_every
装饰器用于定义定时任务,可以通过scheduled_task
实例来指定任务的执行时间间隔。在多线程环境中,如果有多个线程同时定义了同一个scheduled_task
实例,会导致任务被重复执行的问题。
为了避免并行定义scheduled_task
实例,可以采取以下几种方法:
scheduled_task
实例时,使用全局锁来确保只有一个线程可以进行定义,可以使用Python中的threading.Lock()
实现。具体实现如下:import threading
task_lock = threading.Lock()
@repeat_every(seconds=60)
def scheduled_task():
# 任务逻辑
with task_lock:
scheduled_task()
scheduled_task
实例被创建。具体实现如下:class ScheduledTask:
_instance = None
def __new__(cls):
if not cls._instance:
cls._instance = super().__new__(cls)
return cls._instance
@repeat_every(seconds=60)
def scheduled_task(self):
# 任务逻辑
scheduled_task_instance = ScheduledTask()
scheduled_task_instance.scheduled_task()
以上方法可以确保在多线程环境中,只有一个线程能够定义并执行scheduled_task
实例,避免了并行定义导致的重复执行问题。
FastAPI相关产品和介绍链接地址:
请注意,以上产品和链接仅供参考,具体选择应根据实际需求和项目情况进行评估和决策。
领取专属 10元无门槛券
手把手带您无忧上云