如何高效并行化实现Radix Sort？附待优化串行代码

问题背景 #

我正在学习并行编程，想了解如何最有效地并行化实现Radix Sort（基数排序）。我有一个思路：在执行基数排序前先按数字的位数对数组排序，再将分组后的数组送入radixSort算法，最后按位数合并排序后的数组，但我认为这并非最优方案。以下是我想要改造为并行版本的串行代码：

import time


def countingSort(array, place):
    size = len(array)
    output = [0] * size
    count = [0] * 10

    # Calculate count of elements
    for i in range(0, size):
        index = array[i] // place
        count[index % 10] += 1

    # Calculate cumulative count
    for i in range(1, 10):
        count[i] += count[i - 1]

    # Place the elements in sorted order
    i = size - 1
    while i >= 0:
        index = array[i] // place
        output[count[index % 10] - 1] = array[i]
        count[index % 10] -= 1
        i -= 1

    for i in range(0, size):
        array[i] = output[i]


# Main function to implement radix sort
def radixSort(array):
    # Get maximum element
    max_element = max(array)

    # Apply counting sort to sort elements based on place value.
    place = 1
    while max_element // place > 0:
        countingSort(array, place)
        place *= 10
    return array


start_time = time.time()
data = [23950, 55149, 6262, 80988, 61698, 46356, 44771, 86534, 17326, 45511, 3214, 90191, 15316, 18330, 60983, 70108, 37905, 94157]
data = radixSort(data)
end_time = time.time()
print("Time was", round(end_time - start_time, 3), "sekúnd.")
print(data)

并行化核心优化方案 #

1. 放弃预按位数分组的思路 #

你提到的按位数分组排序方案会额外增加分组、合并的开销，且不同位数的子数组大小不均，易导致负载失衡。更高效的方向是直接针对基数排序的核心——多轮计数排序——做并行优化，因为每轮计数排序针对独立数位，天然具备并行潜力。

2. 计数排序的并行拆解 #

计数排序分为三个阶段，其中前两个阶段可针对性并行化：

• 数位计数统计：遍历数组统计0-9的出现次数，这一步可将数组分片，由多线程/进程分别统计分片内的计数，最后合并所有分片结果。
• 前缀和计算：由于计数数组仅10个元素，串行计算的开销可忽略，无需强行并行。
• 逆序填充输出：为保证排序稳定性，逆序遍历是必要的，强行并行需引入原子操作避免冲突，反而得不偿失。更优方式是基于全局计数数组，给每个分片分配独立的输出偏移范围，分片内独立逆序填充。

3. Python并行实现示例 #

基于multiprocessing模块改造的并行基数排序代码，核心是并行化计数统计阶段：

import time
from multiprocessing import Pool, cpu_count

def count_chunk(chunk, place):
    """统计分片内每个数位的出现次数"""
    count = [0] * 10
    for num in chunk:
        index = num // place
        count[index % 10] += 1
    return count

def merge_counts(count_list):
    """合并多个分片的计数结果"""
    total_count = [0] * 10
    for count in count_list:
        for i in range(10):
            total_count[i] += count[i]
    return total_count

def countingSort_parallel(array, place):
    size = len(array)
    output = [0] * size
    cpu_num = cpu_count()
    
    # 数组分片，适配CPU核心数
    chunk_size = max(1, size // cpu_num)
    chunks = [array[i:i+chunk_size] for i in range(0, size, chunk_size)]
    
    # 并行统计分片计数
    with Pool(cpu_num) as pool:
        count_list = pool.starmap(count_chunk, [(chunk, place) for chunk in chunks])
    
    # 合并计数并计算前缀和
    count = merge_counts(count_list)
    for i in range(1, 10):
        count[i] += count[i - 1]
    
    # 逆序填充输出数组（保证稳定性）
    i = size - 1
    while i >= 0:
        index = array[i] // place
        output[count[index % 10] - 1] = array[i]
        count[index % 10] -= 1
        i -= 1
    
    # 结果写回原数组
    for i in range(size):
        array[i] = output[i]

def radixSort_parallel(array):
    max_element = max(array)
    place = 1
    while max_element // place > 0:
        countingSort_parallel(array, place)
        place *= 10
    return array

# 测试代码
if __name__ == "__main__":
    start_time = time.time()
    data = [23950, 55149, 6262, 80988, 61698, 46356, 44771, 86534, 17326, 45511, 3214, 90191, 15316, 18330, 60983, 70108, 37905, 94157]
    data = radixSort_parallel(data)
    end_time = time.time()
    print(f"耗时: {round(end_time - start_time, 3)} 秒")
    print(data)

4. 进一步优化建议 #

• 复用进程池：多次执行排序时，提前创建进程池避免重复创建销毁的开销。
• 规模判断分支：数组较小时，并行开销可能超过收益，可回退到串行实现。
• 语言适配优化：Python因GIL限制，CPU密集型任务用多进程更合适；C++等语言可直接用多线程提升效率。
• 输出分片填充：超大规模数组时，可提前计算各分片的输出偏移范围，让进程独立负责对应区域的填充，减少全局同步。

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