最近将Flask添加到一个随机打印单词的示例无限循环中。然而，当添加app.run(host='0.0.0.0')时，在我停止Flask运行后，该行后面的代码将不会执行。
if __name__ == '__main__':
app.run(host='0.0.0.0')
while True: # won't run until I press stop once (stop Flask) when running directly from IDE
...
我想要的是能够在Flask应用程序运行时运行while循环。
有什么办法解决这个问题吗？
您可以使用多线程来执行您想要的操作：
from flask import Flask
import threading
import time
app = Flask(__name__)
@app.route("/")
def hello_world():
return "Hello, World!"
def run_app():
app.run(debug=False, threaded=True)
def while_function():
i = 0
while i < 20:
time.sleep(1)
print(i)
i += 1
if __name__ == "__main__":
first_thread = threading.Thread(target=run_app)
second_thread = threading.Thread(target=while_function)
first_thread.start()
second_thread.start()
输出：
* Serving Flask app "app"
* Environment: production
* Debug mode: off
* Running on [...] (Press CTRL+C to quit)
0
1
2
3
4
5
6
7
8
[...]
想法很简单：
- 创建两个函数，一个用于运行应用程序，另一个用于执行wile循环
- 然后在一个单独的线程中执行每个函数，使它们并行运行
您也可以使用多处理而不是多线程来实现这一点：
这里的(主要(区别在于，函数将在不同的CPU和内存空间中运行。
from flask import Flask
from multiprocessing import Process
import time
# Helper function to easly parallelize multiple functions
def parallelize_functions(*functions):
processes = []
for function in functions:
p = Process(target=function)
p.start()
processes.append(p)
for p in processes:
p.join()
# The function that will run in parallel with the Flask app
def while_function():
i = 0
while i < 20:
time.sleep(1)
print(i)
i += 1
app = Flask(__name__)
@app.route("/")
def hello_world():
return "Hello, World!"
def run_app():
app.run(debug=False)
if __name__ == '__main__':
parallelize_functions(while_function, run_app)
如果您想使用@Triet Doan提出的before_first_request：您必须将while函数作为before_first_request的参数，如下所示：
from flask import Flask
import time
app = Flask(__name__)
def while_function(arg):
i = 0
while i < 5:
time.sleep(1)
print(i)
i += 1
@app.before_first_request(while_function)
@app.route("/")
def index():
print("index is running!")
return "Hello world"
if __name__ == "__main__":
app.run()
在这个设置中，while函数将被执行，当它完成时，你的应用程序将运行，但我不认为这是你想要的吗？
更新2023
要在启动Flask服务器之前运行一些代码，现在建议使用ApplicationFactory模式。
基本上，这个想法是用一个返回app对象的适当函数来替换app = Flask(__name__)。例如，我们可以在__init__.py:中有这样的东西
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
# Globally accessible libraries
db = SQLAlchemy()
r = FlaskRedis()
def init_app():
"""Initialize the core application."""
app = Flask(__name__, instance_relative_config=False)
app.config.from_object('config.Config')
# Initialize Plugins
db.init_app(app)
r.init_app(app)
with app.app_context():
# before_first_request equivalent here
# Include our Routes
from . import routes
# Register Blueprints
app.register_blueprint(auth.auth_bp)
app.register_blueprint(admin.admin_bp)
return app
所以，无论以前放在before_first_request下的是什么，现在都放在with app.app_context():下。
然后，在应用程序入口(例如main.py(中，我们简单地使用它：
app = init_app()
if __name__ == "__main__":
app.run(host='0.0.0.0')
有关更多信息，推荐这篇不错的文章。
自烧瓶2.2以来已弃用
您可以改为使用before_first_request。用@app.before_first_request修饰的函数将在对该应用程序实例的第一个请求之前运行一次。
代码如下：
from flask import Flask
app = Flask(__name__)
@app.route("/")
def index():
print("index is running!")
return "Hello world"
@app.before_first_request
def before_first_request_func():
print("This function will run once")
if __name__ == "__main__":
app.run(host="0.0.0.0")
before_first_request_func中的代码将在向服务器发出第一个请求之前执行一次。因此，在启动Flask实例之后，可以使用curl左右来模拟对服务器的第一个请求