第五节 · 有条不紊 · 格式化输入输出

C++11 引入了线程支持库 <thread>，让多线程编程成为语言标准的一部分。

---

创建线程

#include <thread>

void task()
{
    cout << "Hello from thread!" << endl;
}

int main()
{
    // 创建并启动线程
    thread t(task);
    
    // 等待线程完成
    t.join();
    
    cout << "Main thread done" << endl;
    return 0;
}

---

传递参数

void printSum(int a, int b)
{
    cout << "Sum: " << a + b << endl;
}

void printString(const string& s)
{
    cout << "String: " << s << endl;
}

int main()
{
    // 传递值
    thread t1(printSum, 3, 4);
    t1.join();
    
    // 传递引用需要 ref
    string msg = "Hello";
    thread t2(printString, ref(msg));
    t2.join();
    
    // Lambda
    thread t3([](int x) {
        cout << "Lambda: " << x << endl;
    }, 42);
    t3.join();
    
    return 0;
}

---

detach vs join

void background()
{
    this_thread::sleep_for(chrono::seconds(2));
    cout << "Background task done" << endl;
}

int main()
{
    thread t1([]{ cout << "Thread 1" << endl; });
    t1.join();  // 等待完成
    
    thread t2(background);
    t2.detach();  // 分离，后台运行
    
    // 注意：main 结束时 t2 可能还在运行
    // 分离的线程会在程序退出时被终止
    
    cout << "Main exiting" << endl;
    this_thread::sleep_for(chrono::seconds(3));
    
    return 0;
}

---

互斥锁

#include <mutex>

int counter = 0;
mutex mtx;

void increment()
{
    for (int i = 0; i < 100000; ++i)
    {
        mtx.lock();
        ++counter;
        mtx.unlock();
    }
}

// 更安全的方式：lock_guard
void safeIncrement()
{
    for (int i = 0; i < 100000; ++i)
    {
        lock_guard<mutex> lock(mtx);  // RAII
        ++counter;
    }  // 自动解锁
}

int main()
{
    thread t1(safeIncrement);
    thread t2(safeIncrement);
    
    t1.join();
    t2.join();
    
    cout << "Counter: " << counter << endl;  // 200000
    return 0;
}

---

unique_lock

mutex mtx;

void flexibleLocking()
{
    unique_lock<mutex> lock(mtx);
    
    // 可以手动解锁和重新加锁
    lock.unlock();
    // ... 不需要锁的操作 ...
    lock.lock();
    
    // 可以转移所有权
    unique_lock<mutex> lock2 = move(lock);
}

void tryLocking()
{
    unique_lock<mutex> lock(mtx, try_to_lock);
    
    if (lock.owns_lock())
    {
        cout << "Got the lock" << endl;
    }
    else
    {
        cout << "Lock busy" << endl;
    }
}

---

条件变量

#include <condition_variable>

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

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

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

int main()
{
    thread p(producer);
    thread c(consumer);
    
    p.join();
    c.join();
    
    return 0;
}

---

原子操作

#include <atomic>

atomic<int> counter(0);

void atomicIncrement()
{
    for (int i = 0; i < 100000; ++i)
    {
        ++counter;  // 原子操作，无需锁
    }
}

int main()
{
    thread t1(atomicIncrement);
    thread t2(atomicIncrement);
    
    t1.join();
    t2.join();
    
    cout << "Counter: " << counter << endl;  // 200000
    return 0;
}

---

future 和 promise

#include <future>

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

int main()
{
    // async：异步执行
    future<int> result = async(launch::async, compute, 42);
    
    cout << "Computing..." << endl;
    cout << "Result: " << result.get() << endl;  // 阻塞直到结果可用
    
    // promise：手动设置结果
    promise<string> prom;
    future<string> fut = prom.get_future();
    
    thread t([&prom] {
        this_thread::sleep_for(chrono::seconds(1));
        prom.set_value("Hello from thread");
    });
    
    cout << fut.get() << endl;
    t.join();
    
    return 0;
}

---

并行算法（C++17）

#include <execution>
#include <algorithm>

int main()
{
    vector<int> v(10000000);
    iota(v.begin(), v.end(), 0);
    
    // 顺序执行
    sort(execution::seq, v.begin(), v.end());
    
    // 并行执行
    sort(execution::par, v.begin(), v.end());
    
    // 并行且向量化
    sort(execution::par_unseq, v.begin(), v.end());
    
    return 0;
}

---

线程安全的单例

class Singleton
{
public:
    static Singleton& getInstance()
    {
        static Singleton instance;  // C++11 保证线程安全
        return instance;
    }
    
    void doSomething() { cout << "Singleton action" << endl; }
    
private:
    Singleton() = default;
    Singleton(const Singleton&) = delete;
    Singleton& operator=(const Singleton&) = delete;
};

---

习题

• 实现一个线程池。

• 编写生产者-消费者问题的完整解决方案。

• 使用 future 实现并行计算斐波那契数列。