95

我正忙着写一个小型游戏服务器来尝试烧瓶。游戏通过 REST 向用户公开一个 API。用户执行操作和查询数据很容易，但是我想为循环"game world"外部提供服务app.run()以更新游戏实体等。鉴于 Flask 实现得如此干净，我想看看是否有 Flask 方法可以做这个。

4

3 回答 3

97

您的附加线程必须从 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:
            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()

从 Gunicorn 调用它，如下所示：

gunicorn -b 0.0.0.0:5000 --log-config log.conf --pid=app.pid myfile:app
于 2014-04-06T21:27:07.707 回答
10

除了使用纯线程或 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()
于 2017-07-19T04:18:49.060 回答
0

首先，您应该使用任何 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()
于 2021-02-08T11:38:33.353 回答