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

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) 

pip install flask

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 

set FLASK_APP=main flask run

执行任务

0.00% 

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