Python 中的进度跟踪

最后修改日期：2025 年 4 月 1 日

对于 Python 中的长时间运行操作，跟踪进度至关重要。本教程介绍了实现进度跟踪的各种方法，从简单的控制台指示器到高级 GUI 进度条。

我们将探讨多种方法，包括流行的 tqdm 库、手动进度条、基于回调的跟踪以及与不同 Python 框架的集成。每种方法都有其针对不同用例的优势。

基本控制台进度

简单的进度跟踪可以使用 print 语句来实现。

basic_progress.py

import time

total = 100
for i in range(total):
    # Process item here
    time.sleep(0.1)  # Simulate work
    
    # Update progress
    progress = (i + 1) / total * 100
    print(f"\rProgress: {progress:.1f}%", end="")

print("\nDone!")

此示例显示了在控制台应用程序中跟踪进度的最简单方法。\r 回车符将光标移动到行首，从而可以在同一行上更新进度。end="" 参数阻止换行。

虽然基本，但此方法可以在任何地方工作，并且不需要任何依赖项。它非常适合只需要显示完成百分比的简单脚本。sleep 调用模拟演示的处理时间。

使用 tqdm 显示进度条

tqdm 库以最少的代码提供高级进度条。

tqdm_progress.py

from tqdm import tqdm
import time

# Basic progress bar
for i in tqdm(range(100)):
    time.sleep(0.02)  # Simulate work

# With description and unit
items = ["apple", "banana", "cherry", "date"]
for item in tqdm(items, desc="Processing", unit="item"):
    time.sleep(0.5)

tqdm 自动显示进度条，其中包含剩余估计时间、每秒迭代次数等。desc 参数添加描述，而 unit 指定正在处理的内容。

该库自动处理所有格式化和更新。它可以在 Jupyter notebooks 和常规控制台中工作。tqdm 是 Python 中大多数进度跟踪需求的常用解决方案。

手动进度条

对于自定义需求，您可以构建自己的进度条。

manual_bar.py

import sys
import time

def draw_progress(progress, width=50):
    filled = int(width * progress)
    bar = "[" + "=" * filled + " " * (width - filled) + "]"
    sys.stdout.write(f"\r{bar} {progress*100:.1f}%")
    sys.stdout.flush()

total = 200
for i in range(total):
    time.sleep(0.02)
    draw_progress((i + 1) / total)
    
print("\nComplete!")

此实现使您可以完全控制进度条的外观。draw_progress 函数创建一个指定宽度的条，其填充与当前进度成比例。

使用 sys.stdout.write 和 flush 可确保平滑更新。当您需要自定义格式或无法使用外部库时，此方法非常有用。

文件操作的进度跟踪

使用回调跟踪文件操作期间的进度。

file_progress.py

import shutil
import os

def copy_with_progress(src, dst):
    total = os.path.getsize(src)
    copied = 0
    
    def callback(chunk):
        nonlocal copied
        copied += len(chunk)
        progress = copied / total * 100
        print(f"\rCopying: {progress:.1f}%", end="")
    
    shutil.copyfileobj(
        open(src, "rb"),
        open(dst, "wb"),
        callback=callback
    )
    print("\nDone!")

copy_with_progress("large_file.dat", "copy.dat")

此示例演示了文件复制操作期间的进度跟踪。回调函数接收正在复制的数据块，并相应地更新进度显示。

该方法适用于任何类文件对象，使其可用于网络传输或自定义数据处理管道。 nonlocal 关键字允许从嵌套函数修改 copied 计数器。

线程进度更新

对于后台任务，请使用线程来分别更新进度。

threaded_progress.py

import threading
import time
import random

class ProgressTracker:
    def __init__(self, total):
        self.progress = 0
        self.total = total
        self.running = True
        
    def update(self):
        while self.running and self.progress < self.total:
            self.progress += random.randint(1, 5)
            time.sleep(0.1)
    
    def display(self):
        while self.running and self.progress < self.total:
            pct = self.progress / self.total * 100
            print(f"\rProgress: {pct:.1f}%", end="")
            time.sleep(0.05)
        print("\nDone!")

tracker = ProgressTracker(100)
worker = threading.Thread(target=tracker.update)
display = threading.Thread(target=tracker.display)

worker.start()
display.start()
worker.join()
tracker.running = False
display.join()

此模式将工作和进度显示分离到不同的线程中。ProgressTracker 类使用共享的进度变量在它们之间进行协调。

running 标志确保干净的关闭。此方法非常适合 GUI 应用程序，或者当正在完成的工作本身无法轻松报告进度时。

Jupyter 中的进度跟踪

Jupyter notebooks 支持丰富的进度显示。

jupyter_progress.ipynb

from IPython.display import display
import ipywidgets as widgets
import time

progress = widgets.FloatProgress(
    value=0, min=0, max=100, description="Loading:"
)
display(progress)

for i in range(100):
    time.sleep(0.05)
    progress.value = i + 1

IPython 提供特殊的 widget 用于交互式进度跟踪。FloatProgress widget 创建一个实时更新的可视条。

此方法在 notebooks 中创建专业外观的进度指示器。这些 widget 与 Jupyter 的显示系统无缝集成，并支持其他功能，例如样式设置和多个进度条。

使用多进程进行进度跟踪

使用共享内存跨多个进程跟踪进度。

multiprocessing_progress.py

import multiprocessing
import time

def worker(shared_progress, total_work):
    for i in range(total_work):
        time.sleep(0.1)
        with shared_progress.get_lock():
            shared_progress.value += 1

def display_progress(shared_progress, total):
    while True:
        with shared_progress.get_lock():
            current = shared_progress.value
        progress = current / total * 100
        print(f"\rProgress: {progress:.1f}%", end="")
        if current >= total:
            break
        time.sleep(0.1)

if __name__ == "__main__":
    total_tasks = 100
    progress = multiprocessing.Value("i", 0)
    
    processes = []
    for _ in range(4):
        p = multiprocessing.Process(
            target=worker,
            args=(progress, total_tasks//4)
        )
        processes.append(p)
        p.start()
    
    display_progress(progress, total_tasks)
    
    for p in processes:
        p.join()
    print("\nAll tasks completed!")

此示例演示了跨多个工作进程的进度跟踪。共享的 Value 在进程之间协调进度更新。

get_lock() 上下文管理器可确保线程安全的更新。对于分布在多个核心上的 CPU 密集型任务，同时仍提供进度反馈，此模式至关重要。

使用 Tkinter 的 GUI 进度条

为桌面应用程序创建图形进度条。

tkinter_progress.py

import tkinter as tk
from tkinter import ttk
import threading
import time

class ProgressApp:
    def __init__(self, root):

        self.root = root
        self.progress = ttk.Progressbar(
            root, orient="horizontal",
            length=300, mode="determinate"
        )

        self.progress.pack(pady=20)
        
        self.button = tk.Button(
            root, text="Start Task",
            command=self.start_task
        )

        self.button.pack()
    
    def start_task(self):

        self.button.config(state="disabled")
        thread = threading.Thread(target=self.run_task)
        thread.start()
        self.monitor(thread)
    
    def run_task(self):

        for i in range(100):
            time.sleep(0.05)
            self.progress["value"] = i + 1
    
    def monitor(self, thread):
    
        if thread.is_alive():
            self.root.after(100, lambda: self.monitor(thread))
        else:
            self.button.config(state="normal")

root = tk.Tk()
app = ProgressApp(root)
root.mainloop()

此 Tkinter 应用程序显示一个专业的进度条，该进度条在后台任务期间更新。线程可防止长时间操作期间 GUI 冻结。

monitor 方法定期检查线程状态。此模式确保响应迅速的 UI，同时为需要一段时间才能完成的操作提供视觉反馈。

使用 Flask 进行 Web 进度跟踪

在 Web 应用程序中实现进度跟踪。

flask_progress.py

from flask import Flask, jsonify, render_template_string
import time
import threading

app = Flask(__name__)

progress_data = {"value": 0, "total": 100}

def long_running_task():
    for i in range(progress_data["total"]):
        time.sleep(0.1)
        progress_data["value"] = i + 1

@app.route("/")
def index():
    return render_template_string("""
        <h1>Progress Demo</h1>
        <div id="progress">0%</div>
        <button onclick="startTask()">Start</button>
        <script>
        function startTask() {
            fetch("/start").then(() => pollProgress());
        }
        function pollProgress() {
            fetch("/progress").then(r => r.json())
                .then(data => {
                    document.getElementById("progress").innerHTML = 
                        data.value + "%";
                    if (data.value < 100) {
                        setTimeout(pollProgress, 500);
                    }
                });
        }
        </script>
    """)

@app.route("/start")
def start():
    thread = threading.Thread(target=long_running_task)
    thread.start()
    return "Started"

@app.route("/progress")
def progress():
    return jsonify({
        "value": progress_data["value"],
        "total": progress_data["total"]
    })

if __name__ == "__main__":
    app.run(threaded=True)

此 Flask 应用程序演示了基于 Web 的进度跟踪。前端轮询后端以获取进度更新，而任务在后台线程中运行。

template_string 显示一个最小的 HTML 界面。在生产中，您将分离模板并添加错误处理。threaded=True 选项允许 Flask 处理并发请求。

最佳实践

选择正确的方法： CLI 使用 tqdm，Jupyter 使用 widget
更新频率：不要太频繁地更新（至少 100 毫秒）
线程安全：使用锁保护共享的进度变量
错误处理：确保进度在出错时停止
用户体验：提供清晰的标签和完成消息

资料来源

作者

我的名字是 Jan Bodnar，我是一位充满激情的程序员，拥有丰富的编程经验。自 2007 年以来，我一直在撰写编程文章。迄今为止，我已经撰写了 1,400 多篇文章和 8 本电子书。我拥有超过十年的编程教学经验。

列出所有 Python 教程。