从C向C++25——C++11线程池
一.线程库的使用
1.基本使用
要创建线程，我们需要一个可调用的函数或函数对象（回调函数），作为线程的入口点。在C++11中，我们可以使用函数指针、函数对象或lambda表达式来实现。创建线程的基本语法如下：
#include <thread>
std::thread t(function_name, args...);
-
function_name是线程入口点的函数或可调用对象 -
args...是传递给函数的参数
创建线程后，我们可以使用t.join()等待线程完成，或者使用t.detach()分离线程，让它在后台运行。当我们创建一个线程后，我们可能需要等待它完成，以便获取线程的执行结果或执行清理操作。我们可以使用t.join()方法来等待线程完成。
#include <iostream>
#include <thread>
#include <string>
void printIDA(std::string meg)
{
std::cout << meg << std::endl;
}
int main()
{
std::thread thread1(printIDA,"hello thread!");
thread1.join();
return 0;
}
joinable()方法返回一个布尔值，如果线程可以被join()或detach()，则返回true，否则返回false。如果我们试图对一个不可加入的线程调用join()或detach()，则会抛出一个std::system_error异常。
所以上述代码严格规范为：
#include <iostream>
#include <thread>
#include <string>
void printIDA(std::string meg)
{
std::cout << meg << std::endl;
}
int main()
{
std::thread thread1(printIDA,"hello thread!");
bool is= thread1.joinable();
//thread1.detach(); //分离线程
if (is)
{
thread1.join(); //主进程阻塞，等待线程结束
//thread1.detach(); //分离线程
}
else{}
return 0;
}
2.常见错误
-
1：传递局部变量的问题
#include <iostream> #include <thread> #include <string> void foo(int& x) { x += 1; } int main() { int a = 1; //std::thread thread1(foo,1); //编译报错，原因左值引用无法引用右值 //std::thread thread1(foo,a); //编译报错，局部变量用完自动销毁 std::thread thread1(foo, std::ref(a)); thread1.join(); std::cout << a << std::endl; return 0; }这是最基础的错误，同类型的例如在某一个全局函数中建立线程调用
foo函数即使使用std::ref也会报错。这里主要的一个点就是调用的是局部变量，执行完当行后被释放，而线程执行是需要一个过程的，中间有可能会使用到被释放的变量控件而出错。 -
2.类成员函数作为入口函数，类对象被提前释放
#include <iostream> #include <thread> class MyClass { public: void func() { std::cout << "Thread " << std::this_thread::get_id() << " started" << std::endl; // do some work std::cout << "Thread " << std::this_thread::get_id() << " finished" << std::endl; } }; int main() { MyClass obj; std::thread t(&MyClass::func, &obj); // obj 被提前销毁了，会导致未定义的行为 return 0; }解释一下
std::thread t(&MyClass::func, &obj);这一行，对于在线程中调用一个类的成员函数，即使该成员函数没有参数，也也必须使用函数指针的格式，即&Class::memberFunction，因为编译器需要知道你是在引用一个成员函数。然后，你需要提供对象实例的指针，这样线程就知道要对哪个对象实例执行该成员函数。解决问题的方法，其实就是之前了解的智能指针，头文件
#include <memory>，其用来解决内存泄露的方法：自动释放，刚好用来解决这里的问题。
二.补充使用
1.互斥量
有考研408《操作系统》的基础，其实就是线程访问同一个变量的问题，利用互斥量保证同一时刻只能有一个线程访问该变量
互斥量（mutex）是一种用于实现多线程同步的机制，用于确保多个线程之间对共享资源的访问互斥。互斥量通常用于保护共享数据的访问，以避免多个线程同时访问同一个变量或者数据结构而导致的数据竞争问题。
互斥量提供了两个基本操作：lock() 和 unlock()。当一个线程调用 lock() 函数时，如果互斥量当前没有被其他线程占用，则该线程获得该互斥量的所有权，可以对共享资源进行访问。如果互斥量当前已经被其他线程占用，则调用 lock() 函数的线程会被阻塞，直到该互斥量被释放为止。
#include <iostream>
#include <thread>
#include <mutex>
int shared_data = 0;
std::mutex mtx;
void func(int n) {
for (int i = 0; i < 100000; ++i) {
mtx.lock(); //加锁
shared_data++;
mtx.unlock(); //解锁
}
}
int main() {
std::thread t1(func, 1);
std::thread t2(func, 2);
t1.join();
t2.join();
std::cout << "Final shared_data = " << shared_data << std::endl;
return 0;
}
2.互斥量死锁
假设有两个线程 T1 和 T2，它们需要对两个互斥量 mtx1 和 mtx2 进行访问，而且需要按照以下顺序获取互斥量的所有权：
-
T1 先获取 mtx1 的所有权，再获取 mtx2 的所有权。
-
T2 先获取 mtx2 的所有权，再获取 mtx1 的所有权。
如果两个线程同时执行，就会出现死锁问题。因为 T1 获取了 mtx1 的所有权，但是无法获取 mtx2 的所有权，而 T2 获取了 mtx2 的所有权，但是无法获取 mtx1 的所有权，两个线程互相等待对方释放互斥量，导致死锁。
3.lock_guard
std::lock_guard 是 C++ 标准库中的一种互斥量封装类，用于保护共享数据，防止多个线程同时访问同一资源而导致的数据竞争问题。
std::lock_guard 的特点如下：
- 当构造函数被调用时，该互斥量会被自动锁定。
- 当析构函数被调用时，该互斥量会被自动解锁。
注意：std::lock_guard 对象不能复制或移动，因此它只能在局部作用域中使用。
void func(int n) {
for (int i = 0; i < 10000; ++i) {
mtx.lock(); //加锁
shared_data++;
mtx.unlock(); //解锁
}
}
那么就可以替换为：
void func(int n) {
for (int i = 0; i < 10000; ++i) {
std::lock_guard<std::mutex> lg(mtx);
shared_data++;
}
}
4.std::unique_lock
std::unique_lock 是 C++ 标准库中提供的一个互斥量封装类，用于在多线程程序中对互斥量进行加锁和解锁操作。它的主要特点是可以对互斥量进行更加灵活的管理，包括延迟加锁、条件变量、超时等。
std::unique_lock 提供了以下几个成员函数：
lock()：尝试对互斥量进行加锁操作，如果当前互斥量已经被其他线程持有，则当前线程会被阻塞，直到互斥量被成功加锁。try_lock()：尝试对互斥量进行加锁操作，如果当前互斥量已经被其他线程持有，则函数立即返回false，否则返回true。try_lock_for(const std::chrono::duration<Rep, Period>& rel_time)：尝试对互斥量进行加锁操作，如果当前互斥量已经被其他线程持有，则当前线程会被阻塞，直到互斥量被成功加锁，或者超过了指定的时间。try_lock_until(const std::chrono::time_point<Clock, Duration>& abs_time)：尝试对互斥量进行加锁操作，如果当前互斥量已经被其他线程持有，则当前线程会被阻塞，直到互斥量被成功加锁，或者超过了指定的时间点。unlock()：对互斥量进行解锁操作。
简单学习的话，记住两点：
- 可以替代
lock_guard，同样支持自动加锁和自动解锁 - 可以选择加锁的种类，其提供了一个延时加锁的功能
所以
void func(int n) {
for (int i = 0; i < 10000; ++i) {
std::lock_guard<std::mutex> lg(mtx);
shared_data++;
}
}
等价于
void func(int n) {
for (int i = 0; i < 10000; ++i) {
std::unique_lock<std::mutex> lg(mtx);
shared_data++;
}
}
可以ctrl+shift+左键查看源码，根据其构造函数查看其种类：
void func(int n) {
for (int i = 0; i < 10000; ++i) {
std::unique_lock<std::mutex> lg(mtx，std::defer_lock);
lg.trr_lock_for(std::chrono::seconds(5));
shared_data++;
}
}
注意：保证mtx是可以支持时间的互斥量。
5.原子变量
std::atomic 是 C++11 标准库中的一个模板类，用于实现多线程环境下的原子操作。它提供了一种线程安全的方式来访问和修改共享变量，可以避免多线程环境中的数据竞争问题。
std::atomic 的使用方式类似于普通的 C++ 变量，但是它的操作是原子性的。也就是说，在多线程环境下，多个线程同时对同一个 std::atomic 变量进行操作时，不会出现数据竞争问题。
以下是一些常用的 std::atomic 操作：
-
load()：将std::atomic变量的值加载到当前线程的本地缓存中，并返回这个值。 -
store(val)：将val的值存储到std::atomic变量中，并保证这个操作是原子性的。 -
exchange(val)：将val的值存储到std::atomic变量中，并返回原先的值。 -
compare_exchange_weak(expected, val)和compare_exchange_strong(expected, val)：比较std::atomic变量的值和expected的值是否相同，如果相同，则将val的值存储到std::atomic变量中，并返回true；否则，将std::atomic变量的值存储到expected中，并返回false。
以下是一个示例，演示了如何使用 std::atomic 进行原子操作：
#include <atomic>
#include <iostream>
#include <thread>
std::atomic<int> count = 0;
void increment() {
for (int i = 0; i < 1000000; ++i) {
count++;
}
}
int main() {
std::thread t1(increment);
std::thread t2(increment);
t1.join();
t2.join();
std::cout << count << std::endl;
return 0;
}
使用原子变量，比较互斥量更加快！
三.其他场景
1.std::call_once
涉及到单例模式，要详细学习C++的设计模式
单例设计模式是一种常见的设计模式，用于确保某个类只能创建一个实例。由于单例实例是全局唯一的，因此在多线程环境中使用单例模式时，需要考虑线程安全的问题。下面是一个简单的单例模式的实现：
class Singleton {
public:
static Singleton& getInstance() {
static Singleton instance;
return instance;
}
void setData(int data) {
m_data = data;
}
int getData() const {
return m_data;
}
private:
Singleton() {}
Singleton(const Singleton&) = delete; //禁用拷贝构造
Singleton& operator=(const Singleton&) = delete; //禁用操作符=
int m_data = 0;
};
在这个单例类中，我们使用了一个静态成员函数 getInstance() 来获取单例实例，该函数使用了一个静态局部变量 instance 来存储单例实例。由于静态局部变量只会被初始化一次，因此该实现可以确保单例实例只会被创建一次。
但是，该实现并不是线程安全的。如果多个线程同时调用 getInstance() 函数，可能会导致多个对象被创建，从而违反了单例模式的要求。此外，如果多个线程同时调用 setData() 函数来修改单例对象的数据成员 m_data，可能会导致数据不一致或不正确的结果。
为了解决这些问题，我们可以使用 std::call_once 来实现一次性初始化，从而确保单例实例只会被创建一次。下面是一个使用 std::call_once 的单例实现：
class Singleton {
public:
static Singleton& getInstance() {
std::call_once(m_onceFlag, &Singleton::init);
return *m_instance;
}
void setData(int data) {
m_data = data;
}
int getData() const {
return m_data;
}
private:
Singleton() {}
Singleton(const Singleton&) = delete;
Singleton& operator=(const Singleton&) = delete;
static void init() {
m_instance.reset(new Singleton);
}
static std::unique_ptr<Singleton> m_instance;
static std::once_flag m_onceFlag;
int m_data = 0;
};
std::unique_ptr<Singleton> Singleton::m_instance;
std::once_flag Singleton::m_onceFlag;
在这个实现中，我们使用了一个静态成员变量 m_instance 来存储单例实例，使用了一个静态成员变量 m_onceFlag 来标记初始化是否已经完成。在 getInstance() 函数中，我们使用 std::call_once 来调用 init() 函数，确保单例实例只会被创建一次。在 init() 函数中，我们使用了 std::unique_ptr 来创建单例实例。
使用 std::call_once 可以确保单例实例只会被创建一次，从而避免了多个对象被创建的问题。此外，使用 std::unique_ptr 可以确保单例实例被正确地释放，避免了内存泄漏的问题。
emplate<class Callable, class... Args> void call_once(std::once_flag& flag, Callable&& func, Args&&... args);使用
std::call_once的过程中，需要注意以下几点：
flag参数必须是一个std::once_flag类型的对象，并且在多次调用call_once函数时需要使用同一个flag对象。func参数是需要被调用的函数或可调用对象。该函数只会被调用一次，因此应该确保该函数是幂等的。args参数是func函数或可调用对象的参数。如果func函数没有参数，则该参数可以省略。std::call_once函数会抛出std::system_error异常，如果在调用func函数时发生了异常，则该异常会被传递给调用者。
2.条件变量
std::condition_variable 的步骤如下：
- 创建一个
std::condition_variable对象。 - 创建一个互斥锁
std::mutex对象，用来保护共享资源的访问。 - 在需要等待条件变量的地方使用
std::unique_lock<std::mutex>对象锁定互斥锁，并调用std::condition_variable::wait()、std::condition_variable::wait_for()或std::condition_variable::wait_until()函数等待条件变量。 - 在其他线程中需要通知等待的线程时，调用
std::condition_variable::notify_one()或std::condition_variable::notify_all()函数通知等待的线程。
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
std::mutex g_mutex;
std::condition_variable g_cv;
std::queue<int> g_queue;
void Producer() {
for (int i = 0; i < 10; i++) {
{
std::unique_lock<std::mutex> lock(g_mutex);
g_queue.push(i);
std::cout << "Producer: produced " << i << std::endl;
}
g_cv.notify_one();
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
void Consumer() {
while (true) {
std::unique_lock<std::mutex> lock(g_mutex);
g_cv.wait(lock, []() { return !g_queue.empty(); });
int value = g_queue.front();
g_queue.pop();
std::cout << "Consumer: consumed " << value << std::endl;
}
}
int main() {
std::thread producer_thread(Producer);
std::thread consumer_thread(Consumer);
producer_thread.join();
consumer_thread.join();
return 0;
}
3.实现线程池
理解线程池：有一个线程数组和一个任务队列，每当生产者往任务队列添加任务时，线程数组派出一个线程来执行相应的任务。
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <string>
#include <vector>
#include <queue>
#include <functional>
class Thread_pool
{
public:
Thread_pool(int num)
{
this->stop = false;
for (int i = 0; i < num; i++)
{
threads.emplace_back([this] {
while (1)
{
std::unique_lock<std::mutex> lock(mtx);
condition.wait(lock, [this] {
return !tasks.empty() || stop;
});
if (stop && tasks.empty())
{
return;
}
std::function<void()> task(std::move(tasks.front()));
tasks.pop();
lock.unlock();
task();
}
});
}
}
~Thread_pool()
{
{
std::unique_lock<std::mutex> lock(mtx);
stop = true;
}
condition.notify_all();
for (auto& t : threads)
{
t.join();
}
}
template<class F,class...Args> //模板
void enqueue(F&& f, Args&&... args) //&&万能引用
{
std::function<void()>task = std::bind(std::forward<F>(f), std::forward<Args>(args)...); //可调用对象
{
std::unique_lock<std::mutex> lock(mtx);
tasks.emplace(std::move(task));
}
condition.notify_one();
}
private:
std::vector<std::thread> threads;
std::queue<std::function<void()>> tasks; //需要学习一下可调用对象
std::mutex mtx;
std::condition_variable condition;
bool stop;
};
int main() {
Thread_pool pool(4);
for (int i = 0; i < 10; i++)
{
pool.enqueue([i] {
std::cout << "task :" << i << "is runing" << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1)); //C++11时间库chrono
std::cout << "task :" << i << "is done" << std::endl;
});
}
return 0;
}
4.后续知识点
- 异步并发： async future packaged_task promise
更多推荐
16
0- 0
已为社区贡献2条内容
所有评论(0)