C++多线程归并排序慢于迭代版：问题排查与优化方案

原问题代码 #

#include <iostream>
#include <vector>
#include <thread>

void mergeSort(std::vector<int>& v, int left, int right) {
    if (left < right) {
        int mid = left + (right - left) / 2;

        std::thread t1(mergeSort, std::ref(v), left, mid);
        std::thread t2(mergeSort, std::ref(v), mid + 1, right);

        t1.join();
        t2.join();

        merge(v, left, mid, right);
    }
}

测试发现，处理100万规模数据时，该多线程版本的运行速度远慢于单线程迭代版归并排序，以下是问题分析与优化方案：

问题分析 #

• 线程创建与销毁开销过高：每一次递归拆分都创建两个新线程，线程的初始化、调度、销毁都有显著开销。当递归到深层，子数组规模极小时，线程开销完全覆盖了并行计算的收益，甚至拖慢整体速度。
• 过度并行引发CPU资源竞争：无限制创建线程会导致系统同时运行的线程数远超CPU核心数，频繁的上下文切换会消耗大量CPU资源，反而降低执行效率。
• 未设置任务粒度阈值：对于小规模的子数组，单线程排序（比如插入排序）的效率远高于多线程拆分处理，继续并行完全没有必要。

优化方案 #

1. 设置并行阈值，小规模子数组用单线程处理 #

当子数组的元素数量小于某个阈值（比如1000，可根据CPU性能调整）时，直接使用单线程归并排序或插入排序，避免在小数据上创建线程。

2. 使用线程池控制并发数 #

避免手动创建大量线程，改用线程池（或利用std::async的自动调度）来控制并发线程数，使其不超过CPU核心数，减少线程创建开销和上下文切换。

3. 优化merge函数的效率 #

确保merge过程高效，比如提前分配临时数组，避免在每次merge时重复内存分配；使用连续内存访问，利用CPU缓存优化。

优化后代码示例 #

带阈值的多线程归并排序 #

#include <iostream>
#include <vector>
#include <thread>
#include <algorithm>

const int THRESHOLD = 1000; // 可根据实际情况调整

void merge(std::vector<int>& v, int left, int mid, int right) {
    std::vector<int> temp(right - left + 1);
    int i = left, j = mid + 1, k = 0;
    while (i <= mid && j <= right) {
        temp[k++] = (v[i] <= v[j]) ? v[i++] : v[j++];
    }
    while (i <= mid) temp[k++] = v[i++];
    while (j <= right) temp[k++] = v[j++];
    std::copy(temp.begin(), temp.end(), v.begin() + left);
}

void mergeSortSingleThread(std::vector<int>& v, int left, int right) {
    if (left < right) {
        int mid = left + (right - left) / 2;
        mergeSortSingleThread(v, left, mid);
        mergeSortSingleThread(v, mid + 1, right);
        merge(v, left, mid, right);
    }
}

void mergeSort(std::vector<int>& v, int left, int right) {
    if (right - left + 1 <= THRESHOLD) {
        mergeSortSingleThread(v, left, right);
        return;
    }

    int mid = left + (right - left) / 2;

    std::thread t1(mergeSort, std::ref(v), left, mid);
    std::thread t2(mergeSort, std::ref(v), mid + 1, right);

    t1.join();
    t2.join();

    merge(v, left, mid, right);
}

利用std::async优化版本（自动控制并发） #

#include <iostream>
#include <vector>
#include <future>
#include <algorithm>

const int THRESHOLD = 1000;

void merge(std::vector<int>& v, int left, int mid, int right) {
    std::vector<int> temp(right - left + 1);
    int i = left, j = mid + 1, k = 0;
    while (i <= mid && j <= right) {
        temp[k++] = (v[i] <= v[j]) ? v[i++] : v[j++];
    }
    while (i <= mid) temp[k++] = v[i++];
    while (j <= right) temp[k++] = v[j++];
    std::copy(temp.begin(), temp.end(), v.begin() + left);
}

void mergeSortSingleThread(std::vector<int>& v, int left, int right) {
    if (left < right) {
        int mid = left + (right - left) / 2;
        mergeSortSingleThread(v, left, mid);
        mergeSortSingleThread(v, mid + 1, right);
        merge(v, left, mid, right);
    }
}

void mergeSort(std::vector<int>& v, int left, int right) {
    if (right - left + 1 <= THRESHOLD) {
        mergeSortSingleThread(v, left, right);
        return;
    }

    int mid = left + (right - left) / 2;

    auto future1 = std::async(std::launch::async, mergeSort, std::ref(v), left, mid);
    auto future2 = std::async(std::launch::async, mergeSort, std::ref(v), mid + 1, right);

    future1.get();
    future2.get();

    merge(v, left, mid, right);
}

内容的提问来源于stack exchange，提问作者Jeleriu Antonio Emanuel