日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學無先后，達者為師

網站首頁 編程語言 正文

Python異步處理返回進度——使用Flask實現進度條_python

作者：XerCis ? 更新時間： 2022-07-04 編程語言

使用Flask實現進度條

問題描述

Python異步處理，新起一個進程返回處理進度

解決方案

使用 tqdm 和 multiprocessing.Pool

安裝

pip install tqdm

代碼

import time
import threading
from multiprocessing import Pool
from tqdm import tqdm
def do_work(x):
    time.sleep(x)
    return x
def progress():
    time.sleep(3)  # 3秒后查進度
    print(f'任務有: {pbar.total} 已完成:{pbar.n}')
tasks = range(10)
pbar = tqdm(total=len(tasks))
if __name__ == '__main__':
    thread = threading.Thread(target=progress)
    thread.start()
    results = []
    with Pool(processes=5) as pool:
        for result in pool.imap_unordered(do_work, tasks):
            results.append(result)
            pbar.update(1)
    print(results)

效果

Flask

安裝

pip install flask

main.py

import time
from multiprocessing import Pool
from tqdm import tqdm
from flask import Flask, make_response, jsonify
app = Flask(__name__)
def do_work(x):
    time.sleep(x)
    return x
total = 5  # 總任務數
tasks = range(total)
pbar = tqdm(total=len(tasks))
@app.route('/run/')
def run():
    """執行任務"""
    results = []
    with Pool(processes=2) as pool:
        for _result in pool.imap_unordered(do_work, tasks):
            results.append(_result)
            if pbar.n >= total:
                pbar.n = 0  # 重置
            pbar.update(1)
    response = make_response(jsonify(dict(results=results)))
    response.headers.add('Access-Control-Allow-Origin', '*')
    response.headers.add('Access-Control-Allow-Headers', '*')
    response.headers.add('Access-Control-Allow-Methods', '*')
    return response
@app.route('/progress/')
def progress():
    """查看進度"""
    response = make_response(jsonify(dict(n=pbar.n, total=pbar.total)))
    response.headers.add('Access-Control-Allow-Origin', '*')
    response.headers.add('Access-Control-Allow-Headers', '*')
    response.headers.add('Access-Control-Allow-Methods', '*')
    return response

啟動（以 Windows 為例）

set FLASK_APP=main
flask run

接口列表

  • 執行任務：http://127.0.0.1:5000/run/
  • 查看進度：http://127.0.0.1:5000/progress/

test.html

<!DOCTYPE html>
<html lang="zh">
<head>
    <meta charset="UTF-8">
    <title>進度條</title>
    <script src="https://cdn.bootcss.com/jquery/3.0.0/jquery.min.js"></script>
    <script src="https://cdn.bootcdn.net/ajax/libs/twitter-bootstrap/3.3.7/js/bootstrap.min.js"></script>
    <link    >
</head>
<body>
<button id="run">執行任務</button>
<br><br>
<div class="progress">
    <div class="progress-bar" role="progressbar" aria-valuenow="1" aria-valuemin="0" aria-valuemax="100"
         style="width: 10%">0.00%
    </div>
</div>
</body>
<script>
    function set_progress_rate(n, total) {
        //設置進度
        var rate = (n / total * 100).toFixed(2);
        if (n > 0) {
            $(".progress-bar").attr("aria-valuenow", n);
            $(".progress-bar").attr("aria-valuemax", total);
            $(".progress-bar").text(rate + "%");
            $(".progress-bar").css("width", rate + "%");
        }
    }
    $("#run").click(function () {
        //執行任務
        $.ajax({
            url: "http://127.0.0.1:5000/run/",
            type: "GET",
            success: function (response) {
                set_progress_rate(100, 100);
                console.log('執行完成，結果為：' + response['results']);
            }
        });
    });
    setInterval(function () {
        //每1秒請求一次進度
        $.ajax({
            url: "http://127.0.0.1:5000/progress/",
            type: "GET",
            success: function (response) {
                console.log(response);
                var n = response["n"];
                var total = response["total"];
                set_progress_rate(n, total);
            }
        });
    }, 1000);
</script>
</html>

效果

Flask使用簡單異步任務

在Flask中使用簡單異步任務最簡潔優雅的原生實現：

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 was 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()

原文鏈接：https://xercis.blog.csdn.net/article/details/121920979

欄目分類
最近更新