如何对滑动窗口切图的循环做并行化提速且保证子图顺序不变？

一、原有代码性能瓶颈分析 #

• 单进程执行受Python GIL限制，仅能使用1个CPU核心，无法利用多核服务器资源
• 子图处理循环内频繁调用gc.collect()触发全量垃圾回收，是耗时极高的核心原因
• 存在不必要的numpy数组拷贝操作，增加额外开销
• 原代码末尾存在笔误：image.append(class_dictionary[cl]) 应为 image_label.append(class_dictionary[cl])

二、优化实现方案 #

2.1 基础逻辑优化 #

首先优化sliding_window函数，移除冗余操作：

import numpy as np
from tensorflow.keras.preprocessing import image
from tensorflow.keras.applications import preprocess_input # 按实际使用的模型导入对应预处理方法
import concurrent.futures
import gc

def sliding_window(image, stride, imgSize):
    height, width, _ = image.shape
    img = []
    a1 = list(range(0, height-imgSize+stride, stride))
    a2 = list(range(0, width-imgSize+stride, stride))
    if a1[-1] + imgSize != height:
        a1[-1] = height - imgSize
    if a2[-1] + imgSize != width:
        a2[-1] = width - imgSize
    for y in a1:
        for x in a2:
            # 移除冗余的np.array转换，切片本身已是numpy数组
            im1 = image[y:y+imgSize, x:x+imgSize, :]
            img.append(im1)
    return img

2.2 多进程并行实现 #

通过任务顺序映射保证输出结果完全和原串行逻辑一致，不会打乱子图序列：

# 全局参数，按实际业务值修改
SIZE = 128
STRIDE = 32 # 示例步长
img_path = "./你的图像根目录/"
class_dictionary = {} # 你的类别映射字典
file_images = [] # 你的类别列表
data = {} # 每个类别对应的图像文件名列表

# 单图处理函数：多进程独立任务单元
def process_single_image(task):
    task_idx, cl, img_file = task
    path = f"{img_path}{cl}/{img_file}"
    im = image.load_img(path)
    im = image.img_to_array(im)
    sub_imgs = []
    sub_labels = []
    if im.shape[0] >= SIZE and im.shape[1] >= SIZE:
        windows = sliding_window(im, STRIDE, SIZE)
        for win in windows:
            if win.shape[2] >=3:
                processed_win = preprocess_input(win)
                sub_imgs.append(processed_win)
                sub_labels.append(class_dictionary[cl])
    # 返回任务索引、子图列表、对应标签列表，用于后续按顺序拼接
    return (task_idx, sub_imgs, sub_labels)

if __name__ == "__main__":
    # 预先生成所有待处理任务，分配全局唯一索引，保证后续顺序正确
    tasks = []
    task_idx = 0
    for cl in file_images:
        for img_file in data[cl]:
            tasks.append( (task_idx, cl, img_file) )
            task_idx += 1

    image_data = []
    image_label = []
    total = len(tasks)
    processed_count = 0

    # max_workers设为服务器物理CPU核心数，不要超过物理核心数避免性能下降
    with concurrent.futures.ProcessPoolExecutor(max_workers=16) as executor:
        # 按任务提交顺序返回结果，完全保证顺序和原串行逻辑一致
        for result in executor.map(process_single_image, tasks):
            idx, sub_imgs, sub_labels = result
            image_data.extend(sub_imgs)
            image_label.extend(sub_labels)
            processed_count += 1
            if processed_count % 500 == 0:
                print(f"{processed_count}/{total} images processed...")

    # 所有处理完成后统一触发一次垃圾回收即可
    gc.collect()

三、效果说明 #

• 顺序保证：executor.map会严格按照任务提交的顺序返回结果，单图内部的子图顺序也和原逻辑完全一致，不会出现乱序问题
• 性能提升：16核CPU下相比原单进程代码，运行速度可提升10_{15倍，3000张图的处理时间可压缩到2}3小时范围内，实际耗时和步长、原图大小相关
• 内存优化：子图处理完直接返回拼接，无冗余内存占用，内存紧张时可将任务拆分为多个批次处理，每批处理完成后清理内存即可

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