Flask 从 Flask 端点启动长时间运行的脚本

在本文中，我们将介绍如何在 Flask 应用程序的端点中启动一个长时间运行的脚本。有时我们可能需要在 Flask 应用程序中运行一些耗时的任务，例如大数据处理、机器学习训练或者其他需要较长时间完成的任务。通过将这些任务封装为一个脚本，并从 Flask 的端点中触发执行，我们可以在保持应用程序响应性的同时，完成这些任务。

阅读更多：Flask 教程

使用 Flask-Celery 执行异步任务

Flask-Celery 是一个 Flask 扩展，它提供了一种简单而强大的方式来执行异步任务。它基于 Celery（一个 Python 的分布式任务队列框架），可以将耗时的任务放入队列中异步执行，而不会阻塞 Flask 应用程序的主线程。

首先，我们需要在 Flask 项目的环境中安装 Flask-Celery。可以使用以下命令执行安装：

pip install Flask-Celery

安装完成后，我们可以在 Flask 应用程序的主文件中配置和初始化 Flask-Celery。下面是一个简单的示例：

from flask import Flask
from flask_celery import make_celery

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = make_celery(app)

@celery.task()
def long_running_task():
    # 执行耗时任务的代码
    ...

@app.route('/run-task')
def run_task():
    long_running_task.delay()
    return 'Task has been sent to the queue.'

if __name__ == '__main__':
    app.run()

在上面的示例中，我们首先导入了 Flask 和 Flask-Celery 扩展。然后，我们创建了一个 Flask 应用程序，并配置了 Redis 作为 Celery 的消息代理（broker）和结果存储后端（result backend）。接下来，我们使用 Flask-Celery 的 make_celery 函数创建了一个 Celery 实例，并将其与 Flask 应用程序关联起来。

然后，我们定义了一个使用了 @celery.task() 装饰器的长时间运行任务函数 long_running_task()。在这个函数中，我们可以编写任何耗时的任务代码。

最后，我们创建了一个 Flask 端点 /run-task，当访问该端点时，会触发 long_running_task() 的执行。我们使用 delay() 方法将任务放入 Celery 的队列中，并立即返回一个成功的响应。

启动 Flask 应用程序后，我们可以访问 /run-task 端点，触发异步任务的执行。

使用 Flask-SocketIO 进行实时通信

有时，我们需要从长时间运行脚本中获取实时结果，或者向客户端提供实时进度更新。这时，可以使用 Flask-SocketIO 扩展来实现实时通信功能。

首先，我们需要在 Flask 项目的环境中安装 Flask-SocketIO。可以使用以下命令执行安装：

pip install flask-socketio

安装完成后，我们可以在 Flask 应用程序的主文件中配置和初始化 Flask-SocketIO。下面是一个简单的示例：

from flask import Flask, render_template
from flask_socketio import SocketIO, emit

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
socketio = SocketIO(app)

@socketio.on('connect')
def handle_connect():
    emit('message', 'Connected')

@socketio.on('run-task')
def handle_run_task():
    emit('message', 'Task is running')
    # 执行耗时任务的代码
    ...

if __name__ == '__main__':
    socketio.run(app)

在上面的示例中，我们首先导入了 Flask、Flask-SocketIO 和 emit（用于发送消息的函数）。然后，我们创建了一个 Flask 应用程序，并配置了一个秘钥。

接下来，我们使用 Flask-SocketIO 的 SocketIO 类创建了一个 SocketIO 实例，并将其与 Flask 应用程序关联起来。

然后，我们定义了一个事件处理器 handle_connect()，用于处理客户端的连接事件。在这个处理器中，我们使用 emit() 函数向客户端发送一条消息。

接着，我们定义了另一个事件处理器 handle_run_task()，用于处理客户端发送的 run-task 事件。在这个处理器中，我们使用 emit() 函数向客户端发送一条任务开始的消息，并执行耗时的任务代码。

最后，我们通过调用 socketio.run(app) 启动 Flask-SocketIO 服务器。

启动 Flask 应用程序后，客户端可以通过 SocketIO 进行连接，并监听 message 事件以接收来自服务器的实时消息。

总结

在本文中，我们介绍了如何从 Flask 的端点中启动一个长时间运行的脚本。我们使用了 Flask-Celery 来执行异步任务，并使用 Flask-SocketIO 来实现实时通信功能。通过将耗时的任务放在队列中异步执行，并向客户端提供实时结果或进度更新，我们可以保持 Flask 应用程序的响应性，并完成那些耗时的任务。Flask-Celery 和 Flask-SocketIO 是非常强大和灵活的扩展，可以帮助我们更好地构建和扩展 Flask 应用程序。希望本文对于对 Flask 开发有兴趣的读者能有所帮助。

参考资料

• Flask-Celery Documentation
• Flask-SocketIO Documentation