C++20 线程协调类：从入门到精通

发布于：2025-03-20 ⋅ 阅读:(18) ⋅ 点赞:(0)

文章目录


在多线程编程中，线程之间的协调是一个关键问题。C++20 引入了三种新的同步原语： std::barrier、 std::latch 和 std::counting_semaphore，它们极大地简化了线程间的同步操作。本文将从入门到精通，逐步深入地介绍这三种同步原语的使用方法和适用场景。

1. 初识线程协调

在多线程编程中，线程协调是指控制多个线程的执行顺序，确保它们在特定的点上同步或互斥。常见的线程协调问题包括：

  • 同步执行：多个线程需要在某个点上同步，然后一起继续执行。
  • 资源限制：限制同时访问某个资源的线程数量。
  • 任务完成：等待多个线程完成任务后继续执行。

C++20 引入的三种同步原语正是为了解决这些问题而设计的。

2. std::barrier：多线程同步的屏障

std::barrier 是一种用于多线程同步的机制，它允许一组线程在某个点（称为屏障点）上同步。当线程到达屏障点时，它会被阻塞，直到所有线程都到达该点，然后所有线程同时继续执行。

2.1 核心函数

  • arrive_and_wait()：线程到达屏障点并等待，直到所有线程都到达屏障点。
  • arrive_and_drop()：线程到达屏障点，减少一个期待线程数，并重置屏障。

2.2 示例代码

以下代码展示了如何使用 std::barrier 来同步多个线程：

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

void worker(std::barrier& barrier, int id) {
    std::cout << "Thread " << id << " is working.\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000 * id));
    std::cout << "Thread " << id << " reached the barrier.\n";
    barrier.arrive_and_wait();
}

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

    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(worker, std::ref(barrier), i + 1);
    }

    for (auto& t : threads) {
        t.join();
    }

    std::cout << "All threads have finished.\n";
    return 0;
}

2.3 高级用法

std::barrier 支持一个可选的回调函数，当所有线程到达屏障点时，回调函数会被自动调用。这可以用于执行一些同步操作，例如更新状态或清理资源。

std::barrier barrier(num_threads, [](std::size_t) {
    std::cout << "All threads have reached the barrier. Executing callback.\n";
});

2.4 适用场景

  • 分阶段任务：适用于多线程分阶段任务，每个阶段结束后同步。
  • 并行算法：例如矩阵乘法等多阶段并行算法。

3. std::latch：一次性同步原语

std::latch 是一种一次性同步原语，用于确保一组线程在某个条件满足后继续执行。它类似于 std::counting_semaphore，但只能使用一次。

3.1 核心函数

  • count_down()：减少计数器的值。
  • wait()：阻塞线程，直到计数器减少到零。

3.2 示例代码

以下代码展示了如何使用 std::latch 等待多个线程完成任务：

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

void worker(std::latch& latch, int id) {
    std::cout << "Thread " << id << " is working.\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000 * id));
    std::cout << "Thread " << id << " finished. Counting down latch.\n";
    latch.count_down();
}

int main() {
    const int num_threads = 3;
    std::latch latch(num_threads);

    std::vector<std::thread> threads;
    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(worker, std::ref(latch), i + 1);
    }

    latch.wait();
    std::cout << "All threads have finished.\n";

    for (auto& t : threads) {
        t.join();
    }
    return 0;
}

3.3 高级用法

std::latch 的计数器可以初始化为任意值，但一旦计数器减少到零，后续的 wait() 调用将立即返回，而不会阻塞。

3.4 适用场景

  • 一次性同步：适用于一次性同步，确保所有线程完成某个任务后继续。
  • 资源初始化：例如初始化资源或等待所有任务完成。

4. std::counting_semaphore：可重用的同步原语

std::counting_semaphore 是一种计数信号量，用于控制对共享资源的访问。它类似于 std::latch，但可以多次使用。

4.1 核心函数

  • acquire()：减少信号量的计数，阻塞直到计数大于零。
  • release()：增加信号量的计数。

4.2 示例代码

以下代码展示了如何使用 std::counting_semaphore 控制线程的执行：

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

void worker(std::counting_semaphore<>& sem, int id) {
    std::cout << "Thread " << id << " is waiting.\n";
    sem.acquire();
    std::cout << "Thread " << id << " is working.\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(1000 * id));
    std::cout << "Thread " << id << " finished.\n";
}

int main() {
    const int num_threads = 3;
    std::counting_semaphore<> sem(2); // 允许同时运行两个线程

    std::vector<std::thread> threads;
    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(worker, std::ref(sem), i + 1);
    }

    for (auto& t : threads) {
        t.join();
    }
    return 0;
}

4.3 高级用法

std::counting_semaphore 的计数器可以初始化为任意值，且可以通过 try_acquire() 尝试非阻塞地获取信号量。

4.4 适用场景

  • 资源限制：适用于控制对共享资源的访问。
  • 线程池：例如限制同时运行的线程数量。

5. 性能与优化

虽然 std::barrier、std::latch 和 std::counting_semaphore 提供了强大的同步功能，但过度使用同步原语可能会导致性能问题。以下是一些优化建议：

  1. 减少同步点：尽量减少线程同步的次数，避免不必要的阻塞。
  2. 使用局部变量：尽量使用线程局部变量（thread_local），减少线程间的竞争。
  3. 避免死锁：确保线程同步的顺序一致，避免死锁。
  4. 使用无锁编程：在可能的情况下，使用无锁编程技术，减少同步开销。

6. 实际应用案例

6.1 并行矩阵乘法

以下代码展示了如何使用 std::barrier 实现并行矩阵乘法：

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

void multiply(const std::vector<std::vector<int>>& A, const std::vector<std::vector<int>>& B, std::vector<std::vector<int>>& result, int row, int col, std::barrier& barrier) {
    for (size_t i = 0; i < A.size(); ++i) {
        result[row][col] += A[row][i] * B[i][col];
    }
    barrier.arrive_and_wait();
}

int main() {
    const int size = 4;
    std::vector<std::vector<int>> A(size, std::vector<int>(size, 1));
    std::vector<std::vector<int>> B(size, std::vector<int>(size, 2));
    std::vector<std::vector<int>> result(size, std::vector<int>(size, 0));

    std::barrier barrier(size * size);
    std::vector<std::thread> threads;

    for (size_t i = 0; i < size; ++i) {
        for (size_t j = 0; j < size; ++j) {
            threads.emplace_back(multiply, std::ref(A), std::ref(B), std::ref(result), i, j, std::ref(barrier));
        }
    }

    for (auto& t : threads) {
        t.join();
    }

    for (const auto& row : result) {
        for (const auto& val : row) {
            std::cout << val << " ";
        }
        std::cout << "\n";
    }

    return 0;
}

6.2 线程池

以下代码展示了如何使用 std::counting_semaphore 实现一个简单的线程池：

#include <iostream>
#include <thread>
#include <semaphore>
#include <queue>
#include <functional>
#include <vector>

class ThreadPool {
public:
    ThreadPool(size_t num_threads) : semaphore(num_threads) {
        for (size_t i = 0; i < num_threads; ++i) {
            threads.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex);
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) {
                            return;
                        }
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for (auto& t : threads) {
            t.join();
        }
    }

    template <class F, class... Args>
    auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> {
        using return_type = typename std::result_of<F(Args...)>::type;
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );

        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if (stop) {
                throw std::runtime_error("enqueue on stopped ThreadPool");
            }
            tasks.emplace([task]() { (*task)(); });
        }
        condition.notify_one();
        return res;
    }

private:
    std::vector<std::thread> threads;
    std::queue<std::function<void()>> tasks;
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop = false;
    std::counting_semaphore<> semaphore;
};

int main() {
    ThreadPool pool(4);

    auto result1 = pool.enqueue([] { return 42; });
    auto result2 = pool.enqueue([] { return 43; });

    std::cout << "Result 1: " << result1.get() << "\n";
    std::cout << "Result 2: " << result2.get() << "\n";

    return 0;
}

7. 总结

C++20 引入的 std::barrier、std::latch 和 std::counting_semaphore 提供了强大的线程协调机制，简化了多线程编程中的同步操作。它们各具特色，适用于不同的场景：

  • std::barrier：适用于多线程分阶段任务，每个阶段结束后同步。
  • std::latch：适用于一次性同步，确保所有线程完成某个任务后继续。
  • std::counting_semaphore：适用于控制对共享资源的访问。

通过合理使用这些同步原语，可以显著提高代码的可读性和性能，同时减少死锁和竞态条件的可能性。

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布