解决 python 定时任务重复触发

最近在flask项目中写了个定时任务，用的gunicorn服务器， worker_class设置为1， 这样的话， master进程就会fork出1个子进程，导致定时任务被触发2次，为了解决这个问题，写了个装饰器，具体代码如下：

def lock_file(func):
    """保证同一时刻 只有一个任务执行"""
    def inner(*args, **kwargs):
        f = open("single_scheduler.lock", "wb")
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
            res = func(*args, **kwargs)
            time.sleep(0.1) 
        except:
            return
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)
            f.close()

        return res
    return inner

  # 代码中 sleep 0.1 秒就是为了避免func执行很快，把文件锁释放掉后，被第2个进程拿到锁后重复执行，也是这个装饰器能限制任务只被执行1次的核心代码

然后将这个装饰器放到定时任务上

@lock_file
def task():
	print("定时任务启动")

启动定时任务

schedule.add_job(job_id="job",func=task, seconds=10, max_instances=1)

其实这个重复执行的问题也可以用redis的分布式锁实现，只是这个方案不需要引入redis， 比较方便 。