同时进行的任务 · 并行与并发库

现代计算机都配备了多核处理器，如何充分利用这些计算资源成为了程序性能优化的关键。C++11 引入了标准的线程库，让并发编程不再依赖平台特定的 API。

让我们探索 C++ 的并行与并发世界。

---

线程基础

创建和管理线程是并发编程的起点：

#include <iostream>
#include <thread>

void hello()
{
    std::cout << "Hello from thread!" << std::endl;
}

void greet(const std::string& name)
{
    std::cout << "Hello, " << name << "!" << std::endl;
}

int main()
{
    // 创建线程
    std::thread t1(hello);
    
    // 带参数的线程
    std::thread t2(greet, "World");
    
    // Lambda 线程
    std::thread t3([]() {
        std::cout << "Hello from lambda!" << std::endl;
    });
    
    // 等待线程完成
    t1.join();
    t2.join();
    t3.join();
    
    return 0;
}

分离线程

#include <thread>
#include <chrono>

void background_task()
{
    // 后台运行的任务
    std::this_thread::sleep_for(std::chrono::seconds(5));
    // ... 做一些工作 ...
}

int main()
{
    std::thread t(background_task);
    t.detach();  // 分离线程，让它在后台运行
    
    // 主线程继续执行，不等待 t
    // 注意：程序退出时分离的线程会被强制终止
    
    return 0;
}

---

互斥量与锁

多线程访问共享数据需要同步：

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

std::mutex mtx;
int counter = 0;

void increment(int times)
{
    for (int i = 0; i < times; ++i)
    {
        std::lock_guard<std::mutex> lock(mtx);  // RAII 风格的锁
        ++counter;
    }
}

int main()
{
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i)
    {
        threads.emplace_back(increment, 1000);
    }
    
    for (auto& t : threads)
    {
        t.join();
    }
    
    std::cout << "Counter: " << counter << std::endl;  // 10000
    
    return 0;
}

unique_lock：更灵活的锁

#include <mutex>

std::mutex mtx;

void flexible_locking()
{
    std::unique_lock<std::mutex> lock(mtx);
    
    // 可以手动解锁和重新加锁
    lock.unlock();
    // ... 不需要锁的操作 ...
    lock.lock();
    
    // 可以延迟加锁
    std::unique_lock<std::mutex> lock2(mtx, std::defer_lock);
    // ... 一些操作 ...
    lock2.lock();  // 现在加锁
}

避免死锁：同时锁定多个互斥量

#include <mutex>

std::mutex mtx1, mtx2;

void safe_operation()
{
    // std::lock 同时锁定多个互斥量，避免死锁
    std::lock(mtx1, mtx2);
    std::lock_guard<std::mutex> lg1(mtx1, std::adopt_lock);
    std::lock_guard<std::mutex> lg2(mtx2, std::adopt_lock);
    
    // C++17: scoped_lock 更简洁
    // std::scoped_lock lock(mtx1, mtx2);
}

---

条件变量

用于线程间的等待和通知：

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

std::mutex mtx;
std::condition_variable cv;
std::queue<int> dataQueue;
bool finished = false;

void producer()
{
    for (int i = 0; i < 10; ++i)
    {
        {
            std::lock_guard<std::mutex> lock(mtx);
            dataQueue.push(i);
            std::cout << "Produced: " << i << std::endl;
        }
        cv.notify_one();  // 通知消费者
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    
    {
        std::lock_guard<std::mutex> lock(mtx);
        finished = true;
    }
    cv.notify_all();
}

void consumer()
{
    while (true)
    {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, []() { return !dataQueue.empty() || finished; });
        
        while (!dataQueue.empty())
        {
            int value = dataQueue.front();
            dataQueue.pop();
            std::cout << "Consumed: " << value << std::endl;
        }
        
        if (finished && dataQueue.empty())
            break;
    }
}

int main()
{
    std::thread prod(producer);
    std::thread cons(consumer);
    
    prod.join();
    cons.join();
    
    return 0;
}

---

原子操作

对于简单的共享变量，原子操作比互斥量更高效：

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

std::atomic<int> counter{0};

void increment(int times)
{
    for (int i = 0; i < times; ++i)
    {
        ++counter;  // 原子操作，线程安全
    }
}

int main()
{
    std::vector<std::thread> threads;
    
    for (int i = 0; i < 10; ++i)
    {
        threads.emplace_back(increment, 10000);
    }
    
    for (auto& t : threads)
        t.join();
    
    std::cout << "Counter: " << counter << std::endl;  // 100000
    
    return 0;
}

原子标志与自旋锁

#include <atomic>

std::atomic_flag spinlock = ATOMIC_FLAG_INIT;

void lock()
{
    while (spinlock.test_and_set(std::memory_order_acquire))
    {
        // 自旋等待
    }
}

void unlock()
{
    spinlock.clear(std::memory_order_release);
}

---

future 与 promise：异步返回值

#include <iostream>
#include <future>
#include <thread>

int compute(int x)
{
    std::this_thread::sleep_for(std::chrono::seconds(1));
    return x * x;
}

int main()
{
    // 使用 async 启动异步任务
    std::future<int> result = std::async(std::launch::async, compute, 42);
    
    std::cout << "Computing..." << std::endl;
    
    // 获取结果（阻塞直到计算完成）
    std::cout << "Result: " << result.get() << std::endl;
    
    return 0;
}

promise：手动设置结果

#include <iostream>
#include <future>
#include <thread>

void calculate(std::promise<int>& prom, int x)
{
    try
    {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        prom.set_value(x * x);
    }
    catch (...)
    {
        prom.set_exception(std::current_exception());
    }
}

int main()
{
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();
    
    std::thread t(calculate, std::ref(prom), 10);
    
    std::cout << "Result: " << fut.get() << std::endl;
    
    t.join();
    return 0;
}

---

packaged_task：包装可调用对象

#include <iostream>
#include <future>
#include <thread>

int compute(int a, int b)
{
    return a + b;
}

int main()
{
    // 包装函数
    std::packaged_task<int(int, int)> task(compute);
    std::future<int> result = task.get_future();
    
    // 在另一个线程中执行
    std::thread t(std::move(task), 10, 20);
    
    std::cout << "Result: " << result.get() << std::endl;
    
    t.join();
    return 0;
}

---

并行算法（C++17）

C++17 为标准算法添加了并行执行策略：

#include <iostream>
#include <vector>
#include <algorithm>
#include <execution>
#include <numeric>
#include <chrono>

int main()
{
    std::vector<int> v(10000000);
    std::iota(v.begin(), v.end(), 1);
    
    // 顺序执行
    auto start = std::chrono::high_resolution_clock::now();
    auto sum1 = std::reduce(std::execution::seq, v.begin(), v.end(), 0LL);
    auto end = std::chrono::high_resolution_clock::now();
    auto seq_time = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
    
    // 并行执行
    start = std::chrono::high_resolution_clock::now();
    auto sum2 = std::reduce(std::execution::par, v.begin(), v.end(), 0LL);
    end = std::chrono::high_resolution_clock::now();
    auto par_time = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
    
    std::cout << "Sequential: " << seq_time.count() << " ms" << std::endl;
    std::cout << "Parallel:   " << par_time.count() << " ms" << std::endl;
    
    // 其他并行算法
    std::sort(std::execution::par, v.begin(), v.end());
    std::for_each(std::execution::par, v.begin(), v.end(), [](int& x) { x *= 2; });
    
    return 0;
}

执行策略：

• std::execution::seq - 顺序执行

• std::execution::par - 并行执行

• std::execution::par_unseq - 并行且可向量化

---

线程池实现

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

class ThreadPool
{
private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex mtx;
    std::condition_variable cv;
    bool stop = false;
    
public:
    ThreadPool(size_t numThreads)
    {
        for (size_t i = 0; i < numThreads; ++i)
        {
            workers.emplace_back([this]() {
                while (true)
                {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(mtx);
                        cv.wait(lock, [this]() { 
                            return stop || !tasks.empty(); 
                        });
                        
                        if (stop && tasks.empty())
                            return;
                        
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }
    
    template<typename F, typename... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>
    {
        using ReturnType = typename std::result_of<F(Args...)>::type;
        
        auto task = std::make_shared<std::packaged_task<ReturnType()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<ReturnType> result = task->get_future();
        
        {
            std::lock_guard<std::mutex> lock(mtx);
            tasks.emplace([task]() { (*task)(); });
        }
        
        cv.notify_one();
        return result;
    }
    
    ~ThreadPool()
    {
        {
            std::lock_guard<std::mutex> lock(mtx);
            stop = true;
        }
        cv.notify_all();
        
        for (auto& worker : workers)
            worker.join();
    }
};

int main()
{
    ThreadPool pool(4);
    
    std::vector<std::future<int>> results;
    
    for (int i = 0; i < 8; ++i)
    {
        results.emplace_back(pool.enqueue([i]() {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            return i * i;
        }));
    }
    
    for (auto& result : results)
    {
        std::cout << result.get() << " ";
    }
    std::cout << std::endl;
    
    return 0;
}

---

读写锁（C++14/17）

#include <shared_mutex>
#include <mutex>
#include <map>
#include <string>

class ThreadSafeCache
{
private:
    mutable std::shared_mutex mtx;
    std::map<std::string, int> cache;
    
public:
    // 读操作：允许多个线程同时读
    int get(const std::string& key) const
    {
        std::shared_lock<std::shared_mutex> lock(mtx);
        auto it = cache.find(key);
        return (it != cache.end()) ? it->second : -1;
    }
    
    // 写操作：独占访问
    void set(const std::string& key, int value)
    {
        std::unique_lock<std::shared_mutex> lock(mtx);
        cache[key] = value;
    }
};

---

线程局部存储

#include <iostream>
#include <thread>

thread_local int threadId = 0;

void setAndPrint(int id)
{
    threadId = id;
    std::cout << "Thread " << threadId << std::endl;
}

int main()
{
    std::thread t1(setAndPrint, 1);
    std::thread t2(setAndPrint, 2);
    std::thread t3(setAndPrint, 3);
    
    t1.join();
    t2.join();
    t3.join();
    
    return 0;
}

---

协程（C++20）

C++20 引入了协程，实现更自然的异步编程：

#include <coroutine>
#include <iostream>

// 简化的生成器
template<typename T>
struct Generator
{
    struct promise_type
    {
        T value;
        
        Generator get_return_object()
        {
            return Generator{std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        
        std::suspend_always initial_suspend() { return {}; }
        std::suspend_always final_suspend() noexcept { return {}; }
        std::suspend_always yield_value(T v) { value = v; return {}; }
        void return_void() {}
        void unhandled_exception() { std::terminate(); }
    };
    
    std::coroutine_handle<promise_type> handle;
    
    ~Generator() { if (handle) handle.destroy(); }
    
    bool next()
    {
        handle.resume();
        return !handle.done();
    }
    
    T value() const { return handle.promise().value; }
};

Generator<int> range(int start, int end)
{
    for (int i = start; i < end; ++i)
    {
        co_yield i;
    }
}

int main()
{
    auto gen = range(1, 10);
    while (gen.next())
    {
        std::cout << gen.value() << " ";
    }
    std::cout << std::endl;
    
    return 0;
}

---

总结

工具	用途
std::thread	创建和管理线程
std::mutex	互斥访问共享资源
std::lock_guard	RAII 风格的锁
std::condition_variable	线程间等待/通知
std::atomic	无锁的原子操作
std::future/promise	异步返回值
std::async	简单的异步任务
std::execution	并行算法执行策略
std::shared_mutex	读写锁
thread_local	线程局部存储
协程	轻量级异步

掌握这些工具，你就能写出高效的并发程序，充分发挥多核处理器的威力！