我怎样才能添加一个背景线程烧瓶？

我正忙着编写一个小游戏服务器来尝试烧瓶。游戏通过 REST 向用户公开一个 API。用户很容易执行动作和查询数据，但我想服务的 "game world"以外的 app.run()循环更新游戏实体等。考虑到 Flask 的实现非常干净，我想看看是否有一种 Flask 方法可以实现这一点。

140903 次浏览
小开
最佳答案

Your additional threads must be initiated from the same app that is called by the WSGI server.

The example below creates a background thread that executes every 5 seconds and manipulates data structures that are also available to Flask routed functions.

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:
pass
# 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 -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app
小开

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:

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

A simple example copied from 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()
小开

First, you should use any WebSocket or polling mechanics to notify the frontend part about changes that happened. I use Flask-SocketIO wrapper, and very happy with async messaging for my tiny apps.

Nest, you can do all logic which you need in a separate thread(s), and notify the frontend via SocketIO object (Flask holds continuous open connection with every frontend client).

As an example, I just implemented page reload on backend file modifications:

<!doctype html>
<script>
sio = io()


sio.on('reload',(info)=>{
console.log(['sio','reload',info])
document.location.reload()
})
</script>

class App(Web, Module):


def __init__(self, V):
## flask module instance
self.flask = flask
## wrapped application instance
self.app = flask.Flask(self.value)
self.app.config['SECRET_KEY'] = config.SECRET_KEY
## `flask-socketio`
self.sio = SocketIO(self.app)
self.watchfiles()


## inotify reload files after change via `sio(reload)``
def watchfiles(self):
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class Handler(FileSystemEventHandler):
def __init__(self,sio):
super().__init__()
self.sio = sio
def on_modified(self, event):
print([self.on_modified,self,event])
self.sio.emit('reload',[event.src_path,event.event_type,event.is_directory])
self.observer = Observer()
self.observer.schedule(Handler(self.sio),path='static',recursive=True)
self.observer.schedule(Handler(self.sio),path='templates',recursive=True)
self.observer.start()

提交答案