最近我在学习多线程处理的Python代码，想加个进度条，但是tqdm（泰拳大妈）进度条只适用于单线程处理。我想要设计一个进度条能显示多线程处理的总进度，我琢磨了好久，终于自己写出来一个。

有一个缺点：进度条的显示在命令行才能在一行内显示，在IDLE里运行代码进度条会分多行显示。

2022/11/5更新：进度条的宽度的变量width在原先的代码中设为80，但在python输出窗口比较小的情况下，进度条会分行显示，为了保持在一行内输出进度条，width的值不宜设得太大，现在由80更正为60。

2023/2/7更新：改进代码，封装成一个函数，使之更容易理解和运用。

下面的代码主程序创建100个随机数字，然后进行多线程处理，多线程处理中使用等待随机的时长，模拟处理数据的耗时。

import time
import random
import threading
from queue import Queue
from threading import Timer


def multi_thread(total_thread, task_list, show_progress=True):
    global count, data, bar, percentage
    total_number = len(task_list)
    exit_flag = 0  # 线程退出的标志
    count = 0  # 统计进度
    bar = 0   # 画进度条的方格数
    percentage = 0  # 进度条百分比
    bar_width = 60    # 进度条宽度
    data = None
    threads = []
    queue_lock = threading.Lock()
    work_queue = Queue()
    now_time = time.time()

    class myThread (threading.Thread):
        def __init__(self, q):
            threading.Thread.__init__(self)
            self.q = q

        def run(self):
            while not exit_flag:
                queue_lock.acquire()
                if not work_queue.empty():
                    # 设定全局变量，使得其他子线程可以调用
                    global count, data
                    data = self.q.get()
                    queue_lock.release()
                    process_data(data)
                    count += 1
                else:
                    queue_lock.release()

    def update_progress():
        global count, data, bar, percentage
        bar = int(bar_width*count/total_number)
        percentage = int((count/total_number)*100)
        print(
            f'正在处理数据编号#{data}\t[{"█"*bar}{" "*(bar_width-bar)}] {percentage}％', end='\r', flush=True)

    # 创建新线程
    for i in range(total_thread):
        th = myThread(work_queue)
        th.daemon = False
        th.start()
        threads.append(th)

    # 填充队列
    queue_lock.acquire()
    for j in task_list:
        work_queue.put(j)

    # 等待队列清空
    queue_lock.release()
    while not work_queue.empty():
        # 每隔0.5秒更新进度条
        if show_progress:
            timer = Timer(0.5, update_progress)
            timer.start()
            timer.join()

    # 通知线程是时候退出
    exit_flag = 1
    # 修复最后90％~100％显不出来的bug
    percentage_difference = (100-percentage)/total_thread
    bar_difference = (bar_width-bar)/total_thread
    # 等待所有线程完成
    for t in threads:
        t.join()
        if show_progress:
            percentage += percentage_difference
            bar += bar_difference
            print(
                f'正在处理数据编号#{data}\t[{"█"*int(round(bar))}{" "*(bar_width-int(round(bar)))}] {int(round(percentage))}％', end='\r', flush=True)

    print(f'\n退出所有线程\n{total_thread}线程总共耗时：{time.time()-now_time:.2f}s。')


def process_data(data):
    time.sleep(random.random())  # 等待随机的秒数，模拟处理数据等待的时间


if __name__ == '__main__':
    task_list = []
    total_number = 100
    # 创建100个100以内的随机数字，装进列表
    for i in range(total_number):
        task_list.append(random.randint(1, 100))
    multi_thread(10, task_list)

在命令行里运行的截图：

在IDLE里运行的截图：