能否为Python内置list.sort()排序列表添加进度条?
给Python内置排序添加大致进度条的可行方案
针对你的需求,直接给list.sort()加进度条确实做不到——因为Python内置的Timsort排序没有暴露中间进度的钩子。但可以通过以下几种思路实现大致的进度提示,满足你要的25%/50%/75%这类粗略进度:
方案1:分块排序+合并(兼顾效率与进度准确性)
把大列表拆分成若干小分块,用内置sort分别排序每个分块,再合并所有有序分块。每完成一个分块排序、每合并一批元素都更新进度条,进度颗粒度可以通过分块数量控制。
示例代码(用tqdm做进度条):
from typing import List import tqdm import heapq class Foo: def __init__(self, x): self.x = x def sort_with_progress(my_list: List[Foo], chunk_size: int = 5000): # 拆分列表为若干小分块 chunks = [my_list[i:i+chunk_size] for i in range(0, len(my_list), chunk_size)] # 逐个排序分块,更新进度 sorted_chunks = [] for chunk in tqdm.tqdm(chunks, desc="排序分块", total=len(chunks)): chunk.sort(key=lambda item: item.x) sorted_chunks.append(chunk) # 合并所有有序分块,实时更新进度 merged = [] total_elements = len(my_list) with tqdm.tqdm(total=total_elements, desc="合并分块") as pbar: for item in heapq.merge(*sorted_chunks, key=lambda item: item.x): merged.append(item) pbar.update(1) return merged # 测试用例 myList = [Foo(i) for i in reversed(range(50000))] sorted_list = sort_with_progress(myList)
优点:保留了内置sort的高效性,进度提示相对准确;缺点:合并步骤需要额外内存(5万元素完全没问题)。
方案2:自定义排序算法+进度计数
自己实现排序逻辑(比如快速排序),在排序过程中统计已处理的元素数量,实时更新进度条。虽然效率略低于内置sort,但能提供更实时的进度反馈。
示例代码:
from typing import List import tqdm class Foo: def __init__(self, x): self.x = x def quicksort_with_progress(arr: List[Foo], key, pbar): if len(arr) <= 1: pbar.update(len(arr)) return arr pivot_val = key(arr[len(arr)//2]) left = [item for item in arr if key(item) < pivot_val] middle = [item for item in arr if key(item) == pivot_val] right = [item for item in arr if key(item) > pivot_val] return quicksort_with_progress(left, key, pbar) + middle + quicksort_with_progress(right, key, pbar) def sort_with_progress(my_list: List[Foo]): total = len(my_list) with tqdm.tqdm(total=total, desc="排序中") as pbar: sorted_list = quicksort_with_progress(my_list.copy(), key=lambda item: item.x, pbar=pbar) return sorted_list # 测试用例 myList = [Foo(i) for i in reversed(range(50000))] sorted_list = sort_with_progress(myList)
优点:进度反馈实时性强;缺点:自定义排序效率略低于内置Timsort。
方案3:时间预估法(完全保留内置sort效率)
先拿小样本测试排序耗时,预估整体排序时间,然后通过线程+定时刷新的方式显示大致进度。进度是基于时间的估算,虽然不够精准,但完全保留了内置sort的最高效率。
示例代码:
from typing import List import tqdm import time import threading class Foo: def __init__(self, x): self.x = x def sort_in_thread(my_list, key, result_container): my_list.sort(key=key) result_container.append(my_list) def sort_with_progress(my_list: List[Foo]): # 用1000个元素测试耗时,预估总时间 sample = my_list[:1000].copy() start = time.time() sample.sort(key=lambda item: item.x) sample_cost = time.time() - start # 加1.2倍系数抵消大样本的额外开销 estimated_total = (len(my_list)/1000)*sample_cost*1.2 # 启动排序线程 result = [] sort_thread = threading.Thread(target=sort_in_thread, args=(my_list.copy(), lambda item: item.x, result)) sort_thread.start() # 按预估时间更新进度条 with tqdm.tqdm(total=100, desc="排序中", bar_format="{l_bar}{bar}| {n_fmt}%") as pbar: start_time = time.time() while sort_thread.is_alive(): elapsed = time.time() - start_time progress = min(int((elapsed/estimated_total)*100), 99) pbar.n = progress pbar.refresh() time.sleep(0.5) # 排序完成后强制设为100% pbar.n = 100 pbar.refresh() return result[0] # 测试用例 myList = [Foo(i) for i in reversed(range(50000))] sorted_list = sort_with_progress(myList)
优点:完全复用内置sort的高效性;缺点:进度是预估的,可能和实际排序速度有偏差。
内容的提问来源于stack exchange,提问作者M.X
相关产品推荐
相关产品推荐

