13.C++多线程

13.多线程

目录介绍

• 13.1 线程管理
  • 13.1.1 线程介绍
  • 13.1.2 创建线程
  • 13.1.3 等待线程结束
  • 13.1.4 分离线程
  • 13.1.5 检查线程
  • 13.1.6 线程休眠
  • 13.1.9 线程管理总结
• 13.2 多线程
  • 13.2.1 多线程头文件
  • 13.2.2 创建多线程
  • 13.2.3 多线程数据竞争
  • 13.2.4 死锁
  • 13.2.5 虚假唤醒
• 13.3 同步机制
  • 13.3.1 mutex
  • 13.3.2 condition_variable
  • 13.3.3 atomic
• 13.4 异步机制
  • 13.4.1 异步操作
• 13.5 线程局部存储
  • 13.5.1 thread_local
• 13.6 延迟任务
  • 13.6.1 sleep_for
  • 13.6.2 std::async
  • 13.6.3 使用定时器
  • 13.6.4 std::thread
• 13.7 线程池
  • 13.7.1 简易线程池
• 13.8 综合案例
  • 13.8.1
  • 13.8.4 生产者-消费者

13.1 线程

13.1.1 线程介绍

在 C++ 中，线程 是并发执行的基本单位。C++11 引入了标准库支持多线程编程，提供了 std::thread 类来创建和管理线程。

std::thread 在 <thread> 头文件中声明，因此使用 std::thread 需包含 <thread> 头文件。

13.1.2 创建线程

使用 std::thread 创建线程，需要传入一个可调用对象（如函数、Lambda 表达式、函数对象等）。

13.1.2.1 使用函数创建线程

#include <iostream>
#include <thread>

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

int main() {
    std::thread t(threadFunction); // 创建线程
    t.join(); // 等待线程结束
    std::cout << "Main thread finished." << std::endl;
    return 0;
}

13.1.2.2 Lambda表达式创建线程

#include <iostream>
#include <thread>

int main() {
    std::thread t([]() {
        std::cout << "Hello from Lambda thread!" << std::endl;
    });
    t.join();
    std::cout << "Main thread finished." << std::endl;
    return 0;
}

13.1.2.3 传递参数给线程函数

值传递：直接传递

void func(int a, const std::string& s);
std::thread t(func, 42, "Hello");

引用传递：用 std::ref

void modify(int& x);
int value = 0;
std::thread t(modify, std::ref(value)); // 传递引用

然后看一个具体的案例，如下所示：

#include <iostream>
#include <thread>

void printMessage(const std::string& message) {
    std::cout << message << std::endl;
}

int main() {
    std::thread t(printMessage, "Hello from thread with arguments!");
    t.join();
    return 0;
}

13.1.3 等待线程结束

join() 的作用

1. 等待线程完成：调用 join() 会阻塞当前线程，直到目标线程执行完毕。
2. 释放线程资源：线程执行完毕后，join() 会释放线程占用的资源。
3. 确保线程安全：如果不调用 join() 或 detach()，std::thread 对象析构时会调用 std::terminate，导致程序崩溃。

使用 join() 等待线程执行完毕。

std::thread t(threadFunction);
t.join(); // 等待线程结束

13.1.4 分离线程

detach() 的作用

1. 分离线程：将线程与 std::thread 对象分离，线程在后台独立运行。
2. 释放 std::thread 对象：分离后，std::thread 对象不再管理该线程，可以安全销毁。
3. 避免资源泄漏：分离线程后，线程的资源会在其执行完毕后由操作系统自动回收。

使用 detach() 将线程与主线程分离，线程在后台独立运行。

std::thread t(threadFunction);
t.detach(); // 分离线程

join() 与 detach() 的区别

13.1.5 检查线程

使用 joinable() 检查线程是否可以被 join() 或 detach()。

std::thread t(threadFunction);
if (t.joinable()) {
    t.join();
}

13.1.6 线程休眠

• sleep_until: 线程休眠至某个指定的时刻(time point)，该线程才被重新唤醒。
• sleep_for: 线程休眠某个指定的时间片(time span)，该线程才被重新唤醒，不过由于线程调度等原因，实际休眠时间可能比 sleep_duration 所表示的时间片更长。

13.1.9 线程管理总结

13.2 多线程

13.2.1 多线程头文件

C++11 新标准中引入了五个头文件来支持多线程编程，它们分别是 <atomic>, <thread>, <mutex>, <condition_variable> 和 <future>。

• <atomic>：该头文主要声明了两个类, std::atomic 和 std::atomic_flag，另外还声明了一套 C 风格的原子类型和与 C 兼容的原子操作的函数。
• <thread>：该头文件主要声明了 std::thread 类，另外 std::this_thread 命名空间也在该头文件中。
• <mutex>：该头文件主要声明了与互斥量(Mutex)相关的类，包括 std::mutex_* 一系列类，std::lock_guard, std::unique_lock, 以及其他的类型和函数。
• <condition_variable>：该头文件主要声明了与条件变量相关的类，包括 std::condition_variable 和 std::condition_variable_any。
• <future>：该头文件主要声明了 std::promise, std::package_task 两个 Provider 类，以及 std::future 和 std::shared_future 两个 Future 类，另外还有一些与之相关的类型和函数，std::async() 函数就声明在此头文件中。

13.2.2 创建多线程

使用 std::thread 创建多个线程，每个线程执行不同的任务。

1.基本示例

#include <iostream>
#include <thread>

void task(int id) {
    std::cout << "Thread " << id << " is running." << std::endl;
}

int main() {
    std::thread t1(task, 1);
    std::thread t2(task, 2);

    t1.join(); // 等待线程 t1 结束
    t2.join(); // 等待线程 t2 结束

    std::cout << "Main thread finished." << std::endl;
    return 0;
}

2.使用 Lambda 表达式

#include <iostream>
#include <thread>

int main() {
    std::thread t1([]() {
        std::cout << "Thread 1 is running." << std::endl;
    });

    std::thread t2([]() {
        std::cout << "Thread 2 is running." << std::endl;
    });

    t1.join();
    t2.join();

    std::cout << "Main thread finished." << std::endl;
    return 0;
}

13.2.3 多线程数据竞争

多个线程同时访问共享资源，导致未定义行为。解决方法：使用互斥锁（std::mutex）保护共享资源。

13.2.4 死锁

多个线程互相等待对方释放锁，导致程序无法继续执行。解决方法：

• 按固定顺序加锁。
• 使用 std::lock() 同时锁定多个互斥锁。

13.2.5 虚假唤醒

线程在条件变量上被唤醒，但条件并未满足。解决方法：在 wait() 中使用谓词检查条件。

13.3 同步机制

多线程访问共享资源时，需要使用同步机制避免数据竞争。

13.3.1 mutex

使用 std::mutex 保护共享资源

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

std::mutex mtx;
int sharedData = 0;

void increment() {
    for (int i = 0; i < 1000; ++i) {
        std::lock_guard<std::mutex> lock(mtx); // 自动加锁和解锁
        // 或手动管理: mtx.lock(); ... mtx.unlock();
        ++sharedData;
    }
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);

    t1.join();
    t2.join();

    std::cout << "Shared Data: " << sharedData << std::endl; // 输出: Shared Data: 2000
    return 0;
}

13.3.2 condition_variable

std::condition_variable 是 C++ 标准库中用于线程同步的工具，通常与 std::mutex 结合使用，用于实现线程间的等待和通知机制。

它允许一个或多个线程等待某个条件成立，而另一个线程在条件满足时通知等待的线程。

基本概念

1. 条件变量（Condition Variable）：
   • 用于线程间的同步，允许线程等待某个条件成立。
   • 通常与互斥锁（std::mutex）一起使用，以确保对共享资源的线程安全访问。
2. 主要操作：
   • wait()：线程等待条件成立。
   • notify_one()：通知一个等待的线程。
   • notify_all()：通知所有等待的线程。
3. 典型使用场景： 生产者-消费者模型。 线程池任务调度。事件驱动编程。

使用 std::condition_variable 实现线程间通信

std::mutex mtx;
std::condition_variable halCv;
bool isHalReady = false;

void delayed_task() {
    // 模拟任务准备（可以是复杂计算或IO操作）
    std::this_thread::sleep_for(std::chrono::milliseconds(3000)); {
        std::lock_guard<std::mutex> lock(mtx);
        std::cout << ">>> 任务准备完成 <<<" << std::endl;
        //如果 ready提前变为 true（即任务准备完成），则立即唤醒
        isHalReady = true;
    }
    halCv.notify_one();  // 通知等待中的线程
}

int main() {
    // 启动延迟任务的线程
    std::thread worker(delayed_task); {
        std::unique_lock<std::mutex> lock(mtx);
        // 主线程继续执行其他工作
        std::cout << "主线程执行其他任务..." << std::endl;
        // 等待 2 秒或被提前唤醒
        if (halCv.wait_for(lock, std::chrono::seconds(2), []() -> bool { return isHalReady; })) {
            std::cout << " 条件达成 - 提前执行任务" << std::endl;
        } else {
            std::cout << "⏰ 等待超时 - 强制唤醒执行" << std::endl;
        }
    }
    worker.join();  // 确保任务线程结束
    // 任务执行
    std::cout << "=== 执行核心任务 ===" << std::endl;
    return 0;
}

执行结果：

主线程执行其他任务...
>>> 任务准备完成 <<<
 条件达成 - 提前执行任务
=== 执行核心任务 ===

或者结果如下所示：

主线程执行其他任务...
（等待2秒后）
⏰ 等待超时 - 强制唤醒执行
=== 执行核心任务 ===
>>> 任务准备完成 <<<  (稍后出现)

注意事项

1. 死锁：确保在调用 cv.wait() 前已经加锁。确保在调用 cv.notify_one() 或 cv.notify_all() 前已经解锁。
2. 资源竞争：使用互斥锁保护共享资源，避免数据竞争。
3. 性能：频繁的线程唤醒和等待可能会影响性能，尽量优化条件检查逻辑。

13.3.3 atomic

原子操作是不可分割的操作，确保在多线程环境下的正确性。

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

std::atomic<int> counter(0);

void increment() {
    for (int i = 0; i < 1000; ++i) {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);
    t1.join();
    t2.join();
    std::cout << "Counter: " << counter << std::endl; // 输出: Counter: 2000
    return 0;
}

13.4 异步机制

13.4.1 异步操作

使用 std::async 和 std::future：

#include <future>

int compute() { return 100; }

int main() {
    std::future<int> fut = std::async(std::launch::async, compute);
    int result = fut.get(); // 阻塞等待结果
    std::cout << "Result: " << result;
}

13.4 线程局部存储

使用 thread_local 关键字声明线程局部变量，每个线程拥有独立的变量副本。

13.4.1 thread_local

使用 thread_local

#include <iostream>
#include <thread>

thread_local int threadLocalData = 0;

void threadFunction(int id) {
    threadLocalData = id;
    std::cout << "Thread " << id << " has data: " << threadLocalData << std::endl;
}

int main() {
    std::thread t1(threadFunction, 1);
    std::thread t2(threadFunction, 2);

    t1.join();
    t2.join();

    return 0;
}

13.6 延迟任务

13.6.1 sleep_for

std::this_thread::sleep_for 是 C++11 引入的标准库函数，用于让当前线程休眠指定的时间。

示例代码

#include <iostream>
#include <chrono>  // 包含时间库
#include <thread>  // 包含线程库

int main() {
    std::cout << "Task started." << std::endl;
    // 延迟 300 毫秒
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    std::cout << "Task executed after 300ms." << std::endl;
    return 0;
}

代码说明

• std::this_thread::sleep_for 接受一个时间间隔作为参数。
• std::chrono::milliseconds(300) 表示 300 毫秒。

13.6.2 std::async

如果需要延迟执行任务而不阻塞当前线程，可以使用 std::async 和 std::future。

示例代码

#include <iostream>
#include <chrono>
#include <thread>
#include <future>  // 包含异步任务库

void delayedTask() {
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    std::cout << "Task executed after 300ms." << std::endl;
}

int main() {
    std::cout << "Task scheduled." << std::endl;
    // 使用 std::async 异步执行任务
    auto future = std::async(std::launch::async, delayedTask);
    // 主线程继续执行其他任务
    std::cout << "Main thread is doing other work..." << std::endl;
    // 等待异步任务完成
    future.wait();
    std::cout << "Main thread finished." << std::endl;
    return 0;
}

代码说明

• std::async 启动一个异步任务。
• std::launch::async 确保任务在新线程中执行。
• future.wait() 等待异步任务完成。

执行结果如下所示：

Task scheduled.
Main thread is doing other work...
Task executed after 3000ms.
Main thread finished.

13.6.3 使用定时器

13.6.4 std::thread

如果需要手动控制线程，可以使用 std::thread 和 std::this_thread::sleep_for。

示例代码

#include <iostream>
#include <chrono>
#include <thread>

void delayedTask() {
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    std::cout << "Task executed after 300ms." << std::endl;
}

int main() {
    std::cout << "Task scheduled." << std::endl;
    // 创建线程执行任务
    std::thread t(delayedTask);
    // 主线程继续执行其他任务
    std::cout << "Main thread is doing other work..." << std::endl;
    // 等待线程完成
    t.join();
    std::cout << "Main thread finished." << std::endl;
    return 0;
}

执行结果如下所示：

Task scheduled.
Main thread is doing other work...
Task executed after 3000ms.
Main thread finished.

13.7 线程池

C++ 标准库没有直接提供线程池，但可以使用第三方库或手动实现。

13.7.1 简易线程池

简单线程池实现

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

class ThreadPool {
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(queueMutex);
                        condition.wait(lock, [this] { return !tasks.empty() || stop; });
                        if (stop && tasks.empty()) return;
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    template <class F>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            tasks.emplace(std::forward<F>(f));
        }
        condition.notify_one();
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();
        for (std::thread& worker : workers) {
            worker.join();
        }
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop = false;
};

int main() {
    ThreadPool pool(4);

    for (int i = 0; i < 8; ++i) {
        pool.enqueue([i] {
            std::cout << "Task " << i << " is running on thread " << std::this_thread::get_id() << std::endl;
        });
    }

    return 0;
}

13.8 综合案例

13.8.4 生产者-消费者

#include <queue>
#include <mutex>
#include <condition_variable>

std::queue<int> data_queue;
std::mutex mtx;
std::condition_variable cv;

void producer() {
    for (int i = 0; i < 10; ++i) {
        {
            std::lock_guard lock(mtx);
            data_queue.push(i);
        }
        cv.notify_one(); // 通知消费者
    }
}

void consumer() {
    while (true) {
        std::unique_lock lock(mtx);
        cv.wait(lock, []{ return !data_queue.empty(); }); // 等待数据
        int val = data_queue.front();
        data_queue.pop();
        lock.unlock();
        // 处理数据...
    }
}