当前位置：首页 > article >正文

C++ 并发编程之std::find的并发版本

article 2025/3/1 13:02:32

在C++中，std::find 是一个用于顺序查找容器中特定元素的算法。为了提高性能，我们可以设计并实现一个并行版本的 std::find，以便在多核处理器上并行执行查找操作。基本思想是将容器中的元素划分为若干块，每个块由一个单独的线程处理，并使用原子变量来确保只有一个线程返回找到的结果。

基本思想

任务划分：将容器中的元素划分成多个子范围（块），每个子范围由一个线程处理。任务划分的粒度可以根据容器的规模和处理器的核心数进行调整。
线程执行：每个线程独立处理分配给它的子范围，查找目标元素。如果某个线程找到了目标元素，它会将结果存储在一个原子变量中，并通知其他线程停止查找。
等待同步：在所有线程完成其任务后，主线程等待所有线程完成，然后返回找到的结果。

实现代码

我们可以使用 C++11 的 std::thread 和 std::atomic 来实现并行版本的 std::find。为了简化实现，我们可以使用 std::vector 来管理线程，并使用 std::atomic<bool> 来控制线程是否需要继续查找。

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

// 并行版本的 std::find
template<typename Iterator, typename T>
Iterator parallel_find(Iterator first, Iterator last, const T& value) {
    const unsigned long length = std::distance(first, last);

    // 如果没有元素，直接返回 last
    if (length == 0) {
        return last;
    }

    // 获取系统支持的并发线程数
    const unsigned long max_threads = std::thread::hardware_concurrency();
    const unsigned long num_threads = std::min(max_threads != 0 ? max_threads : 2, length);

    // 每个线程处理的元素数量
    const unsigned long block_size = length / num_threads;

    std::vector<std::thread> threads(num_threads - 1);
    std::atomic<bool> found(false);
    std::vector<Iterator> results(num_threads, last);

    // 启动线程
    for (unsigned long i = 0; i < num_threads - 1; ++i) {
        Iterator block_start = first + i * block_size;
        Iterator block_end = block_start + block_size;
        threads[i] = std::thread([block_start, block_end, &value, &found, &results, i]() {
            for (Iterator it = block_start; it != block_end && !found.load(); ++it) {
                if (*it == value) {
                    results[i] = it;
                    found.store(true);
                    return;
                }
            }
        });
    }

    // 主线程处理最后一个块
    Iterator block_start = first + (num_threads - 1) * block_size;
    for (Iterator it = block_start; it != last && !found.load(); ++it) {
        if (*it == value) {
            results[num_threads - 1] = it;
            found.store(true);
            break;
        }
    }

    // 等待所有线程完成
    std::for_each(threads.begin(), threads.end(), std::mem_fn(&std::thread::join));

    // 返回找到的结果
    for (auto result : results) {
        if (result != last) {
            return result;
        }
    }

    return last;
}

int main() {
    std::vector<int> v = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};

    // 使用并行版本的 std::find
    auto result = parallel_find(v.begin(), v.end(), 7);

    if (result != v.end()) {
        std::cout << "Found " << *result << " at position " << std::distance(v.begin(), result) << std::endl;
    } else {
        std::cout << "Not found" << std::endl;
    }

    return 0;
}

代码说明

任务划分：
- length 是容器中元素的总数。
- max_threads 是系统支持的并发线程数，num_threads 是我们实际使用的线程数（不超过元素数量）。
- block_size 是每个线程处理的元素数量。
线程执行：
- 我们创建了一个 std::vector<std::thread> 来存储所有线程。
- 每个线程处理一个子范围 [block_start, block_end)，并查找目标元素。如果在子范围内找到了目标元素，线程会将结果存储在 results 中，并设置 found 为 true，通知其他线程停止查找。
等待同步：
- 主线程通过 std::thread::join 等待所有子线程完成。
- 主线程遍历 results，返回第一个找到的结果。