• C++多线程学习06

    参考

    参考资料:恋恋风辰官方博客

    测试异步执行std::async

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5.  
    6. // 模拟一个从数据库查询的操作
    7. std::string fetchDataFromDB(std::string query)
    8. {
    9.     // 模拟一个耗时操作
    10.     std::this_thread::sleep_for(std::chrono::seconds(5));
    11.     return "Data is: " + query;
    12. }
    13.  
    14. void use_async()
    15. {
    16.     // async异步调用fetchDataFromDB
    17.     std::future result_from_DB = std::async(std::launch::async, fetchDataFromDB, "Test Query");
    18.  
    19.     // 主线程做其他事情
    20.     std::cout << "Do something else... \n";
    21.  
    22.     // 从future中获取数据
    23.     std::string DB_Data = result_from_DB.get();
    24.     std::cout << DB_Data << std::endl;
    25.  
    26.     std::cout << "Async test finished! \n";
    27. }
    28.  
    29. int main()
    30. {
    31.     // 1. 测试异步执行
    32.     use_async();
    33.  
    34.     std::cout << "Finished!\n";
    35. }

    异步执行时，数据如果没有准备好，取数据时会阻塞。

    使用std::package_task打包任务

    1. int my_task()
    2. {
    3.     std::this_thread::sleep_for(std::chrono::seconds(5));
    4.     std::cout << "My Task run 5 seconds \n";
    5.     return 42;
    6. }
    7.  
    8. void use_package()
    9. {
    10.     // 创建一个包装了任务的 std::packaged_task 对象 
    11.     // package_task和future一起，可以实现一个线程池操作
    12.     std::packaged_task<int()> task(my_task);
    13.  
    14.     // 获取与任务有关的std::future对象, 代表一个异步操作结果
    15.     std::future<int> result = task.get_future();
    16.  
    17.     // 在另一个线程上执行
    18.     // package_task只支持移动构造
    19.     std::thread t(std::move(task));
    20.     t.detach();   // 后台执行，和变量t解耦，但是主线程依然是守护线程
    21.  
    22.     // 等待任务执行完毕，获取结果
    23.     int value = result.get();
    24.     std::cout << "The result is: " << value << std::endl;
    25. }
    26.  
    27. int main()
    28. {
    29.     // 2. 使用std::package_task打包任务
    30.     use_package();
    31.  
    32.     std::cout << "Finished!\n";
    33. }

    std::promise获取值

    1.  
    2. void set_value(std::promise<int> prom)
    3. {
    4.     std::this_thread::sleep_for(std::chrono::seconds(5));
    5.     prom.set_value(10);
    6.     std::cout << "promise set value success. \n";
    7. }
    8.  
    9. void use_promise()
    10. {
    11.     std::promise<int> prom;
    12.  
    13.     // promise与future对象关联
    14.     std::future<int> fut = prom.get_future();
    15.  
    16.     // 新线程中使用promise
    17.     std::thread t(set_value, std::move(prom));
    18.  
    19.     // 在主线程中等待future的值
    20.     std::cout << "Waiting for the thread to set the value...\n";
    21.     std::cout << "Value set by the thread: " << fut.get() << '\n';
    22.     t.join();
    23. }
    24.  
    25. int main()
    26. {
    27.     // 3. 使用promise
    28.     /*
    29.         promise类似package，package需要等待函数全部执行完毕后，才可以通过future获取值；
    30.         promise：绑定函数执行过程中如果设置了value后，可以在中途获取到.
    31.     */
    32.     use_promise();
    33.  
    34.     std::cout << "Finished!\n";
    35. }

    shared_future

    1. void myFunction(std::promise<int>&& promise) {
    2.     // 模拟一些工作
    3.     std::this_thread::sleep_for(std::chrono::seconds(1));
    4.     promise.set_value(42); // 设置 promise 的值
    5. }
    6.  
    7. void threadFunction(std::shared_future<int> future)
    8. {
    9.     try {
    10.         int result = future.get();
    11.         std::cout << "Result: " << result << std::endl;
    12.     }
    13.     catch (const std::future_error& e) {
    14.         std::cout << "Future error: " << e.what() << std::endl;
    15.     }
    16. }
    17.  
    18. void use_shared_future()
    19. {
    20.     std::promise<int> promise;
    21.     std::shared_future<int> future = promise.get_future();
    22.  
    23.     // 移动promise到线程中，执行类似计算的操作
    24.     std::thread my_thread1(myFunction, std::move(promise));
    25.  
    26.     // 定义取结果的线程
    27.     std::thread get_result1(threadFunction, future);
    28.     std::thread get_result2(threadFunction, future);
    29.  
    30.     my_thread1.join();
    31.     get_result1.join();
    32.     get_result2.join();
    33. }
    34.  
    35. void use_shared_future_error()
    36. {
    37.     std::promise<int> promise;
    38.  
    39.     // 隐式转换，默认转换为shared_future
    40.     std::shared_future<int> future = promise.get_future();
    41.  
    42.     // 移动promise到线程中，执行类似计算的操作
    43.     std::thread my_thread1(myFunction, std::move(promise));
    44.  
    45.     // 定义取结果的线程
    46.     std::thread get_result1(threadFunction, std::move(future));
    47.     std::thread get_result2(threadFunction, std::move(future));
    48.  
    49.     my_thread1.join();
    50.     get_result1.join();
    51.     get_result2.join();
    52. }
    53.  
    54.  
    55. int main()
    56. {
    57.     // 3. shared_future使用方法：供多个线程使用
    58.     use_shared_future();
    59.  
    60.     /*
    61.         在shared_future中，定义的future不可以被移动构造，否则另一个线程无法使用
    62.     */
    63.     // use_shared_future_error();   // 错误用法
    64.  
    65.     std::cout << "Finished!\n";
    66. }

    抛出异常

    1. // 定义异常，通过子线程抛出
    2. void set_exception(std::promise<void> prom)
    3. {
    4.     try
    5.     {
    6.         throw std::runtime_error("An error occurred! \n");
    7.     }
    8.     catch (const std::exception&)
    9.     {
    10.         prom.set_exception(std::current_exception());
    11.     }
    12. }
    13.  
    14. void use_promise_exception()
    15. {
    16.     std::promise<void> prom;
    17.     std::future<void> fut = prom.get_future();
    18.  
    19.     // 在新线程中设置promise异常
    20.     std::thread t(set_exception, std::move(prom));
    21.  
    22.     // 主线程中获取异常
    23.     try
    24.     {
    25.         std::cout << "Waiting for the thread to set the exception...\n";
    26.         fut.get();
    27.     }
    28.     catch (const std::exception& e)
    29.     {
    30.         std::cout << "Exception set by the thread: " << e.what() << '\n';
    31.     }
    32.     t.join();
    33. }
    34.  
    35. void use_promise_destruct()
    36. {
    37.     std::thread t;
    38.     std::future<int> fut;
    39.     {
    40.         std::promise<int> prom;
    41.         fut = prom.get_future();
    42.         t = std::thread(set_value, std::move(prom));
    43.     }
    44.     // 作用域结束后，prom可能因为release或者linux系统原因马上释放，导致错误.
    45.  
    46.     // 主线程中获取future值
    47.     std::cout << "Waiting for the thread to set the value...\n";
    48.     std::cout << "Value set by the thread: " << fut.get() << '\n';
    49.     t.join();
    50. }
    51.  
    52. int main()
    53. {
    54.     // 4. 使用promise，在主线程中获取异常
    55.     // 子线程抛出异常，主线程一定需要try捕获，否则会导致主线程崩溃。
    56.     use_promise_exception();
    57.  
    58.     // 作用域原因的错误用法
    59.     // 可能出错，也可能不会出错; 解决方法就是将prom按照智能指针方式传递.
    60.     // use_promise_destruct();
    61.  
    62.     std::cout << "Finished!\n";
    63. }

    线程池

    首先在头文件中定义ThreadPool.h：

    1. #ifndef __THREAD_POOL_H__
    2. #define __THREAD_POOL_H__
    3.  
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11. #include 
    12.  
    13. class ThreadPool
    14. {
    15. public:
    16. 	ThreadPool& operator=(const ThreadPool&) = delete;
    17. 	
    18. 	// 实现一个单例模式
    19. 	static ThreadPool& instance()
    20. 	{
    21. 		static ThreadPool instance;
    22. 		return instance;
    23. 	}
    24.  
    25. 	using Task = std::packaged_task<void()>;
    26.  
    27. 	~ThreadPool()
    28. 	{
    29. 		stop();
    30. 	}
    31.  
    32. 	// 一个返回类型后置语法
    33. 	template<class F, class ...Args>
    34. 	auto commit(F&& f, Args&&... args) -> std::future<decltype(f(args...))>
    35. 	{
    36. 		using RetType = decltype(f(args...));
    37. 		
    38. 		// 如果线程池已经结束了，那么直接返回空的future.
    39. 		if (stop_.load()) {
    40. 			return std::future{};
    41. 		}
    42.  
    43. 		// 将回调函数f以及可变参数绑定一起，生成一个无参函数。
    44. 		auto task = std::make_sharedRetType()>>(
    45. 			std::bind(std::forward(f), std::forward(args)...));
    46.  
    47. 		// 通过std::future获取运行结果
    48. 		std::future ret = task->get_future();
    49. 		{
    50. 			// 将需要执行的任务放入任务队列, 等待线程领取.
    51. 			std::lock_guard		cv_mt(cv_mt_);
    52.  
    53. 			// 传递一个回调函数, lambda表达式，在弹出时执行
    54. 			// 通过lambda表达式捕获task，保证task在回调之前避免被销毁，类似伪闭包.
    55. 			tasks_.emplace([task] { (*task)(); });
    56. 		}
    57.  
    58. 		// 通知挂起的线程执行
    59. 		cv_lock_.notify_one();
    60. 		return ret;
    61. 	}
    62.  
    63. 	int idleThreadCount() 
    64. 	{
    65. 		return thread_num_;
    66. 	}
    67.  
    68. private:
    69. 	ThreadPool(unsigned int num = 5): stop_(false)
    70. 	{
    71. 		{
    72. 			if (num < 1) thread_num_ = 1;
    73. 			else thread_num_ = num;
    74. 		}
    75. 		start();
    76. 	}
    77.  
    78. 	void start()
    79. 	{
    80. 		for (int i = 0; i < thread_num_; i++) {
    81. 			pool_.emplace_back([this]() {
    82. 				while (!this->stop_.load()) {
    83. 					Task task;
    84. 					{
    85. 						std::unique_lock cv_mt(cv_mt_);
    86. 						
    87. 						// 通过谓词判断，如果线程池已经停止，或者任务队列为空
    88. 						// 通过条件变量让线程挂起
    89. 						this->cv_lock_.wait(cv_mt, [this] {
    90. 							return this->stop_.load() || !this->tasks_.empty();
    91. 							});
    92.  
    93. 						if (this->tasks_.empty())
    94. 							return;
    95.  
    96. 						task = std::move(this->tasks_.front());
    97. 						this->tasks_.pop();
    98. 					}
    99. 					this->thread_num_--;
    100. 					task();
    101. 					this->thread_num_++;
    102. 				}
    103. 			});
    104. 		}
    105. 	}
    106.  
    107. 	void stop()
    108. 	{
    109. 		// 原子操作，将停止信号设置为true.
    110. 		stop_.store(true);
    111.  
    112. 		// 通知所有线程，停止前避免卡死.
    113. 		cv_lock_.notify_all();
    114. 		for (auto& td : pool_) {
    115. 			if (td.joinable()) {
    116. 				std::cout << "join thread " << td.get_id() << std::endl;
    117. 				td.join();
    118. 			}
    119. 		}
    120. 	}
    121.  
    122. private:
    123. 	std::mutex               cv_mt_;
    124. 	std::condition_variable  cv_lock_;
    125. 	std::atomic_bool         stop_;
    126. 	std::atomic_int          thread_num_;
    127. 	std::queue         tasks_;		// 任务队列
    128. 	std::vector pool_;			// 线程池
    129. };
    130.  
    131. #endif // __THREAD_POOL_H__

    然后在主函数中测试调用:

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5. #include "ThreadPool.h"
    6.  
    7. int main()
    8. {
    9.     // 5. 使用线程池, 以及使用例子
    10.     int m = 0;
    11.     ThreadPool::instance().commit([](int& m) {
    12.         m = 1024;
    13.         std::cout << "Inner set m is " << m << std::endl;
    14.     }, m);
    15.     std::this_thread::sleep_for(std::chrono::seconds(3));
    16.     std::cout << "Main: m is " << m << std::endl;
    17.  
    18.     ThreadPool::instance().commit([](int& m) {
    19.         m = 1024;
    20.         std::cout << "Inner set m is " << m << std::endl;
    21.         }, std::ref(m));
    22.     std::this_thread::sleep_for(std::chrono::seconds(3));
    23.     std::cout << "Main: m is " << m << std::endl;
    24.  
    25.     /*
    26.         1. 如果任务有顺序要求，不可以用线程池.
    27.         2. 任务是有顺序要求的.    
    28.     */
    29.  
    30.     std::cout << "Finished!\n";
    31. }

    最后的测试结果：

  • 相关阅读:
    C++指针
    完整实现-通过DelayQueue实现延时任务
    C++----二叉树的进阶
    【网络】计算机网络基础概念入门
    HackTheBox You know 0xDiablos pwn题目
    C++常见容器实现原理
    linux 中的根文件系统
    【 第十四章 网络编程要素（IP和端口，网络通信协议），实现TCP的网络编程】
    spring boot启动报错java.lang.UnsupportedOperationException
    SpringCloud之Feign集成Ribbon
  • 原文地址：https://blog.csdn.net/weixin_42130300/article/details/136291061