深入探究C++并发编程：信号 异步 原子

1.c++中的 "信号"

1.1 std::condition_variable、wait与notify_one

std::condition_variable是C++11引入的线程同步原语，用于实现线程间的条件等待和通知机制。它通常与std::mutex配合使用，以确保线程安全。

1. 构造函数

std::condition_variable的构造函数非常简单，它不需要任何参数：

std::condition_variable cv;

它默认构造一个条件变量对象。

2. wait方法

wait方法用于阻塞当前线程，直到条件变量被通知（notify_one或notify_all），或者直到谓词条件成立。wait方法有两种重载形式：

1. 无谓词版本：

   void wait(std::unique_lock<std::mutex>& lock);

   • 参数：std::unique_lock<std::mutex>& lock，表示一个已锁定的互斥锁。

   • 功能：调用wait时，会释放lock所持有的互斥锁，并将当前线程阻塞，直到被通知。

   • 注意：在被唤醒后，线程会重新尝试获取互斥锁。

2. 带谓词版本：

   template <class Predicate>
   void wait(std::unique_lock<std::mutex>& lock, Predicate pred);

   • 参数：除了互斥锁外，还接受一个谓词函数pred。

   • 功能：只有当pred返回true时，线程才会继续执行；否则会继续等待通知。

   • 示例：

     std::unique_lock<std::mutex> lock(mtx);
     cv.wait(lock, []{ return condition_met; });

   • 优点：可以减少虚假唤醒。

    3. wait_for方法

wait_for方法允许线程等待一个特定的超时时间：

template <class Rep, class Period>
std::cv_status wait_for(std::unique_lock<std::mutex>& lock, const std::chrono::duration<Rep, Period>& timeout);

• 参数：

  • std::unique_lock<std::mutex>& lock：已锁定的互斥锁。

  • std::chrono::duration<Rep, Period>& timeout：超时时间。

• 返回值：std::cv_status，表示等待结果：

  • std::cv_status::no_timeout：在超时时间内被通知或条件满足。

  • std::cv_status::timeout：超时。

  4. wait_until方法

wait_until方法允许线程等待直到一个特定时间点：

template <class Clock, class Duration>
std::cv_status wait_until(std::unique_lock<std::mutex>& lock, const std::chrono::time_point<Clock, Duration>& timeout_time);

• 参数：

  • std::unique_lock<std::mutex>& lock：已锁定的互斥锁。

  • std::chrono::time_point<Clock, Duration>& timeout_time：等待的截止时间点。

• 返回值：std::cv_status，表示等待结果。

 5. 通知方法

条件变量提供了两种通知方法：

1. notify_one：

   void notify_one();

   • 功能：唤醒一个等待该条件变量的线程。如果有多个线程等待，唤醒哪个线程是未定义的。

2. notify_all：

   void notify_all();

   • 功能：唤醒所有等待该条件变量的线程。

  6. 使用注意事项

1. 互斥锁的必要性：wait方法必须在已锁定的互斥锁下调用。

2. 虚假唤醒：即使没有通知，wait方法也可能因虚假唤醒而返回，因此建议使用带谓词版本。

3. 线程安全：条件变量和互斥锁共同保护共享数据，确保线程安全。

1.2 示例代码

以下是一个使用std::condition_variable的生产者-消费者模型：

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

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

void producer() {
    for (int i = 0; i < 10; ++i) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        std::unique_lock<std::mutex> lock(mtx);
        queue.push(i);
        std::cout << "Producer: " << i << std::endl;
        cv.notify_one();
    }
}

void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, []{ return !queue.empty(); });
        int value = queue.front();
        queue.pop();
        std::cout << "Consumer: " << value << std::endl;
        if (value == 9) break; // 结束条件
    }
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer);
    t1.join();
    t2.join();
    return 0;
}
//roducer: 0
Consumer: 0
Producer: 1
Consumer: 1
Producer: 2
Consumer: 2
Producer: 3
Consumer: 3
Producer: 4
Consumer: 4
Producer: 5
Consumer: 5
Producer: 6
Consumer: 6
Producer: 7
Consumer: 7
Producer: 8
Consumer: 8
Producer: 9
Consumer: 9

在这个例子中，生产者线程通过notify_one通知消费者线程，消费者线程通过wait等待队列非空

2 异步任务的多种实现方式

C++11引入了面向异步操作的组件，包括std::async、std::future、std::packaged_task和std::promise，它们为我们提供了多种实现异步任务的方式。

std::async 是 C++11 引入的一个非常强大的工具，用于启动一个异步任务，并返回一个 std::future 对象，通过该对象可以获取任务的返回值。它允许我们以一种简单的方式实现并发编程，而无需直接管理线程。

2.1 std::async 和std::future的参数和用法

template <class Function, class... Args>
std::future<typename std::result_of<Function(Args...)>::type>
async(Function&& f, Args&&... args);

std::future是一个模板类，用于表示异步操作的结果。它允许一个线程获取另一个线程执行任务的结果，或者等待任务完成

1. 参数详解

std::async 的参数主要包括：

1. Function&& f

   • 这是需要异步执行的任务函数。它可以是一个普通函数、Lambda 表达式、函数对象或绑定器（如 std::bind 的结果）。

   • 函数的返回值类型决定了 std::future 的模板参数类型。

2. Args&&... args

   • 这是可变参数模板，表示传递给任务函数的参数。

   • 这些参数会被完美转发到任务函数中。

2. 返回值

std::async 返回一个 std::future 对象，其模板参数类型是任务函数的返回值类型。通过 std::future，可以获取异步任务的返回值或状态。

3. 启动策略

std::async 的执行策略由 std::launch 控制，可以指定为以下两种模式之一：

1. std::launch::async

   • 强制任务在新线程中异步运行。

   • 适用于需要并行执行的任务。

2. std::launch::deferred

   • 任务会在调用 std::future::get() 或 std::future::wait() 时才开始执行。

   • 适用于计算密集型任务，避免线程切换的开销。

3. 默认行为

   • 如果不显式指定启动策略，std::async 会根据实现选择 std::launch::async 或 std::launch::deferred。

   • 这种行为的不确定性可能导致一些不可预测的问题，因此建议显式指定启动策略。

4.std::future 使用方法：

• 获取任务结果：通过std::future::get()方法获取异步任务的返回值或异常。

• 检查状态：可以使用std::future::valid()检查std::future对象是否与一个有效的异步任务相关联。

• 等待任务完成：可以使用std::future::wait()或std::future::wait_for()方法等待任务完成。

示例代码

以下是一些使用 std::async 的示例代码，展示其参数的使用方式。

示例 1：简单异步任务

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

int add(int a, int b) {
    std::this_thread::sleep_for(std::chrono::seconds(2)); // 模拟耗时操作
    return a + b;
}

int main() {
    // 启动异步任务
    std::future<int> result = std::async(std::launch::async, add, 5, 3);

    std::cout << "Main thread is doing other work..." << std::endl;

    // 获取异步任务的返回值
    int sum = result.get();
    std::cout << "Result: " << sum << std::endl;

    return 0;
}

输出：

Main thread is doing other work...
Result: 8

示例 2：使用 Lambda 表达式

#include <iostream>
#include <future>

int main() {
    // 使用 Lambda 表达式作为任务函数
    std::future<int> result = std::async(std::launch::async, [](int x, int y) {
        return x * y;
    }, 4, 5);

    std::cout << "Main thread is doing other work..." << std::endl;

    // 获取异步任务的返回值
    int product = result.get();
    std::cout << "Result: " << product << std::endl;

    return 0;
}

输出：

Main thread is doing other work...
Result: 20

示例 3：延迟执行

#include <iostream>
#include <future>

int main() {
    // 使用 std::launch::deferred 策略
    std::future<int> result = std::async(std::launch::deferred, [](int x) {
        return x * x;
    }, 7);

    std::cout << "Main thread is doing other work..." << std::endl;

    // 调用 get() 时才开始执行任务
    int square = result.get();
    std::cout << "Result: " << square << std::endl;

    return 0;
}

输出：

Main thread is doing other work...
Result: 49

注意事项

1. 线程安全

   • std::async 本身是线程安全的，但任务函数内部的逻辑需要确保线程安全。

2. 异常处理

   • 如果任务函数抛出异常，异常会被存储在 std::future 中，可以通过 std::future::get() 捕获。

3. 资源管理

   • 当 std::future 被销毁时，如果任务尚未完成，会触发线程的销毁。因此，需要确保 std::future 的生命周期足够长。

4. 启动策略的选择

   • 显式指定启动策略可以避免不确定的行为，特别是对于性能敏感的应用。

2.2std::packaged_task

std::packaged_task是一个模板类，用于封装一个可调用对象（如函数、Lambda表达式或函数对象），并将其与一个std::future对象相关联。它允许将任务的创建和执行分离，便于任务的调度和结果获取。

使用方法：

1. 创建一个std::packaged_task对象，并绑定一个可调用对象。

2. 调用get_future()获取与任务结果相关的std::future对象。

3. 在另一个线程中调用std::packaged_task对象以执行任务。

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

int add(int a, int b) {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    return a + b;
}

int main() {
    std::packaged_task<int(int, int)> task(add); // 封装任务
    std::future<int> result = task.get_future(); // 获取future对象

    std::thread t(std::move(task), 5, 3); // 在新线程中执行任务
    t.detach();

    std::cout << "Waiting for the result..." << std::endl;
    std::cout << "Result: " << result.get() << std::endl;
    return 0;
}

2.3 std::promise：设置异步任务的结果

std::promise是一个模板类，用于设置异步任务的结果。它与std::future配合使用，允许一个线程将值或异常传递给另一个线程。

使用方法：

1. 创建一个std::promise对象。

2. 调用get_future()获取与std::promise对象关联的std::future对象。

3. 在任务完成时，调用std::promise::set_value()设置任务的结果，或调用std::promise::set_exception()设置异常。

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

void setValue(std::promise<int> prom) {
    prom.set_value(42); // 设置任务结果
}

int main() {
    std::promise<int> prom; // 创建promise对象
    std::future<int> fut = prom.get_future(); // 获取future对象

    std::thread t(setValue, std::move(prom)); // 在新线程中设置结果
    t.detach();

    std::cout << "Waiting for the result..." << std::endl;
    std::cout << "Result: " << fut.get() << std::endl;
    return 0;
}

2.4 总结使用场景总结

• std::async：

  • 适用场景：当你需要快速启动一个异步任务并获取结果时，std::async是最简单的选择。

  • 示例：计算密集型任务（如数学计算）、I/O操作（如文件读写）。

• std::packaged_task：

  • 适用场景：当你需要更灵活地控制任务的执行时机，或者需要重复执行任务时，std::packaged_task是更好的选择。

  • 示例：任务队列、线程池中的任务调度。

• std::promise：

  • 适用场景：当你需要手动设置任务结果，或者需要在多个线程之间传递任务结果时，std::promise是必要的。

  • 示例：线程间通信、异步任务的回调机制。

3 c++中的原子操作

在C++中，原子操作是指在多线程环境中不会被中断的基本操作。这些操作在执行过程中不会被其他线程打断，从而保证了线程安全。C++11引入了std::atomic库，提供了一组用于执行原子操作的类型和函数，使得开发者能够轻松地实现线程安全的变量操作，而无需使用互斥锁（std::mutex）。

1. 原子操作的重要性

在多线程程序中，多个线程可能会同时访问和修改共享变量。如果没有适当的同步机制，可能会导致竞态条件（race condition），从而导致程序行为异常。原子操作通过确保操作的不可分割性，避免了竞态条件，提高了程序的线程安全性。

2. std::atomic 类模板

std::atomic 是 C++ 标准库中用于实现原子操作的核心工具。它是一个模板类，可以对各种基本数据类型（如int、float、bool等）提供原子操作支持。

#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); // 原子加1
    }
}

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

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

    std::cout << "Counter: " << counter.load() << std::endl; // 原子读取
    return 0;
}

3. 常用原子操作函数

std::atomic 提供了一系列原子操作函数，用于对原子变量进行安全操作：

1. load()：

   • 原子地读取变量的值。

   • 示例：int value = counter.load();

2. store(value)：

   • 原子地设置变量的值。

   • 示例：counter.store(10);

3. fetch_add(value)：

   • 原子地将变量的值增加指定值，并返回增加前的值。

   • 示例：int old_value = counter.fetch_add(1);

4. fetch_sub(value)：

   • 原子地将变量的值减去指定值，并返回减去前的值。

   • 示例：int old_value = counter.fetch_sub(1);

5. fetch_and(value)、fetch_or(value)、fetch_xor(value)：

   • 原子地对变量执行按位与、按位或、按位异或操作。

   • 示例：int old_value = counter.fetch_and(0b1111);

6. compare_exchange_weak(expected, desired) 和 compare_exchange_strong(expected, desired)：

   • 原子地比较并交换操作。如果变量的当前值等于expected，则将其设置为desired。

   • 示例：

     int expected = counter.load();
     int desired = 10;
     counter.compare_exchange_weak(expected, desired);

4. 内存顺序（Memory Order）

原子操作的另一个重要方面是内存顺序（Memory Order），它定义了操作的内存可见性和顺序。C++ 提供了多种内存顺序选项，用于控制原子操作的内存同步行为：

1. std::memory_order_relaxed：

   • 最弱的内存顺序，仅保证操作的原子性，不保证内存顺序。

   • 适用于简单的计数器等场景。

2. std::memory_order_acquire：

   • 用于加载操作，表示当前线程在获取锁后可以读取其他线程写入的值。

3. std::memory_order_release：

   • 用于存储操作，表示当前线程在释放锁前写入的值对其他线程可见。

4. std::memory_order_acq_rel：

   • 用于读写操作，结合了acquire和release的语义，常用于互斥锁。

5. std::memory_order_seq_cst（默认值）：

   • 最强的内存顺序，保证操作的全局顺序一致性。

   • 适用于需要严格同步的场景。

5. 原子操作的优势

1. 性能优势：

   • 原子操作通常比互斥锁更高效，因为它们直接利用硬件支持的原子指令，避免了互斥锁的上下文切换开销。

2. 简单易用：

   • std::atomic 提供了简单易用的接口，使得线程安全的变量操作变得非常直观。

3. 避免竞态条件：

   • 原子操作确保了操作的不可分割性，从而避免了竞态条件，提高了程序的可靠性。

6. 示例代码

以下是一个使用原子操作的线程安全计数器示例：

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

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

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

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

    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back(increment);
    }

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

    std::cout << "Counter: " << counter.load() << std::endl; // 应该接近 100000
    return 0;
}

7. 注意事项

1. 内存顺序的选择：

   • 默认情况下，std::atomic 使用std::memory_order_seq_cst，但根据具体需求可以选择更弱的内存顺序以提高性能。

2. 复合操作的原子性：

   • 如果需要多个操作的组合是原子的（例如，读取和修改），可以使用compare_exchange_weak或compare_exchange_strong。

3. 硬件支持：

   • 原子操作依赖于硬件支持，因此在某些平台上可能无法完全实现。不过，现代处理器通常都支持原子操作。