You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

能否为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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.16 06:35:19