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

C++并发编程之线程间数据划分的原则与方法

article 2025/1/16 11:50:10

在设计并发代码时，合理划分数据是确保多线程程序高效运行的关键。数据划分的目标是尽量减少线程间的竞争和同步开销，同时最大化并行度。数据划分可以在线程启动前（静态划分）和线程运行后（动态划分）进行。以下是详细的原则、方法和示例：

1. 数据划分的原则

在设计并发代码时，数据划分应遵循以下原则：

1.1 最小化共享数据

尽量减少线程间共享的数据量，避免线程竞争和锁的开销。
如果必须共享数据，确保共享数据的访问是线程安全的。

1.2 最大化局部性

尽量让每个线程处理的数据是局部的，减少线程间通信和缓存一致性问题。
局部性包括时间局部性（频繁访问的数据）和空间局部性（相邻的数据）。

1.3 负载均衡

确保每个线程的工作量大致相同，避免某些线程过载而其他线程空闲。
负载均衡可以通过静态划分或动态划分实现。

1.4 可扩展性

数据划分应支持动态调整，以适应不同的硬件资源（如CPU核心数）和任务规模。

1.5 简单性和可维护性

数据划分方案应尽量简单，避免过度复杂的设计，以降低代码的维护成本。

2. 数据划分的方法

数据划分可以分为静态划分和动态划分两种方式。

2.1 静态划分

静态划分是在线程启动前将数据划分为固定大小的块，每个线程处理一个或多个块。静态划分适用于数据规模已知且任务均匀分布的场景。

方法：

按数据块划分：将数据划分为固定大小的块，每个线程处理一个块。
按任务划分：将任务划分为独立的子任务，每个线程处理一个子任务。

示例：
假设有一个数组 int data[1000]，需要计算每个元素的平方。可以将数组划分为 4 个块，每个线程处理 250 个元素。

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

void process_data(int* data, int start, int end) {
    for (int i = start; i < end; ++i) {
        data[i] = data[i] * data[i];
    }
}

int main() {
    const int data_size = 1000;
    const int num_threads = 4;
    int data[data_size];

    // 初始化数据
    for (int i = 0; i < data_size; ++i) {
        data[i] = i;
    }

    std::vector<std::thread> threads;
    int chunk_size = data_size / num_threads;

    // 启动线程
    for (int i = 0; i < num_threads; ++i) {
        int start = i * chunk_size;
        int end = (i == num_threads - 1) ? data_size : start + chunk_size;
        threads.emplace_back(process_data, data, start, end);
    }

    // 等待线程完成
    for (auto& t : threads) {
        t.join();
    }

    // 输出结果
    for (int i = 0; i < data_size; ++i) {
        std::cout << data[i] << " ";
    }

    return 0;
}

优点：

实现简单，适合任务均匀分布的场景。
线程间无需同步，开销小。

缺点：

如果任务分布不均匀，可能导致负载不均衡。
数据规模变化时，需要重新划分。

2.2 动态划分

动态划分是在线程运行后根据实际情况动态分配任务。动态划分适用于任务分布不均匀或数据规模未知的场景。

方法：

任务队列：将任务放入一个共享队列中，线程从队列中动态获取任务。
工作窃取（Work Stealing）：每个线程维护一个本地任务队列，当某个线程完成自己的任务后，可以从其他线程的队列中“窃取”任务。

示例：
使用任务队列动态分配任务。

#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <mutex>
#include <condition_variable>

std::queue<int> task_queue;
std::mutex queue_mutex;
std::condition_variable queue_cv;

void worker_thread() {
    while (true) {
        std::unique_lock<std::mutex> lock(queue_mutex);
        queue_cv.wait(lock, [] { return !task_queue.empty(); });

        int task = task_queue.front();
        task_queue.pop();
        lock.unlock();

        if (task == -1) break; // 结束标志

        // 处理任务
        std::cout << "Thread " << std::this_thread::get_id() << " processing task: " << task << std::endl;
    }
}

int main() {
    const int num_threads = 4;
    std::vector<std::thread> threads;

    // 启动线程
    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(worker_thread);
    }

    // 添加任务
    for (int i = 0; i < 20; ++i) {
        std::unique_lock<std::mutex> lock(queue_mutex);
        task_queue.push(i);
        queue_cv.notify_one();
    }

    // 添加结束标志
    for (int i = 0; i < num_threads; ++i) {
        std::unique_lock<std::mutex> lock(queue_mutex);
        task_queue.push(-1);
        queue_cv.notify_one();
    }

    // 等待线程完成
    for (auto& t : threads) {
        t.join();
    }

    return 0;
}

优点：