Python异步处理返回进度——使用Flask实现进度条

本文将详细讲解如何使用 Flask 框架实现 Python 异步处理返回进度，并对进度进行可视化展示，以进度条形式向用户展示异步处理的进度。本文将分为两个部分来讲解，第一部分将介绍如何使用 Flask 实现异步处理并返回进度，第二部分将介绍如何使用 JavaScript 和 Bootstrap 实现进度条。

第一部分：Flask 实现异步处理返回进度

1. 准备工作

我们首先需要安装 Flask 和 Celery，你可以使用 pip 命令来安装，如下所示：

pip install flask celery

2. 建立 Flask 应用

在 Flask 中使用异步处理需要使用 Celery 库，因此我们需要在应用中初始化 Celery。我们可以使用如下代码来初始化 Celery：

from celery import Celery
from flask import Flask

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

celery = Celery(
    app.name,
    broker=app.config['CELERY_BROKER_URL'],
    backend=app.config['CELERY_RESULT_BACKEND']
)
celery.conf.update(app.config)

在上面的代码中，我们使用了 Redis 作为消息队列，因此需要安装 Redis 并启动 Redis 服务。如果你使用的是其他消息队列，例如 RabbitMQ，也可以在上面的代码中进行配置。

3. 实现异步任务

现在我们来实现异步任务，下面是一个简单的异步任务的代码：

import time
from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379', backend='redis://localhost:6379')

@celery.task(bind=True)
def long_task(self):
    """Background task that runs a long function with progress reports."""
    verb = ['Starting up', 'Booting', 'Repairing', 'Loading', 'Checking']
    adjective = ['master', 'radiant', 'silent', 'harmonic', 'fast']
    noun = ['solar array', 'particle reshaper', 'cosmic ray', 'orbiter', 'bit']
    message = ''
    total = 100

    for i in range(total):
        if not message or i % 3 == 0 or i == total - 1:
            message = f'{random.choice(verb)} {random.choice(adjective)} {random.choice(noun)}...'
        self.update_state(state='PROGRESS',
                          meta={'current': i, 'total': total, 'status': message})
        time.sleep(1)
    return {'current': 100, 'total': 100, 'status': 'Task completed!', 'result': 42}

在上面的代码中，我们定义了一个名为 long_task 的异步任务，该任务会模拟一个长时间运行的进程，并会每秒返回一次运行进度。在每次更新进度的时候，我们会使用 self.update_state() 方法来更新异步任务的状态。

4. 建立路由

现在我们需要建立一个路由来启动异步任务，并将异步任务的进度发送给客户端。下面是路由的代码：

from ..tasks import long_task

@app.route('/longtask', methods=['POST'])
def longtask():
    task = long_task.delay()
    return jsonify({}), 202, {'Location': url_for('taskstatus', task_id=task.id)}

@app.route('/status/<task_id>')
def taskstatus(task_id):
    task = long_task.AsyncResult(task_id)
    if task.state == 'PENDING':
        # Job did not start yet
        response = {
            'state': task.state,
            'current': 0,
            'total': 1,
            'status': 'Pending...'
        }
    elif task.state != 'FAILURE':
        # Job is running
        response = {
            'state': task.state,
            'current': task.info.get('current', 0),
            'total': task.info.get('total', 1),
            'status': task.info.get('status', '')
        }
        if 'result' in task.info:
            response['result'] = task.info['result']
    else:
        # Job failed
        response = {
            'state': task.state,
            'current': 1,
            'total': 1,
            'status': str(task.info)
        }

    return jsonify(response)

在上面的代码中，我们建立了两个路由，第一个路由是用来启动异步任务的，它会返回一个 202 Accepted 响应，表示任务已经被接受，但尚未完成。在该路由处理函数中，我们调用 long_task.delay() 方法来启动异步任务，并将任务 ID 保存在响应头的 Location 字段中，客户端可以使用这个 URL 检查任务的状态。

第二个路由用来返回异步任务的状态。我们首先使用任务 ID 获取任务对象，然后检查任务的状态。如果任务还没有开始，我们将返回一个状态为 PENDING 的响应。如果任务正在运行，我们返回当前任务的进度和状态。如果任务已经完成，我们返回状态为 SUCCESS，并且包含任务最终的结果。

第二部分：使用 JavaScript 和 Bootstrap 实现进度条

在本部分中，我们将使用 JavaScript 和 Bootstrap 来实现进度条。我们需要在客户端向服务器发送 HTTP 请求来获取异步任务的状态，并将异步任务的进度可视化展示。

1. 发送 HTTP 请求

我们可以使用 jQuery 来向服务器发送 HTTP 请求，并且定期检查异步任务的状态。下面是向服务器发送 HTTP 请求的代码：

function update_progress() {
    $.getJSON('{{ url_for('taskstatus', task_id=task_id) }}', function(data) {
        var percentage = Math.floor(data.current / data.total * 100);
        $('.progress-bar').css('width', percentage + '%');
        $('.progress-bar').text(percentage + '%');
        if (data.state != 'SUCCESS' && data.state != 'FAILURE') {
            setTimeout(update_progress, 1000);
        }
    });
}

在上面的代码中，我们首先使用 $.getJSON() 方法向服务器发送 HTTP 请求，并获取异步任务的状态。然后我们使用异步任务的进度计算百分比，并更新进度条的宽度和文本。

如果任务还没有完成，我们将使用 JavaScript 的 setTimeout() 方法定期检查任务的状态，以获取最新的进度信息。

2. 初始化进度条

在页面加载后，我们需要初始化进度条的状态。下面是初始化进度条的代码：

$(document).ready(function() {
    update_progress();
});

在上面的代码中，我们使用了 jQuery 的 $(document).ready() 方法，在页面加载完成后调用 update_progress() 函数来初始化进度条的状态。

3. HTML 代码

下面是包含进度条的 HTML 代码：

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Long Task</title>
    <link href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" rel="stylesheet">
    <script src="https://code.jquery.com/jquery-3.2.1.min.js"></script>
</head>
<body>
<div class="container">
    <h1>Long Task</h1>
    <div class="progress">
        <div class="progress-bar" role="progressbar" aria-valuenow="0" aria-valuemin="0" aria-valuemax="100">
            0%
        </div>
    </div>
</div>
<script>
    var task_id = '{{ task_id }}';
    function update_progress() {
        $.getJSON('{{ url_for('taskstatus', task_id=task_id) }}', function(data) {
            var percentage = Math.floor(data.current / data.total * 100);
            $('.progress-bar').css('width', percentage + '%');
            $('.progress-bar').text(percentage + '%');
            if (data.state != 'SUCCESS' && data.state != 'FAILURE') {
                setTimeout(update_progress, 1000);
            }
        });
    }
    $(document).ready(function() {
        update_progress();
    });
</script>
</body>
</html>

在上面的代码中，我们首先引入了 Bootstrap 的 CSS 文件和 jQuery 库，并定义了包含进度条的 HTML 代码。我们使用 Flask 的 url_for() 函数来动态生成路由 URL，并通过 JavaScript 设置了任务 ID。