测试_第22章_多线程.cpp

← 测试总览 · 目录

编译运行：clang++ -std=c++26 -Wall -Wextra -pthread 测试_第22章_多线程.cpp -o t && ./t（需先 cd 测试/）

// 第22章测试：多线程与并发
#include <print>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <future>
#include <vector>
#include <queue>
#include <numeric>

int failures = 0;
#define CHECK(expr)                                                         \
    do {                                                                    \
        if (!(expr)) {                                                      \
            ++failures;                                                     \
            std::println("FAIL 第{}行: {}", __LINE__, #expr);              \
        }                                                                   \
    } while (0)

// 22.3 mutex 版本计数
void test_mutex_counter() {
    constexpr int NTHREAD = 10, NINC = 10'000;
    int counter = 0;
    std::mutex mtx;
    std::vector<std::thread> ts;
    for (int t = 0; t < NTHREAD; ++t) {
        ts.emplace_back([&] {
            for (int i = 0; i < NINC; ++i) {
                std::lock_guard<std::mutex> lock(mtx);
                ++counter;
            }
        });
    }
    for (auto& t : ts) t.join();
    CHECK(counter == NTHREAD * NINC);       // 恒为 100000
}

// 22.4 atomic 版本计数
void test_atomic_counter() {
    constexpr int NTHREAD = 10, NINC = 10'000;
    std::atomic<int> counter{0};
    std::vector<std::thread> ts;
    for (int t = 0; t < NTHREAD; ++t) {
        ts.emplace_back([&] {
            for (int i = 0; i < NINC; ++i) counter.fetch_add(1);
        });
    }
    for (auto& t : ts) t.join();
    CHECK(counter.load() == NTHREAD * NINC);
}

// 22.5 生产者-消费者（条件变量）
void test_producer_consumer() {
    std::mutex mtx;
    std::condition_variable cv;
    std::queue<int> q;
    bool done = false;
    std::vector<int> consumed;

    std::thread consumer([&] {
        std::unique_lock<std::mutex> lock(mtx);
        while (true) {
            cv.wait(lock, [&] { return !q.empty() || done; });
            while (!q.empty()) { consumed.push_back(q.front()); q.pop(); }
            if (done) break;
        }
    });

    std::thread producer([&] {
        for (int i = 0; i < 20; ++i) {
            {
                std::lock_guard<std::mutex> lock(mtx);
                q.push(i);
            }
            cv.notify_one();
        }
        {
            std::lock_guard<std::mutex> lock(mtx);
            done = true;
        }
        cv.notify_all();
    });

    producer.join();
    consumer.join();
    CHECK(consumed.size() == 20);
    CHECK(consumed.back() == 19);
    CHECK(consumed.front() == 0);
}

// 22.6 async + 并行求和
long long parallel_sum(const std::vector<int>& v) {
    const int n = (int)v.size();
    if (n == 0) return 0;
    const int nthreads = 4;
    std::vector<std::future<long long>> futures;
    for (int t = 0; t < nthreads; ++t) {
        auto lo = v.begin() + (std::int64_t)t * n / nthreads;
        auto hi = v.begin() + (std::int64_t)(t + 1) * n / nthreads;
        futures.push_back(std::async(std::launch::async,
            [lo, hi] { return std::accumulate(lo, hi, 0LL); }));
    }
    long long total = 0;
    for (auto& fu : futures) total += fu.get();
    return total;
}

int main() {
    std::println("硬件并发线程数: {}", std::thread::hardware_concurrency());

    test_mutex_counter();
    test_atomic_counter();
    test_producer_consumer();

    // async 基本用法
    auto result = std::async(std::launch::async, [] {
        long long sum = 0;
        for (int i = 1; i <= 1'000'000; ++i) sum += i;
        return sum;
    });
    CHECK(result.get() == 500000500000LL);

    // 并行求和对比串行
    std::vector<int> data(1'000'000);
    std::iota(data.begin(), data.end(), 1);
    long long serial = std::accumulate(data.begin(), data.end(), 0LL);
    CHECK(parallel_sum(data) == serial);

    if (failures == 0) std::println("全部通过");
    return failures;
}