Windows下ProcessPoolExecutor优化：避免子进程加载冗余代码

核心结论 #

1. 与多进程无关的代码：优先放在if __name__ == '__main__':块内或单独文件中

   • Windows下的ProcessPoolExecutor基于multiprocessing实现，创建子进程时会导入主模块。如果把无关代码放在主模块顶层，每个子进程都会重复加载这些内容，浪费内存还可能引发意外错误（比如GUI组件重复初始化）。
   • 放进if __name__ == '__main__':是最直接的方案，这个代码块仅在主进程执行，子进程导入主模块时不会运行它。如果代码量较大，也可以拆分到单独的工具类/函数文件，在主进程需要时导入。

2. 多进程相关函数：单独存入文件是最优解

   • 你当前把多进程函数放到myMulti.py的做法完全正确。子进程只需要导入这个轻量文件，不会加载主进程里的GUI逻辑、无关类等，能显著降低子进程的启动开销和内存占用。
   • 注意：多进程函数所在文件不能包含仅主进程能运行的代码（比如GUI初始化、主进程专属全局变量），否则子进程导入时会出错。

对你当前代码的优化建议 #

• 主程序里的create_filepaths、update_chunks_to_gui等无关函数放在if __name__ == '__main__':块内的做法没问题，后续代码膨胀后可以移到单独工具文件（比如utils.py）。
• 主程序中import myClasses放在if __name__ == '__main__':内是正确的，避免子进程加载该模块。
• Windows路径建议用原始字符串（r'G:\...'）或双反斜杠，避免转义字符引发问题。

你的主程序代码参考 #

import myMulti

if __name__ == '__main__':
    import os
    from time import sleep, time
    import concurrent.futures
    import myClasses

    folder = r'G:\+++CODING+++\Thin-Plate-Spline-Motion-Model-Windows-main\assets\test28\\'

    resultlist = []
    start_time = time()

    def create_filepaths(folder):
        return sorted([os.path.join(folder, f) for f in os.scandir(folder)])


    def update_chunks_to_gui(future_object, local_last_chunk_index, processing_abort=False):
        if processing_abort:  # 如果用户切换目录，取消缩略图生成任务
            for image in future_object:
                image.cancel()
        else:
            for index, image in enumerate(future_object[local_last_chunk_index:]):
                if image.done():
                    resultlist.append((image.result()))  # 示例：将结果存入列表，实际用于更新GUI
                    local_last_chunk_index += 1
                else:
                    break
            return local_last_chunk_index


    def start_job(filepaths):
        print('...starting jobs')
        pool = concurrent.futures.ProcessPoolExecutor(max_workers=12)
        return [pool.submit(myMulti.process_file, file) for file in filepaths]

    def main():
        filepaths = create_filepaths(folder)
        last_chunk_index = 0

        future = start_job(filepaths)

        while True:  # GUI和主逻辑循环
            # 此处可添加GUI更新等其他逻辑
            print('+')
            sleep(0.5)
            if not last_chunk_index == len(future):
                last_chunk_index = update_chunks_to_gui(future, last_chunk_index)  # 若需取消任务，可传入processing_abort=True
            else:
                print("...all done in:", time()-start_time)
                quit()


    main()

你的多进程文件myMulti.py代码参考 #

import PIL.Image
import io
import base64


def convert_to_bytes(file_or_bytes, resize=None, fill_blanc_color=None):
    if isinstance(file_or_bytes, str):
        img = PIL.Image.open(file_or_bytes)
        if file_or_bytes.endswith(".jpg"):
            img.draft('RGB', resize if resize else None )
    else:
        try:
            img = PIL.Image.open(io.BytesIO(base64.b64decode(file_or_bytes)))
        except Exception as e:
            dataBytesIO = io.BytesIO(file_or_bytes)
            img = PIL.Image.open(dataBytesIO)
    if resize:
        img = img.resize((int(img.size[0]*min(resize[0]/img.size[1], resize[0]/img.size[0])),
                          int(img.size[1]*min(resize[1]/img.size[1], resize[0]/img.size[0]))),
                         PIL.Image.HAMMING).convert("RGBA")
    if fill_blanc_color:
        r, g, b = fill_blanc_color
        newimg = PIL.Image.new("RGB", resize if resize else img.size, (r, g, b))
        newimg.paste(img, (int((100-img.width)/2), int((100-img.height)/2)), img)
        img = newimg
    with io.BytesIO() as data:
        img.save(data, format="PNG")
        del img
        return data.getvalue()


def process_file(filepath):
    return convert_to_bytes(filepath, resize=(100, 100), fill_blanc_color=(90, 105, 121))

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