【C++ 语言】线程安全队列( 条件变量 | 线程调度 )

mac2025-02-05 16

文章目录

I . 线程简单使用II . 互斥锁III . 条件变量线程同步IV . 完整代码示例006_ThreadSafeQueue.h006_ThreadSafeQueue.cppSafeQueue.hCMakeLists.txt运行结果 V . 示例代码说明

I . 线程简单使用

线程简单使用流程 :

① 线程方法准备 : 定义一个方法 , 主要使用其方法名称和返回值 ;

//线程的主方法 , 类似于 Java 中的 run 方法 , C++ 中方法名随意 void* pushData(void*) { // ... }

② 声明线程 ID : 线程 ID 类型是 pthread_t 类型的 , 其本质是 int 类型 ;

pthread_t pid_push;

③ 创建线程并执行 : pthread_create() 方法时创建并启动线程 ;

//启动一个线程 , 无限循环向线程安全队列中存储数据 pthread_create(&pid_push, 0, pushData, 0);

该方法需要提供四个参数 :

参数 1 ( pthread_t *tidp ) :线程标识符指针 , 该指针指向线程标识符 ;参数 2 ( const pthread_attr_t *attr ) : 线程属性指针 ;参数 3 ( (void*)(*start_rtn)(void*) ) : 线程运行函数指针 , start_rtn 是一个函数指针 , 其参数和返回值类型是 void* 类型参数 4 ( void *arg ) : 参数 3 中的线程运行函数的参数 ;

④ 等待线程执行完毕 : pthread_join (pthread_t thread, void **value_ptr)方法 , 等待 thread 线程 ID 代表的线程执行完毕 ;

//阻塞 , 等待其中任意一个线程执行完毕 , 实际上是一直在此阻塞 , 如果运行下去主函数就暂停了 pthread_join(pid_push, 0);

II . 互斥锁

互斥锁使用流程 :

① 声明互斥锁变量 :

//互斥锁变量 // 1. 先导入头文件 // 2. 定义互斥锁变量 // 3. 在构造函数中进行初始化 // 4. 在析构函数中释放 pthread_mutex_t mutex;

② 初始化互斥锁 :

//初始化互斥锁 pthread_mutex_init(&mutex, 0);

③ 上锁 :

//使用互斥锁将操作锁起来 pthread_mutex_lock(&mutex);

④ 互斥操作 : 需要进行互斥的操作 , 放在上锁与解锁之间进行 ;

⑤ 解锁 :

//解除互斥锁锁定 pthread_mutex_unlock(&mutex);

⑥ 销毁互斥锁 : 互斥锁使用完毕后进行销毁 ;

//释放互斥锁 pthread_mutex_destroy(&mutex);

III . 条件变量线程同步

条件变量使用步骤 :

① 声明条件变量 :

//条件变量 // 使用流程 : // 1. 在构造函数中进行初始化 // 2. 在析构函数中释放 pthread_cond_t cond;

② 初始化条件变量 : 一般在构造函数中执行 ;

//初始化条件变量 pthread_cond_init(&cond, 0);

③ 阻塞线程 :

//阻塞等待 , 相当于 Java 中的 wait() 方法 pthread_cond_wait(&cond, &mutex);

④ 解除线程阻塞 : 有两种方式 , 前者每次只能唤醒一个线程 , 并且无法确定唤醒哪个线程 ; 后者唤醒所有由 cond 条件变量阻塞的线程 ;

//方式 1 : 唤醒一个线程 , 唤醒哪个线程是无法控制的 ; 该方法相当于 Java 中的 notify() pthread_cond_signal(&cond); //方式 2 : 使用广播通知所有等待的线程 , 唤醒所有的线程 , 相当于 Java 中的 notifyAll pthread_cond_broadcast(&cond);

⑤ 销毁条件变量 : 一般在析构函数中进行 ;

//销毁条件变量 pthread_cond_destroy(&cond);

IV . 完整代码示例

006_ThreadSafeQueue.h

// 006_ThreadSafeQueue.h: 标准系统包含文件的包含文件 // 或项目特定的包含文件。 #pragma once #include <iostream> // TODO: 在此处引用程序需要的其他标头。

006_ThreadSafeQueue.cpp

// 005_Thread.cpp: 定义应用程序的入口点。 // #include "006_ThreadSafeQueue.h" #include <pthread.h> //引入队列的头文件 #include <queue> //引入安全队列头文件 #include "SafeQueue.h" using namespace std; //线程安全队列 SafeQueue<int> safeQueue; //向线程安全队列中添加数据 void* pushData(void*) { //循环放入数据 while (true) { int i; //用户从命令行输入数据 , 将该数据 push 到线程安全队列中 cin >> i; safeQueue.push(i); cout << "存储数据到线程安全队列 : " << i << endl; } return 0; } //从线程安全队列中取出数据 void* popData(void*) { //循环取出数据 while (true) { //无限获取数据, 如果线程安全队列中没有数据, 就会在这里阻塞 , 直到 push 进一个数据 , 解除阻塞 int i = 0; //注意传入的是引用 , 可以直接给 i 赋值 , 当做返回值 safeQueue.popAnyway(i); cout << "从线程安全队列中取出出具 : " << i << "\n" << endl; } return 0; } /* 测试线程安全队列 */ int main() { //两个线程 , 一个 push 数据 ( 生产 ) , 一个 pop 数据 ( 消费 ) pthread_t pid_push, pid_pop; //启动一个线程 , 无限循环向线程安全队列中存储数据 pthread_create(&pid_push, 0, pushData, 0); //启动一个线程 , 无限循环向线程安全队列中取出数据 pthread_create(&pid_pop, 0, popData, 0); //阻塞 , 等待其中任意一个线程执行完毕 , 实际上是一直在此阻塞 , 如果运行下去主函数就暂停了 pthread_join(pid_push, 0); system("pause"); return 0; }

SafeQueue.h

//避免被多次 include #pragma once //避免头文件被多次包含 , 有两种处理方式 // ① 一种是 #ifndef A #define A #endif 方式 // ② 另一种就是使用 #pragma once 宏 #include <queue> //引入头文件 , 需要使用互斥锁相关逻辑 #include <pthread.h> using namespace std; //创建一个模板类 , 对 Queue 进行封装 , // 保证该 queue 队列是一个线程安全的队列 // 对 queue 队列操作是线程安全的 template <typename T> class SafeQueue { public : //定义构造函数 SafeQueue() { //初始化互斥锁 pthread_mutex_init(&mutex, 0); //初始化条件变量 pthread_cond_init(&cond, 0); } //定义析构函数 ~SafeQueue() { //释放互斥锁 pthread_mutex_destroy(&mutex); //销毁条件变量 pthread_cond_destroy(&cond); } //向队列中加入元素 , 或从队列中取出元素 // queue 队列不是线程安全的 , 现在要保证该 queue 存储元素是线程安全的 // 需要使用互斥锁控制 push ( 加入元素 ) 和 pop ( 取出元素 ) 操作 ; //向队列中加入元素 void push(T t) { //使用互斥锁将操作锁起来 pthread_mutex_lock(&mutex); //使用互斥锁 , 向队列中加入数据是安全的 safe_queue.push(t); //唤醒一个线程 , 唤醒哪个线程是无法控制的 ; 该方法相当于 Java 中的 notify() //pthread_cond_signal(&cond); //使用广播通知所有等待的线程 , 唤醒所有的线程 , 相当于 Java 中的 notifyAll pthread_cond_broadcast(&cond); //解除互斥锁 pthread_mutex_unlock(&mutex); } /* 现在要实现这样一个需求 : 如果 pop 方法获取时 , 该队列 q 为空 , 此时肯定获取不到数据了但是我们规定每次调用 pop 必须获取一个数据这样的话 , 如果检测到 pop 中没有数据 , 就必须先将线程阻塞等到有新的元素 push 进来后 , 解除阻塞 , 使用条件变量实现 */ //从队列中取出元素 ( 无论如何都要获取到 , 如果获取不到就阻塞到能获取到的时候 ) void popAnyway(T& t) { //使用互斥锁将操作锁起来 pthread_mutex_lock(&mutex); //如果没有数据 , 那么阻塞等待数据 if (safe_queue.empty()) { //阻塞等待 , 相当于 Java 中的 wait() 方法 pthread_cond_wait(&cond, &mutex); } //如果阻塞解除 , 那么执行下面的内容 //t 参数是传入的引用 , 这里可以直接给 t 引用赋值 t = safe_queue.front(); //将首元素移除 safe_queue.pop(); //解除互斥锁 pthread_mutex_unlock(&mutex); } //从队列中取出元素 ( 取数据时要判空 ) void pop(T& t) { //使用互斥锁将操作锁起来 pthread_mutex_lock(&mutex); //使用互斥锁 , 向队列中加入数据是安全的 , 如果队列是空的 , 就获取不到元素 if (!safe_queue.empty()) { //t 参数是传入的引用 , 这里可以直接给 t 引用赋值 t = safe_queue.front(); //将首元素移除 safe_queue.pop(); } //解除互斥锁 pthread_mutex_unlock(&mutex); } private : //实际操作的队列 ( 先进先出 ) , 该队列不是线程安全的 // 如果要保证该 Queue 是线程安全的话 , 就需要为其设置一个互斥锁 // 下面的 mutex 互斥锁变量 , 就是为了保证该队列是线程安全队列而设置的 queue<T> safe_queue; //互斥锁变量 // 1. 先导入头文件 // 2. 定义互斥锁变量 // 3. 在构造函数中进行初始化 // 4. 在析构函数中释放 pthread_mutex_t mutex; //条件变量 // 使用流程 : // 1. 在构造函数中进行初始化 // 2. 在析构函数中释放 pthread_cond_t cond; };

CMakeLists.txt

# CMakeList.txt: 005_Thread 的 CMake 项目，在此处包括源代码并定义 # 项目特定的逻辑。 # cmake_minimum_required (VERSION 3.8) #引入头文件 include_directories("include") #配置自动根据当前是 32 位还是 64 位程序 , 确定静态库的配置目录 if(CMAKE_CL_64) set(platform x64) else() set(platform x86) endif() #配置静态库 , 用于引导如何链接动态库和静态库 link_directories("lib/${platform}") #处理 “timespec”:“struct” 类型重定义报错信息 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DHAVE_STRUCT_TIMESPEC") # 将源代码添加到此项目的可执行文件。 add_executable (006_ThreadSafeQueue "006_ThreadSafeQueue.cpp" "006_ThreadSafeQueue.h") #链接生成的 006_ThreadSafeQueue 和线程动态库名字 # 动态库是 lib/x64 下的 pthreadVC2.lib target_link_libraries(006_ThreadSafeQueue pthreadVC2) # TODO: 如有需要，请添加测试并安装目标。

运行结果

V . 示例代码说明

下载完项目后 , 使用 Visual Studio 打开 , 注意需要配置 POSIX 线程库 ;

【Visual Studio】Visual Studio 2019 社区版 CMakeList 开发环境安装 ( 下载 | 安装相关组件 | 创建编译执行项目 | 错误处理 )

【Visual Studio 2019】创建导入 CMake 项目

【C++ 语言】Visual Studio 配置 POSIX 线程 ( Windows 不支持 POSIX | 配置文件下载 | 库文件说明 | 配置过程 )

最新回复(0)