解决多进程中APScheduler重复运行的问题

左手的ㄟ右手 2023-10-08 08:37 8阅读 0赞

#### 下面的答案不可行，可行方案往下拉！ ####

问题  
在一个python web应用中需要定时执行一些任务，所以用了APScheduler这个库。又因为是用flask这个web框架，所以用了flask-apscheduler这个插件（本质上与直接用APScheduler一样，这里不作区分）。

在开发中直接测试运行是没有问题的，但是用gunicorn部署以后发生了重复运行的问题：

每个任务在时间到的时刻会同时执行好几遍。

注意了一下重复的数量，恰恰是gunicorn里配置的worker进程数量，显然是每个worker进程都启动了一份scheduler造成。

解决  
可以想到的方案有几个：

用--preload启动gunicorn，确保scheduler只在loader的时候创建一次  
另外创建一个单独的定时任务项目，单独以一个进程运行  
用全局锁确保scheduler只运行一次  
经过实践，只有第三个方案比较好。

preload的问题：

虽然这样可以使用scheduler创建代码只执行一次，但是问题也在于它只执行一次，重新部署以后如果用kill -HUP重启gunicorn，它并不会重启，甚至整个项目都不会更新。这是preload的副作用，除非重写部署脚本，完全重启应用。

单独进程的问题：

也是因为部署麻烦，需要多一套部署方案，虽然用Docker会比较方便，但仍然不喜欢，而且同时维护两个项目也多出很多不必要的事情。

全局锁是一个较好的方案，但问题在于找一个合适的锁。

python自带的多进程多线程锁方案都需要一个共享变量来维护，但是因为worker进程是被gunicorn的主进程启动的，并不方便自己维护，所以需要一个系统级的锁。

在Stackoverflow上看到有人是用了一个socket端口来做锁实现这个方案，但是我也不喜欢这样浪费一个宝贵的端口资源。不过这倒给了我一个启发：

可以用文件锁！

于是有了这个解决方案：

import atexit
    import fcntl
    from flask_apscheduler import APScheduler
    
    def init(app):
        f = open("scheduler.lock", "wb")
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
            scheduler = APScheduler()
            scheduler.init_app(app)
            scheduler.start()
        except:
            pass
        def unlock():
            fcntl.flock(f, fcntl.LOCK_UN)
            f.close()
        atexit.register(unlock)

原理  
init函数为flask项目初始化所调用，这里为scheduler模块的初始化部分。

首先打开（或创建）一个scheduler.lock文件，并加上非阻塞互斥锁。成功后创建scheduler并启动。

如果加文件锁失败，说明scheduler已经创建，就略过创建scheduler的部分。

最后注册一个退出事件，如果这个flask项目退出，则解锁并关闭scheduler.lock文件的锁。

（**好像不可行的样子。。。。反正我是没试成功**）

### **分布式场景下的问题解决：** ###

[https://www.kawabangga.com/posts/2903][https_www.kawabangga.com_posts_2903]

#### 亲测可用方案 ： ####

最后还是用一个socket端口来做锁实现这个方案。

相关讨论：

[https://stackoverflow.com/questions/16053364/make-sure-only-one-worker-launches-the-apscheduler-event-in-a-pyramid-web-app-ru][https_stackoverflow.com_questions_16053364_make-sure-only-one-worker-launches-the-apscheduler-event-in-a-pyramid-web-app-ru]

[https://stackoverflow.com/questions/24352634/would-starting-apscheduler-in-a-uwsgi-app-end-up-with-one-scheduler-for-each-wor?noredirect=1&lq=1][https_stackoverflow.com_questions_24352634_would-starting-apscheduler-in-a-uwsgi-app-end-up-with-one-scheduler-for-each-wor_noredirect_1_lq_1]

[https://github.com/viniciuschiele/flask-apscheduler/issues/79][https_github.com_viniciuschiele_flask-apscheduler_issues_79]

[https://github.com/agronholm/apscheduler/issues/160][https_github.com_agronholm_apscheduler_issues_160]

总的来说，这个属于 Flask 的一个特性，Flask 多进程运行，都是 Fork 的主进程，定义的定时任务肯定也会被复制成多份。

亲测下面的方法可行：

import sys, socket
    
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 47200))
    except socket.error:
        print "!!!scheduler already started, DO NOTHING"
    else:
        from apscheduler.schedulers.background import BackgroundScheduler
        scheduler = BackgroundScheduler()
        scheduler.start()
        print "scheduler started"

让多个进程在开始自身的定时任务时，直接绑定一个端口，利用一个端口只能绑定一个进程的属性来避免多次运行。

其实在我的场景下，我既需要多进程都做的定时任务，也需要运行一次的定时任务。

多进程都做的定时任务其实很好改，在代码里自行实现了定时执行的逻辑。

所以 Flask 程序多进程模式下，定时任务只运行一次，还是使用这个方法更好一些~~

[https_www.kawabangga.com_posts_2903]: https://www.kawabangga.com/posts/2903
[https_stackoverflow.com_questions_16053364_make-sure-only-one-worker-launches-the-apscheduler-event-in-a-pyramid-web-app-ru]: https://stackoverflow.com/questions/16053364/make-sure-only-one-worker-launches-the-apscheduler-event-in-a-pyramid-web-app-ru
[https_stackoverflow.com_questions_24352634_would-starting-apscheduler-in-a-uwsgi-app-end-up-with-one-scheduler-for-each-wor_noredirect_1_lq_1]: https://stackoverflow.com/questions/24352634/would-starting-apscheduler-in-a-uwsgi-app-end-up-with-one-scheduler-for-each-wor?noredirect=1&lq=1
[https_github.com_viniciuschiele_flask-apscheduler_issues_79]: https://github.com/viniciuschiele/flask-apscheduler/issues/79
[https_github.com_agronholm_apscheduler_issues_160]: https://github.com/agronholm/apscheduler/issues/160