Python并行基数排序（Parallel Radix Sort）代码优化与同步问题咨询

核心优化思路 #

不再为每一位单独创建与核心数等量的进程，而是复用固定数量的进程（建议与CPU核心数一致），通过任务分配机制控制总进程数，避免进程爆炸。

方案1：使用multiprocessing.Pool进程池 #

进程池会维护固定数量的工作进程，自动分配任务并复用进程，避免频繁创建/销毁进程的开销。具体实现步骤：

• 对每一位排序时，将待排序数字列表拆分为与进程池大小匹配的子块
• 每个进程负责处理一个子块的按位分组（将数字分到0-9的桶中）
• 主进程收集所有进程的分组结果，合并成有序列表后进入下一位排序

核心代码片段：

from multiprocessing import Pool

def partition_by_digit(numbers, digit_position):
    # 单个进程的分组任务：按指定位将数字分到对应桶中
    buckets = [[] for _ in range(10)]
    divisor = 10 ** digit_position
    for num in numbers:
        digit = (num // divisor) % 10
        buckets[digit].append(num)
    return buckets

def parallel_radix_sort(numbers, num_processes=8):
    if not numbers:
        return numbers
    max_digits = len(str(max(numbers)))
    
    with Pool(num_processes) as pool:
        for digit_pos in range(max_digits):
            # 将数字列表拆分为进程池大小的子块
            chunk_size = len(numbers) // num_processes
            chunks = [numbers[i:i+chunk_size] for i in range(0, len(numbers), chunk_size)]
            # 分配分组任务到进程池
            results = pool.starmap(partition_by_digit, [(chunk, digit_pos) for chunk in chunks])
            # 合并所有进程的桶结果，按0-9顺序拼接
            merged_buckets = [[] for _ in range(10)]
            for res in results:
                for i in range(10):
                    merged_buckets[i].extend(res[i])
            # 更新为当前位排序后的列表
            numbers = []
            for bucket in merged_buckets:
                numbers.extend(bucket)
    return numbers

方案2：队列+固定进程的生产者-消费者模型 #

如果需要更精细的任务控制，可创建固定数量的工作进程，通过multiprocessing.Queue传递任务与结果：

• 主进程按位发起排序任务，将待处理的数字块放入任务队列
• 固定数量的工作进程从队列取任务，完成分组后将结果放入结果队列
• 主进程收集所有结果，合并后进入下一位排序循环

关键注意事项：

• 每一位处理完成后，需确保任务队列清空，可通过向队列放入None作为进程终止信号（每轮结束后重新初始化队列）
• 优先使用JoinableQueue简化任务完成后的同步逻辑

额外优化建议 #

• 控制进程数量：进程数不要超过CPU核心数的1-2倍，过多进程会导致上下文切换开销剧增，反而降低效率
• 减少数据传输：尽量在进程内完成分组逻辑，仅传递分好的桶结果，避免大列表在进程间的频繁序列化/反序列化
• 小数据量降级串行：当待排序数字数量较小时，并行的开销可能超过收益，可自动切换为串行基数排序

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