フラスコにバックグラウンドスレッドを追加するにはどうすればよいですか？

Question

私はフラスコを試すための小さなゲームサーバーを書くのに忙しいです。ゲームは、RESTを介してユーザーにAPIを公開します。ユーザーがアクションやクエリデータを簡単に実行することは簡単ですが、app.run（）ループの外側で "ゲームの世界"にサービスを提供して、ゲームエンティティなどを更新したいと考えています.Flaskがきれいに実装されていることを考えれば、これを行うフラスコの方法があるかどうかを調べる。

Answer 1

追加のスレッドは、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 

Answer 2

a hackish way to do itのようですが、これは技術的にはサポートされていないと思います。

this answerも見つかりました。これについてはフラスコセロリーを使用しています。

Answer 3

フラスコ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()