我正在忙着编写一个小型游戏服务器来试用Flask。游戏通过REST向用户展示API。用户执行操作和查询数据很容易,但是我想在app.run()循环之外为“游戏世界”提供服务,以更新游戏实体等。鉴于Flask的实现如此简洁,我希望看看是否有Flask方式可以做到这一点。
你的其他线程必须从WSGI服务器调用的同一应用程序启动。
下面的示例创建一个后台线程,该线程每5秒执行一次,并处理Flask路由函数也可用的数据结构。
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