Python 如何将后台线程添加到烧瓶？
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/14384739/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
How can I add a background thread to flask?
提问by Marinus
I'm busy writing a small game server to try out flask. The game exposes an API via REST to users. It's easy for users to perform actions and query data, however I'd like to service the "game world" outside the app.run() loop to update game entities, etc. Given that Flask is so cleanly implemented, I'd like to see if there's a Flask way to do this.
我正忙于编写一个小型游戏服务器来试用烧瓶。游戏通过 REST 向用户公开 API。用户执行操作和查询数据很容易，但是我想在 app.run() 循环之外为“游戏世界”提供服务以更新游戏实体等。鉴于 Flask 实现得如此干净，我想看看是否有 Flask 方法可以做到这一点。
采纳答案by caio
Your additional threads must be initiated from the same app that is called by the WSGI server.
您的其他线程必须从 WSGI 服务器调用的同一个应用程序启动。
The example below creates a background thread that executes every 5 seconds and manipulates data structures that are also available to Flask routed functions.
下面的示例创建了一个每 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()
Call it from Gunicorn with something like this:
从 Gunicorn 调用它，如下所示：
gunicorn -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app
回答by Andreas Bergstr?m
In addition to using pure threads or the Celery queue (note that flask-celery is no longer required), you could also have a look at flask-apscheduler:
除了使用纯线程或 Celery 队列（注意不再需要flask-celery），您还可以查看flask-apscheduler：
https://github.com/viniciuschiele/flask-apscheduler
https://github.com/viniciuschiele/flask-apscheduler
A simple example copied from https://github.com/viniciuschiele/flask-apscheduler/blob/master/examples/jobs.py:
从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()
