使用apply_async时tqdm进度条不更新的问题排查

你的进度条不更新的核心问题是**tqdm.update()传入的是当前累计值而非增量**，同时循环无意义空转占用CPU，导致进度条无法正常刷新。以下是两种可行的修复方案：

方案一：修复共享计数器的进度更新逻辑 #

修正add_elevation_mp方法中的进度条更新代码，每次计算进度增量而非直接传入当前值，并添加延迟避免空转：

import multiprocessing as mp
import pandas as pd
from tqdm import tqdm
import time  # 新增导入

class AAA:
    # 其他方法保持不变...

    @staticmethod
    def process_rows(chunk: pd.DataFrame, shared_locked, progress):
        for index, row in chunk.iterrows():
            # 你的行处理逻辑...
            with progress.get_lock():
                progress.value += 1
        return chunk

    def add_elevation_mp(self):
        # 省略shared_locked等初始化代码...
        num_rows = len(self.df)
        num_processes = mp.cpu_count()
        chunk_size = num_rows // num_processes
        chunks = [
            self.df_elevated[i : i + chunk_size]
            for i in range(0, len(self.df), chunk_size)
        ]

        with mp.Pool() as pool:
            progress = mp.Value("i", 0)  # 先初始化共享变量再创建进度条
            progress_bar = tqdm(total=num_rows, desc="---progress bar")
            processed_chunks = [
                pool.apply_async(
                    self.process_rows, args=(chunk, shared_locked, progress)
                )
                for chunk in chunks
            ]

            current_progress = 0
            while True:
                # 加锁读取当前进度，避免线程安全问题
                with progress.get_lock():
                    new_progress = progress.value
                if new_progress >= num_rows:
                    break
                # 计算增量并更新进度条
                delta = new_progress - current_progress
                if delta > 0:
                    progress_bar.update(delta)
                    current_progress = new_progress
                # 短延迟减少CPU占用
                time.sleep(0.1)

            # 等待所有进程完成
            processed_chunks = [result.get() for result in processed_chunks]
            progress_bar.close()

        processed_df = pd.concat(processed_chunks)
        self.df = processed_df

方案二：使用tqdm官方多进程集成（更简洁） #

tqdm提供了process_map工具，无需手动管理共享计数器，自动处理多进程进度条且不会出现覆盖问题：

import pandas as pd
from tqdm.contrib.concurrent import process_map
import multiprocessing as mp

class AAA:
    # 其他方法保持不变...

    @staticmethod
    def process_single_row(row):
        # 替换为你的单行处理逻辑，返回处理后的行数据
        # 示例：row['new_col'] = row['old_col'] * 2
        ...
        return row

    def add_elevation_mp(self):
        num_processes = mp.cpu_count()
        # 将DataFrame转为字典列表，方便逐行处理
        row_list = self.df_elevated.to_dict('records')
        # 使用process_map自动分块、多进程处理并维护进度条
        processed_rows = process_map(
            self.process_single_row, 
            row_list, 
            max_workers=num_processes, 
            total=len(row_list), 
            desc="---progress bar"
        )
        # 将处理后的行转回DataFrame
        processed_df = pd.DataFrame(processed_rows)
        self.df = processed_df

额外优化建议 #

• 避免使用iterrows()：它是Pandas中效率较低的逐行迭代方式，若业务允许，优先使用矢量化操作或df.apply()（单进程）；
• 分块大小优化：可根据每行处理的复杂度调整chunk_size，避免过小的块导致进程调度开销过大。

内容的提问来源于stack exchange，提问作者ilsec