如何向 Flask 添加后台线程？
- 2025-03-05 09:16:00
- admin 原创
- 102
问题描述：
我正忙着编写一个小型游戏服务器来试用 Flask。游戏通过 REST 向用户公开 API。用户可以轻松执行操作和查询数据，但我想在循环"game world"外部提供服务app.run()以更新游戏实体等。鉴于 Flask 的实现如此简洁，我想看看是否有 Flask 方法来做到这一点。
解决方案 1：
您的附加线程必须从 WSGI 服务器调用的同一应用程序启动。
下面的示例创建了一个后台计时器线程，该线程每 5 秒执行一次，并操作 Flask 路由函数可用的数据结构。
import threading
import atexit
from flask import Flask
POOL_TIME = 5 #Seconds
# variables that are accessible from anywhere
common_data_struct = {}
# lock to control access to variable
data_lock = threading.Lock()
# timer handler
your_timer = threading.Timer(0,lambda x: None,())
def create_app():
app = Flask(__name__)
def interrupt():
global your_timer
your_timer.cancel()
def do_stuff():
global common_data_struct
global your_timer
with data_lock:
pass
# Do your stuff with common_data_struct Here
# Set the next timeout to happen
your_timer = threading.Timer(POOL_TIME, do_stuff, ())
your_timer.start()
def do_stuff_start():
# Do initialisation stuff here
global your_timer
# Create your timer
your_timer = threading.Timer(POOL_TIME, do_stuff, ())
your_timer.start()
# Initiate
do_stuff_start()
# When you kill Flask (SIGTERM), cancels the timer
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
信号终止在 Windows 以外的操作系统上效果最佳。尽管这会在每次超时后创建一个新计时器，但其他计时器最终应该会被垃圾收集。
解决方案 2：
除了使用纯线程或 Celery 队列（请注意，不再需要 flask-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()
解决方案 3：
首先，您应该使用任何 WebSocket 或轮询机制来通知前端部分发生的更改。我使用包装器，并且对我的小应用程序的异步消息传递Flask-SocketIO非常满意。
Nest，您可以在单独的线程中执行所需的所有逻辑，并通过SocketIO对象通知前端（Flask 与每个前端客户端保持持续开放的连接）。
举个例子，我刚刚实现了后端文件修改时的页面重新加载：
<!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()
相关推荐
热门文章
项目管理软件有哪些？
热门标签
曾咪二维码
扫码咨询，免费领取项目管理大礼包！
云禅道AD
