【C++ 语言】线程安全队列 ( 条件变量 | 线程调度 )
【摘要】
文章目录
I . 线程简单使用II . 互斥锁III . 条件变量 线程同步IV . 完整代码示例006_ThreadSafeQueue.h006_ThreadSafeQueue.cppSafeQ...
I . 线程简单使用
线程简单使用流程 :
① 线程方法准备 : 定义一个方法 , 主要使用其 方法名称 和 返回值 ;
//线程的主方法 , 类似于 Java 中的 run 方法 , C++ 中方法名随意
void* pushData(void*) {
// ...
}
- 1
- 2
- 3
- 4
② 声明线程 ID : 线程 ID 类型是 pthread_t 类型的 , 其本质是 int 类型 ;
pthread_t pid_push;
- 1
③ 创建线程并执行 : pthread_create() 方法时创建并启动线程 ;
//启动一个线程 , 无限循环 向线程安全队列中存储数据
pthread_create(&pid_push, 0, pushData, 0);
- 1
- 2
该方法需要提供四个参数 :
- 参数 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);
- 1
- 2
更多详细内容 ( 如线程属性设置等细节 ) 参考 下面的博客 :
【C++ 语言】线程 ( 线程创建方法 | 线程标识符 | 线程属性 | 线程属性初始化 | 线程属性销毁 | 分离线程 | 线程调度策略 | 线程优先级 | 线程等待 )
【C++ 语言】Visual Studio 配置 POSIX 线程 ( Windows 不支持 POSIX | 配置文件下载 | 库文件说明 | 配置过程 )
II . 互斥锁
互斥锁使用流程 :
① 声明互斥锁变量 :
//互斥锁变量
// 1. 先导入头文件
// 2. 定义互斥锁变量
// 3. 在构造函数中进行初始化
// 4. 在析构函数中释放
pthread_mutex_t mutex;
- 1
- 2
- 3
- 4
- 5
- 6
② 初始化互斥锁 :
//初始化互斥锁
pthread_mutex_init(&mutex, 0);
- 1
- 2
③ 上锁 :
//使用互斥锁将操作锁起来
pthread_mutex_lock(&mutex);
- 1
- 2
④ 互斥操作 : 需要进行互斥的操作 , 放在 上锁 与 解锁之间进行 ;
⑤ 解锁 :
//解除互斥锁 锁定
pthread_mutex_unlock(&mutex);
- 1
- 2
⑥ 销毁互斥锁 : 互斥锁使用完毕后进行销毁 ;
//释放互斥锁
pthread_mutex_destroy(&mutex);
- 1
- 2
III . 条件变量 线程同步
条件变量使用步骤 :
① 声明 条件变量 :
//条件变量
// 使用流程 :
// 1. 在构造函数中进行初始化
// 2. 在析构函数中释放
pthread_cond_t cond;
- 1
- 2
- 3
- 4
- 5
② 初始化 条件变量 : 一般在构造函数中执行 ;
//初始化条件变量
pthread_cond_init(&cond, 0);
- 1
- 2
③ 阻塞线程 :
//阻塞等待 , 相当于 Java 中的 wait() 方法
pthread_cond_wait(&cond, &mutex);
- 1
- 2
④ 解除线程阻塞 : 有两种方式 , 前者每次只能唤醒一个线程 , 并且无法确定唤醒哪个线程 ; 后者唤醒所有由 cond 条件变量阻塞的线程 ;
//方式 1 : 唤醒一个线程 , 唤醒哪个线程 是无法控制的 ; 该方法 相当于 Java 中的 notify()
pthread_cond_signal(&cond);
//方式 2 : 使用广播通知所有等待的线程 , 唤醒所有的线程 , 相当于 Java 中的 notifyAll
pthread_cond_broadcast(&cond);
- 1
- 2
- 3
- 4
- 5
⑤ 销毁 条件变量 : 一般在析构函数中进行 ;
//销毁条件变量
pthread_cond_destroy(&cond);
- 1
- 2
IV . 完整代码示例
006_ThreadSafeQueue.h
// 006_ThreadSafeQueue.h: 标准系统包含文件的包含文件
// 或项目特定的包含文件。
#pragma once
#include <iostream>
// TODO: 在此处引用程序需要的其他标头。
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
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;
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
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;
};
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
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: 如有需要，请添加测试并安装目标。
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
运行结果
V . 示例代码说明
下载完项目后 , 使用 Visual Studio 打开 , 注意需要配置 POSIX 线程库 ;
【Visual Studio】Visual Studio 2019 社区版 CMakeList 开发环境安装 ( 下载 | 安装相关组件 | 创建编译执行项目 | 错误处理 )
【Visual Studio 2019】创建 导入 CMake 项目
【C++ 语言】Visual Studio 配置 POSIX 线程 ( Windows 不支持 POSIX | 配置文件下载 | 库文件说明 | 配置过程 )
文章来源: hanshuliang.blog.csdn.net，作者：韩曙亮，版权归原作者所有，如需转载，请联系作者。
原文链接：hanshuliang.blog.csdn.net/article/details/102851323
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）