Flask 后台任务|极客笔记

Flask 后台任务

在开发 Web 应用程序时，有时候我们需要执行一些耗时的任务，例如发送邮件、处理大量数据等。为了不影响用户体验，我们可以使用后台任务来进行异步处理，即将这些任务放在后台线程或进程中执行，而不阻塞主线程。

Flask 是一个轻量级的 Python Web 框架，它提供了丰富的扩展库来简化开发。在本文中，我们将介绍如何在 Flask 应用程序中实现后台任务。

使用 Flask-RQ 执行后台任务

Flask-RQ 是一个 Flask 扩展，用于将 RQ（Redis Queue）集成到 Flask 应用程序中，以便在后台执行任务。RQ 是一个 Python 库，用于处理后台任务队列，它基于 Redis 数据库。

为了使用 Flask-RQ，我们首先需要安装它：

pip install Flask-RQ

然后在 Flask 应用程序中配置并初始化 RQ：

from flask import Flask
from flask_rq2 import RQ

app = Flask(__name__)
app.config['RQ_REDIS_URL'] = 'redis://localhost:6379/0'
rq = RQ(app)

在上面的代码中，我们首先导入 Flask 和 RQ，并创建一个 Flask 应用程序实例。然后配置了 Redis 数据库的连接 URL，并将其传递给 RQ 初始化。

接下来，我们可以定义一个简单的异步任务，并将其添加到任务队列中：

from rq import get_current_job

@rq.job
def send_email(user_email, subject, message):
    job = get_current_job()
    job.meta['progress'] = 0

    for i in range(10):
        job.meta['progress'] = i + 1
        job.save_meta()
        time.sleep(1)

    send_email_to_user(user_email, subject, message)

在上面的代码中，我们定义了一个名为 send_email 的异步任务，它接收用户邮箱、主题和消息作为参数。任务会在后台发送邮件，并且在任务执行过程中，更新任务进度，通过 job.save_meta() 方法存储任务元数据。

接下来，我们可以在 Flask 视图函数中调用该任务：

from flask import request

@app.route('/send_email', methods=['POST'])
def send_email_view():
    user_email = request.form.get('email')
    subject = request.form.get('subject')
    message = request.form.get('message')

    job = send_email.queue(user_email, subject, message)

    return f'Task {job.id} queued.'

在上面的代码中，我们创建了一个 send_email_view 视图函数，它接收用户请求中的邮箱、主题和消息，并将 send_email 任务加入队列。最后返回任务 ID。

启动 RQ Worker

为了执行任务队列中的任务，我们需要启动 RQ Worker。RQ Worker 是一个单独的进程，负责从队列中获取任务并执行它们。

我们可以使用以下命令启动 RQ Worker：

rq worker

在启动 Worker 后，它将开始监听队列中的任务，并执行它们。需要注意的是，RQ Worker 必须在 Flask 应用程序可以访问到的环境中运行，例如在同一个虚拟环境下。

监控任务进度

有时候我们需要监控任务的执行进度，以便及时通知用户或者显示进度条。在 Flask-RQ 中，我们可以通过任务元数据来实现任务进度的监控。

下面是一个示例代码，用于获取任务的进度信息：

from rq.job import Job

def get_job_progress(job_id):
    job = Job.fetch(job_id, connection=rq.connection)
    progress = job.meta.get('progress', 0)
    return progress

在上面的代码中，我们定义了一个 get_job_progress 函数，用于获取指定任务 ID 的任务进度信息。函数首先根据任务 ID 获取任务对象，然后获取任务的进度信息。

完整示例

下面是一个完整的 Flask 应用程序示例，用于演示后台任务的使用：

from flask import Flask, request
from flask_rq2 import RQ
from rq import get_current_job
import time

app = Flask(__name__)
app.config['RQ_REDIS_URL'] = 'redis://localhost:6379/0'
rq = RQ(app)

@rq.job
def send_email(user_email, subject, message):
    job = get_current_job()
    job.meta['progress'] = 0

    for i in range(10):
        job.meta['progress'] = i + 1
        job.save_meta()
        time.sleep(1)

    send_email_to_user(user_email, subject, message)

def send_email_to_user(user_email, subject, message):
    print(f'Sending email to {user_email} with subject {subject} and message {message}...')

@app.route('/send_email', methods=['POST'])
def send_email_view():
    user_email = request.form.get('email')
    subject = request.form.get('subject')
    message = request.form.get('message')

    job = send_email.queue(user_email, subject, message)

    return f'Task {job.id} queued.'

@app.route('/get_job_progress/<job_id>', methods=['GET'])
def get_job_progress_view(job_id):
    progress = get_job_progress(job_id)

    return f'Job progress: {progress}'

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

在上面的代码中，我们定义了一个简单的 Flask 应用程序，包含了发送邮件的后台任务以及获取任务进度的接口。我们可以通过 POST 请求 /send_email 接口来触发发送邮件任务，并通过 GET 请求 /get_job_progress/<job_id> 接口来获取任务进度信息。