如何添加一个后台线程烧瓶?

我正忙于写一个小游戏服务器来试用烧瓶。 游戏通过REST向用户显示一个API。 用户很容易执行操作和查询数据,但是我想在app.run()循环之外服务“游戏世界”以更新游戏实体等等。鉴于Flask是如此干净地实现的,我想看看是否有一个瓶子的方式来做到这一点。

您的其他线程必须从WSGI服务器调用的相同应用程序启动。

下面的示例创build一个后台线程,每5秒执行一次,并处理也可用于Flask路由function的数据结构。

import threading import atexit from flask import Flask POOL_TIME = 5 #Seconds # variables that are accessible from anywhere commonDataStruct = {} # lock to control access to variable dataLock = threading.Lock() # thread handler yourThread = threading.Thread() def create_app(): app = Flask(__name__) def interrupt(): global yourThread yourThread.cancel() def doStuff(): global commonDataStruct global yourThread with dataLock: # Do your stuff with commonDataStruct Here # Set the next thread to happen yourThread = threading.Timer(POOL_TIME, doStuff, ()) yourThread.start() def doStuffStart(): # Do initialisation stuff here global yourThread # Create your thread yourThread = threading.Timer(POOL_TIME, doStuff, ()) yourThread.start() # Initiate doStuffStart() # When you kill Flask (SIGTERM), clear the trigger for the next thread atexit.register(interrupt) return app app = create_app() 

从Gunicorn用这样的东西来称呼它:

 gunicorn -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app 

看起来有这样一个黑客办法 ,但我不认为这是技术支持。

我也find了这个答案 , 这个答案是用这个瓶子芹菜。

除了使用纯线程或Celery队列(请注意,不再需要烧瓶芹菜),您还可以看看flask-apscheduler:

https://github.com/viniciuschiele/flask-apscheduler

https://github.com/viniciuschiele/flask-apscheduler/blob/master/examples/jobs.py复制一个简单的例子:;

 from flask import Flask from flask_apscheduler import APScheduler class Config(object): JOBS = [ { 'id': 'job1', 'func': 'jobs:job1', 'args': (1, 2), 'trigger': 'interval', 'seconds': 10 } ] SCHEDULER_API_ENABLED = True def job1(a, b): print(str(a) + ' ' + str(b)) if __name__ == '__main__': app = Flask(__name__) app.config.from_object(Config()) scheduler = APScheduler() # it is also possible to enable the API directly # scheduler.api_enabled = True scheduler.init_app(app) scheduler.start() app.run()