我正在创建一个Flask应用程序，对于某个请求，我需要运行一些不需要等待UI的长时间运行的作业。我将创建一个线程并向UI发送消息。 线程将计算并更新数据库。 但是，提交后UI将看到一条消息。 以下是我的实现，但它先运行线程，然后发送输出到UI，这不是我想要的方式。 我该如何在后台运行此线程？
@app.route('/someJob')
def index():
t1 = threading.Thread(target=long_running_job)
t1.start()
return 'Scheduled a job'
def long_running_job
#some long running processing here
如何使线程t1在后台运行并立即返回消息？
请尝试此示例，经过 Python 3.4.3 / Flask 0.11.1 测试。
from flask import Flask
from time import sleep
from concurrent.futures import ThreadPoolExecutor
# DOCS https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
executor = ThreadPoolExecutor(2)
app = Flask(__name__)
@app.route('/jobs')
def run_jobs():
executor.submit(some_long_task1)
executor.submit(some_long_task2, 'hello', 123)
return 'Two jobs were launched in background!'
def some_long_task1():
print("Task #1 started!")
sleep(10)
print("Task #1 is done!")
def some_long_task2(arg1, arg2):
print("Task #2 started with args: %s %s!" % (arg1, arg2))
sleep(5)
print("Task #2 is done!")
if __name__ == '__main__':
app.run()
看看 Flask-Executor，它在后台使用concurrent.futures使您的生活变得非常容易。
from flask_executor import Executor
executor = Executor(app)
@app.route('/someJob')
def index():
executor.submit(long_running_job)
return 'Scheduled a job'
def long_running_job
#some long running processing here
这不仅可以在后台运行作业，还可以让它们访问应用程序上下文。它还提供了一种存储作业的方式，以便用户可以返回并获取状态。
处理类似问题最好使用消息代理。在Python世界中有一些出色的软件专门用于处理这种情况：
两者都是非常不错的选择。
通过像您现在所做的方式生成线程通常不是一个好主意，因为这可能会导致处理传入请求时出现问题等等。
如果您查看 celery 或 RQ 入门指南，它们将向您详细介绍正确的操作方法！
ThreadPoolExecutor并处理异常相比）这会更容易一些：
cli.py）-因为所有Web应用程序都应该有一个管理cli。subprocess.Popen（不等待）打开命令行。# cli.py
import click
import yourpackage.app
import yourpackage.domain
app = yourpackage.app.create_app()
@click.group()
def cli():
pass
@click.command()
@click.argument('foo_id')
def do_something(foo_id):
with app.app_context():
yourpackage.domain.do_something(foo_id)
if __name__ == '__main__':
cli.add_command(do_something)
cli()
# admin.py (flask view / controller)
bp = Blueprint('admin', __name__, url_prefix='/admin')
@bp.route('/do-something/<int:foo_id>', methods=["POST"])
@roles_required('admin')
def do_something(foo_id):
yourpackage.domain.process_wrapper_do_something(foo_id)
flash("Something has started.", "info")
return redirect(url_for("..."))
并且：
# domain.py
import subprocess
def process_wrapper_do_something(foo_id):
command = ["python3", "-m", "yourpackage.cli", "do_something", str(foo_id)]
subprocess.Popen(command)
def do_something(foo_id):
print("I am doing something.")
print("This takes some time.")
同意@rdegges标记的答案。很抱歉我的账户没有足够的信用在答案下添加评论，但我想明确“为什么使用消息代理而不是生成线程（或进程）”。
关于ThreadPoolExecutor和flask_executor的其他答案正在创建一个新的线程（或进程，因为flask_executor能够执行）来执行“long_running_job”。这些新的线程/进程将具有与主网站相同的上下文：
对于线程：新线程将能够访问网站应用程序的上下文，更改内容或破坏它，如果此线程引发异常； 对于进程：新进程将拥有网站应用程序上下文的副本。如果网站在初始化时某种方式使用了大量内存，则新进程也将拥有其副本，即使该进程不打算利用该内存部分。
另一方面，如果您使用消息代理和另一个应用程序检索作业消息以处理它，则新应用程序将与网站应用程序无关，也不会复制Web应用程序中的内存。
将来，当您的应用程序足够大时，可以将应用程序放置在另一个服务器（或多个服务器）中，轻松扩展。
some_long_task1或some_long_task2引发异常会怎么样？我写了类似的代码，但发现异常被抑制了。 - secsilm