工程建设期刊网站,重庆工程信息网官网首页,机床网,域名被墙检测网站文章目录 前言1、Linux线程概念1-1、什么是线程#xff1f;1-1-1、如何看待页表1-1-2、回顾进程地址空间1-1-3、页表怎么进行虚拟地址到物理地址的映射的#xff1f;1-1-4、Linux中线程的概念#xff08;重点#xff09;1-1-5、原生线程库1-1-6、代码测试1-1-7、知识点1-1-1、如何看待页表1-1-2、回顾进程地址空间1-1-3、页表怎么进行虚拟地址到物理地址的映射的1-1-4、Linux中线程的概念重点1-1-5、原生线程库1-1-6、代码测试1-1-7、知识点重点1-1-8、小结1-1-8-1、什么是线程1-1-8-2、线程的优点1-1-8-3、线程的缺点1-1-8-4、线程异常1-1-8-5、线程用途 2、Linux进程VS线程2-1、轻量级进程接口 3、Linux线程控制3-1、POSIX线程库3-2、创建线程3-3、错误检查3-4、创建一批线程3-5、线程终止3-5-1、return返回3-5-2、pthread_exit函数3-5-3、pthread_cancel线程取消 3-6、线程等待3-7、线程执行函数的返回值3-8、语言层的pthread库3-9、线程分离3-9-1、pthread_detach 3-10、pthread库的深入研究3-10-1、用户级线程id3-10-2、线程的局部存储 4、线程封装重点5、Linux线程互斥5-1、进程线程间的互斥相关背景概念5-2、提出问题5-3、解决问题 锁5-3-1、如何理解锁5-3-2、如何理解加锁和解锁的本质5-3-3、对锁进行封装设计 5-4、可重入VS线程安全5-4-1、常见的线程不安全的情况5-4-2、常见的线程安全的情况5-4-3、常见不可重入的情况5-4-4、常见可重入的情况5-4-5、可重入与线程安全联系5-4-6、可重入与线程安全区别5-4-7、小结 6、常见锁概念6-1、死锁6-2、死锁的4个必要条件6-3、避免死锁6-4、避免死锁算法 7、Linux线程同步7-1、同步的概念7-2、条件变量7-3、理解条件变量7-4、条件变量函数 8、生产者消费者模型8-1、举例说明8-2、深入研究8-3、小结8-4、生产消费模型的特点8-5、基于BlockingQueue的生产者消费者模型 9、总结 前言
如何看待之前学习的单进程具有一个线程执行流的进程 Linux中的执行流统称为轻量级进程这里既不是进程也不是线程而是执行流 cache 高速缓存 局部性原理当前正在访问的资源它附近的资源有很大概率被访问到 os先把当前的代码/其他资源加载到cache高速缓存我们cpu来拿取资源是直接从cache里面拿资源的如果拿到了想要的资源就称为命中了继续执行代码即可。如果没有拿到想要的资源就称为未命中cpu就需要再去内存中加载数据从内存中拿取数据也是先缓存到cache里面然后cpu再从cache里面读取 共享资源并不一定是全局的main函数内部的资源可以被新线程使用的main函数内部资源是共享资源
一般而言 创建线程数与cpu的核数有关cpu的核数 —— cpu内部有多少个运算器 创建进程数和cpu个数有关cpu的个数 —— 有独立的多个cpu
cpu由运算器控制器和寄存器构成存储器不是cpu的组成部分 电脑上面显示的cpu多核分两种一种是真多核一种是伪多核 真多核有多个运算器 —— 多个寄存器 —— 一个控制器 伪多核有多个运算器 —— 一套寄存器并没有多数的寄存器 —— 一个控制器 我们本节讲的多线程内容是在linux平台下面的线程内容其他平台由于底层实现多线程的方式不同这里就不深讲了
补充注意本节上半内容代码的cout打印代码建议大家换成printf函数因为cout不是原子性的打印会出现乱序printf通常不是原子性的但是将输出打印到单个流的时候是原子操作如果两个线程使用printf打印到两个流那么还是会出现乱序的情况 要么就加上cout.flush()函数 要想一劳永逸就需要本节后面锁的知识了
1、Linux线程概念
1-1、什么是线程
1-1-1、如何看待页表 1、地址空间是进程能看到的资源窗口内核区栈区堆区… 2、页表页表MMU内存管理单元决定进程真正拥有资源的情况每一个进程都认为自己独有4GB空间可真正拥有多少物理资源由页表映射关系决定 3、合理对 地址空间页表 进行资源划分我们就可以对一个进程的所有资源进行分类进程看到的资源通过地址空间将虚拟内存划分为内核区栈区堆区…通过页表映射到不同的物理内存 1-1-2、回顾进程地址空间 系统调用接口开始就会有汇编指令帮助我们陷入内核这个时候我们的U/K权限就自动切换为K权限然后就可以通过物理地址找到os开始执行系统调用了 str先通过虚拟地址找到了物理地址但是后续的RWX权限str只有R权限而*str H是W权限所以硬件会报错os识别到这个硬件报错就是11号信号段错误然后向进程发送信号进程收到信号之后默认动作就是终止进程 1-1-3、页表怎么进行虚拟地址到物理地址的映射的 就算os把页目录映射的页表的全部都加载进来其实也不大因为我们虚拟地址最后的12个比特位是没有进行映射的它是页内偏移量
1-1-4、Linux中线程的概念重点 线程是进程内的一个执行流 可能很多人看到这里就很懵了这是很多书上的说法那么为什么书上要这么说呢因为操作系统太宏观了有许多的版本不同的平台底层实现多线程的方法不同。而线程是进程内部都一个执行流这句话放到所有的操作系统上面都是对的
所以我们具体就只具体来谈谈linux中的多线程。但是linux中的多线程也是要满足线程是进程内的一个执行流这句话特点
如何看待虚拟内存呢 虚拟内存决定了进程能够看到的“资源” 进程 人虚拟内存 窗口资源 窗外风景
我们以前进fork子进程的时候子进程要拷贝父进程的mm_struct虚拟内存页表以及物理内存写入时要发生写时拷贝 而我们今天可以创建一批“进程”都指向同一个mm_struct虚拟内存 理解 进程所拥有的资源是可以通过 进程地址空间页表 将一部分资源划分给特定的线程的所以单个线程要比之前进程的执行力度更细就像以前的进程一样fork子进程之后通过判断pid是否为0就可以把一段代码块交给子进程执行 如果我们os真的要设计“线程”这样的概念os未来要不要对线程进行管理呢 - 肯定要管理的那么如何管理呢 - 先描述再组织 - 一定要为线程设计专门的数据结构来表示线程对象 - TCB线程控制块本质上也是一个struct结构体 - 再组织 PCB的内部有一张链表通过链表将TCB对象一个个链接起来。当我们执行代码进行调度时先找到进程然后在进程内部跳到指定的线程再进行调度 那么什么平台采用的就是上面对线程管理的方法呢 - windows平台是这么做的 但是我们仔细想想一个线程被创建的根本目的是什么呢 - 是为了被执行然后被调度 - 被调度就要有对应的id状态优先级上下文栈… 单纯从线程的角度来看线程和进程有很多地方是重叠的
所以linux工程师不想给“线程”专门设计对应的数据结构而是直接复用PCB用PCB来表示linux内部的“线程” 线程是进程内的一个执行流 —— 线程在进程内部运行 —— 线程在进程的地址空间内运行拥有该进程的一部分资源 提出问题 1、今天学习了线程概念之后什么叫进程呢 承担分配系统资源的基本实体单位系统进行资源分配的最小单位 —— 简单来说就是 进程要占用系统很多的资源比如IO资源
2、linux中什么叫做线程呢 线程是 cpu调度的最小单位 3、如何看待我们之前学习进程时对应的进程概念呢也就是说我们以前讲的进程内容是不是有问题呢和今天讲的冲突吗
以前的进程概念 承担分配系统资源的实体单位只不过进程内部只有一个执行流而今天 一个进程内部有多个执行流 也就是说cpu执行的PCB都是轻量级进程了哪怕进程只有一个执行流也是一个轻量级进程
结论 1、linux内核中严格意义上来讲是没有真正意义上的线程的 —— linux是用进程PCB来模拟线程的是一种完全属于这就的一套线程方案 2、站在cpu的视角每一个pcb都可以叫做轻量级进程因为对比其他平台和我们以前的进程今天我们学习的PCBtask_struct可能是进程内部的线程也可能就是一个进程执行流所以无论怎么看今天的pcb都以前的pcb我们就称pcb为轻量级进程 3、linux中进程是 承担分配系统资源的基本实体/单位线程是 cpu调度的最小单位 4、进程用来整体申请资源线程用来伸手向进程要资源线程未来malloc等获取资源本质是进程在获取资源因为线程是进程内一部分 5、linux中没有真正意义上的线程 6、这么做的好处是什么 —— 简单维护成本大大降低可靠高效 os只认线程程序员也只认线程都不认你linux所谓的轻量级进程 —— 所以linux无法直接提供创建线程的系统调用接口而只能提供创建轻量级进程的接口
举例 家庭进程 家庭成员线程 社会os
1-1-5、原生线程库
只认线程程序员也只认线程都不认你linux所谓的轻量级进程 linux无法直接提供创建线程的系统调用接口而只能提供创建轻量级进程的接口 所以为了方便用户创建线程程序员编写一个用户级线程库原生线程库pthread库它位于我们用户层和系统调用接口之间我们使用线程库提供的接口该库就会将我们对线程的操作内部转换为对轻量级进程的操作 pthread线程库 在任何linux操作系统下面都有 —— 原生线程库 这里提个问题 pthread库是这么和linux中的轻量级进程建立连接的呢也就是怎么把我们对线程的操作在内部转换为对轻量级进程的操作 1-1-6、代码测试 返回值为0表示创建线程成功
#include iostream
#include pthread.h
#include unistd.h
#include cassertusing std::cin;
using std::cout;
using std::endl;// 这里是 —— 新线程/从线程
void *pthread_handler(void *arr)
{while (1){cout 我是新线程 endl;sleep(1);}
}
int main()
{pthread_t tid;int n pthread_create(tid, nullptr, pthread_handler, (void *)pthread one);assert(n 0);(void)n;// 这里到后面是 —— 主线程while (1){cout 我是主线程;sleep(1);}return 0;
}结果 cpu进行调度的时候是通过LWP这个id来表示特定的一个执行流的而不是pid
而我们以前使用pid也是没有问题的
因为以前我们只有一个进程一个执行流所以pid等价于lwp 我们上面说了pthread_create的第4个参数是作用于第3个参数的我们来验证一下
void *pthread_handler(void *arr)
{const char *ptr (const char *)arr;while (1){cout 我是新线程name : ptr endl;sleep(1);}
}接下来我们来看一下线程的tid是多少 我们将tid转换为16进制然后打印出来
while (1){char tidbuffer[1024];snprintf(tidbuffer,sizeof(tidbuffer),0x%x,tid);//格式化函数 —— 将tid转化为0x%x十六进制的类型放到tidbuffer里面cout 我是主线程!我的tid是 : tidbuffer endl;sleep(1);fflush(stdout);}什么tid这么长的一串数字看起来很像一个地址其实它就是一个地址不过我们目前还讲不清楚这个地址到底是什么后面才能讲清楚
1-1-7、知识点重点
1、线程一旦被创建几乎所有的资源都是被所有线程所共享的
#include iostream
#include pthread.h
#include unistd.h
#include cassertusing std::cin;
using std::cout;
using std::endl;int g_val 0;
// 这里是 —— 新线程/从线程
void *pthread_handler(void *arr)
{const char *name (const char *)arr;while (1){// cout 我是新线程!name : ptr endl;cout 我是新线程, 我正在运行! name: name : g_val g_val : g_val endl;sleep(1);fflush(stdout);}
}
int main()
{pthread_t tid;int n pthread_create(tid, nullptr, pthread_handler, (void *)pthread one);assert(n 0);(void)n;// 这里到后面是 —— 主线程while (1){char tidbuffer[1024];snprintf(tidbuffer, sizeof(tidbuffer), 0x%x, tid); // 格式化函数 —— 将tid转化为0x%x十六进制的类型放到tidbuffer里面// cout 我是主线程!我的tid是 : tidbuffer endl;cout 我是主线程, 我正在运行!, 我创建出来的线程的tid: tidbuffer : g_val g_val : g_val endl;sleep(1);fflush(stdout);}return 0;
}2、线程也有自己的私有资源那么有哪一些资源是线程私有的呢面试 1、pcb的属性私有线程要被调度所以它的优先级id等等都要私有 2、线程的上下文结构私有(线程也可能会被切换如果线程没有执行完就要先保存该线程的上下文结构方便恢复重新执行该线程) 3、每一个线程都要有独立的栈结构线程都有对应的代码执行时都要形成对应的局部变量而这些局部变量都是存在栈上面的 线程A在堆区上new了一块空间如果将这个堆空间地址保存在全局变量指针中任何线程都能够访问该堆空间 只不过如果线程A new空间的地址没有保存在全局变量中那么该地址就在该线程的栈上面而栈是独立的使用就给人一种感觉堆空间是该线程私有的 线程内的局部变量也是同理只要把线程内的局部变量的地址给给全局变量指针那么其他线程就都能通过全局变量指针拿到该线程的局部变量 全局变量就更不用说了任何线程可以直接拿到 这里再提出一个问题我们在怎么保证每一个线程都有自己的栈结构呢
3、与进程之间的切换相比线程之间的切换要os做的工作要少得多 进程切换 切换页表 —— 切换虚拟地址空间 —— 切换上下文 —— 切换pcb 线程切换切换上下文 —— 切换pcb 线程切换cache高速缓存不用太更新而进程切换cache要全部更新主要的工作 cpu内部有一个cache高速缓存它比cpu其他硬件效率慢一点但是比内存的效率快很多 局部性原理当前正在访问的资源它附近的资源有很大概率被访问到 os先把当前的代码/其他资源加载到cache高速缓存我们cpu来拿取资源是直接从cache里面拿资源的如果拿到了想要的资源就称为命中了继续执行代码即可。如果没有拿到想要的资源就称为未命中cpu就需要再去内存中加载数据从内存中拿取数据也是先缓存到cache里面然后cpu再从cache里面读取 热点数据 cache中被进程多次访问被进程较高概率命中的数据是需要跑一段时间才能知道的 所以cache里面的热点数据的加载才是线程为什么切换比进程切换做的工作要少的原因
4、计算密集型应用和I/O密集型应用 计算密集型应用主要是进程/线程使用的资源是 cpu的资源加密解密算法等等 —— 打包压缩执行打包和压缩对应的算法 I/O密集型应用主要是进程/线程使用的资源是 外设的资源访问磁盘访问网络显示器等等 —— 抖音迅雷 5、多线程健壮性问题 简单来说多个线程共享大部分数据而一个线程在使用某个数据的时候另一个线程可能正在对该数据进行修改 —— 一个线程影响到了另外一个线程缺乏访问控制 我们先来看看正常情况的代码
#include iostream
#include string
#include pthread.h
#include unistd.h
using namespace std;void *start_routine(void *argv)
{string name static_castconst char *(argv);while (1){cout thread name is : name endl;sleep(1);}
}int main()
{pthread_t id;pthread_create(id, nullptr, start_routine, (void *)thread 1); // 最后一个参数要强转while (1){cout thread name is :main thread endl;sleep(1);}return 0;
}所以一个线程出了异常会影响另一个线程 而这种线程就叫做健壮性或者鲁棒性较差
为什么会出现一个线程出了异常会影响另一个线程的情况呢
以前在我们看来一个进程内部产生的线程因为该线程出现异常被os捕捉到了硬件异常软件条件终端按键系统调用等方法产生信号os捕捉到异常之后通过pid给每一个线程/进程发送信号不是LWP。也就是说每一个pid相同的轻量级进程都收到了os发送的信号线程的pid都是相同的。然后每一个线程执行信号的默认动作终止执行流那么所有的线程就都被终止了进程也就被直接终止了
今天在我们看来一个进程内部产生的线程线程是进程内的一个执行流也就是进程内的一部分线程做任何事情就代表进程在做该事情那么一旦线程出问题了进程也就出问题了线程和进程应该是一个整体 一个线程出问题了那么线程所属进程就出问题了线程和进程是一个整体进程被创建os需要分配对应的资源而进程被销毁os要释放进程的资源。而线程的资源是有进程来分配的所有进程被终止了线程也就自动销毁了
1-1-8、小结
1-1-8-1、什么是线程 1、在一个程序里的一个执行路线就叫做线程thread。更准确的定义是线程是“一个进程内部的控制序列” 2、一切进程至少都有一个执行线程 3、线程在进程内部运行本质是在进程地址空间内运行 4、在Linux系统中在CPU眼中看到的PCB都要比传统的进程更加轻量化 5、透过进程虚拟地址空间可以看到进程的大部分资源将进程资源合理分配给每个执行流就形成了线程执行流 1-1-8-2、线程的优点 1、创建一个新线程的代价要比创建一个新进程小得多对比于进程 2、与进程之间的切换相比线程之间的切换需要操作系统做的工作要少很多对比于进程 3、线程占用的资源要比进程少很多对比于进程 4、能充分利用多处理器的可并行数量—— 从这里开始下面的优点进程也有 5、在等待慢速I/O操作结束的同时程序可执行其他的计算任务 6、计算密集型应用为了能在多处理器系统上运行将计算分解到多个线程中实现 7、I/O密集型应用为了提高性能将I/O操作重叠。线程可以同时等待不同的I/O操作。 1-1-8-3、线程的缺点 1、性能损失 一个很少被外部事件阻塞的计算密集型线程往往无法与共它线程共享同一个处理器。如果计算密集型线程的数量比可用的处理器多那么可能会有较大的性能损失这里的性能损失指的是增加了额外的同步和调度开销而可用的资源不变。 2、健壮性降低 编写多线程需要更全面更深入的考虑在一个多线程程序里因时间分配上的细微偏差或者因共享了不该共享的变量而造成不良影响的可能性是很大的换句话说线程之间是缺乏保护的。一个线程异常了可能会影响到其他线程 3、缺乏访问控制 进程是访问控制的基本粒度在一个线程中调用某些OS函数会对整个进程造成影响。 4、编程难度提高 编写与调试一个多线程程序比单线程程序困难得多 1-1-8-4、线程异常 1、单个线程如果出现除零野指针问题导致线程崩溃进程也会随着崩溃 2、线程是进程的执行分支线程出异常就类似进程出异常进而触发信号机制终止进程进程终止该进程内的所有线程也就随即退出 1-1-8-5、线程用途 1、合理的使用多线程能提高CPU密集型程序的执行效率 2、合理的使用多线程能提高IO密集型程序的用户体验如生活中我们一边写代码一边下载开发工具就是多线程运行的一种表现 2、Linux进程VS线程
进程是资源分配的基本单位 线程是调度的基本单位 线程共享进程数据但也拥有自己的一部分数据 1、 线程ID 2、一组寄存器 3、栈 4、errno 5、信号屏蔽字 6、调度优先级
进程的多个线程共享 同一地址空间,因此Text Segment代码段、Data Segment数据段都是共享的,如果定义一个函数,在各线程中都可以调用,如果定义一个全局变量,在各线程中都可以访问到,除此之外,各线程还共享以下进程资源和环境: 1、文件描述符表 2、每种信号的处理方式(SIG_ IGN、SIG_ DFL或者自定义的信号处理函数) 3、当前工作目录 4、用户id和组id
进程线程关系图4种模型
2-1、轻量级进程接口
我们前面说了linux只提供了创建轻量级进程的接口那么这个轻量级进程接口是什么呢 fork和vfork等函数的底层调用的就是这个clone函数ptrhead_create - clone 我们不用clone这个函数而是对应的原生线程库在用这个函数
3、Linux线程控制
3-1、POSIX线程库 1、与线程有关的函数构成了一个完整的系列绝大多数函数的名字都是以“pthread_”打头的 2、要使用这些函数库要通过引入头文pthread.h 3、链接这些线程函数库时要使用编译器命令的“-lpthread”选项 3-2、创建线程
功能创建一个新的线程
原型int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void * (*start_routine)(void*), void *arg);参数thread:返回线程IDattr:设置线程的属性attr为NULL表示使用默认属性start_routine:是个函数地址线程启动后要执行的函数arg:传给线程启动函数的参数返回值成功返回0失败返回错误码3-3、错误检查 传统的一些函数是成功返回0失败返回-1并且对全局变量errno赋值以指示错误。 pthreads函数出错时不会设置全局变量errno而大部分其他POSIX函数会这样做。而是将错误代码通过返回值返回 pthreads同样也提供了线程内的errno变量以支持其它使用errno的代码。对于pthreads函数的错误建议通过返回值业判定因为读取返回值要比读取线程内的errno变量的开销更小 绝大部分线程的错误原因是由返回值来告诉我们的
线程ID及进程地址空间布局 1、pthread_ create函数会产生一个线程ID存放在第一个参数指向的地址中。该线程ID和前面说的线程ID不是一回事。 2、前面讲的线程ID属于进程调度的范畴。因为线程是轻量级进程是操作系统调度器的最小单位所以需要一个数值来唯一表示该线程。 3、pthread_ create函数第一个参数指向一个虚拟内存单元该内存单元的地址即为新创建线程的线程ID属于NPTL线程库的范畴。线程库的后续操作就是根据该线程ID来操作线程的。 4、线程库NPTL提供了pthread_ self函数可以获得线程自身的ID pthread_t pthread_self(void); pthread_t 到底是什么类型呢取决于实现。对于Linux目前实现的NPTL实现而言pthread_t类型的线程ID本质就是一个进程地址空间上的一个地址
我们下面的3-10会讲
3-4、创建一批线程
一次性创建10个线程
#include iostream
#include string
#include vector
#include pthread.h
#include unistd.h
using namespace std;void *start_routine(void *argv)
{string name static_castconst char *(argv);while (1){cout thread name is : name endl;sleep(1);}
}int main()
{vectorpthread_t vp;
#define NUM 10for (size_t i 0; i NUM; i){pthread_t tid;pthread_create(tid, nullptr, start_routine, (void *)thread 1);}while (1){cout thread name is :main thread endl;sleep(1);}return 0;
}在这10个线程的基础上面让每一个线程都打印自己的编号
#include iostream
#include string
#include vector
#include pthread.h
#include unistd.h
using namespace std;void *start_routine(void *argv)
{string name static_castconst char *(argv);while (true){cout thread name is : name endl;sleep(1);}
}int main()
{vectorpthread_t vp;
#define NUM 10for (size_t i 0; i NUM; i){pthread_t tid;char namebuffer[64];snprintf(namebuffer, sizeof(namebuffer), %s:%d, thread, i);// pthread_create(tid, nullptr, start_routine, (void *)thread 1);pthread_create(tid, nullptr, start_routine, namebuffer);sleep(1);}while (true){cout thread name is :main thread endl;sleep(1);}return 0;
}改进
#include iostream
#include string
#include vector
#include pthread.h
#include unistd.h
using namespace std;class ThreadDate
{
public:pthread_t tid;char namebuffer[64];
};void *start_routine(void *argv) // 拷贝td形参是实参的一份临时拷贝
{ThreadDate *name static_castThreadDate *(argv);int cnt 10;while (cnt){cout thread name is : name-namebuffer cnt: cnt-- endl;sleep(1);}delete name;return nullptr;
}int main()
{vectorpthread_t vp;
#define NUM 10for (size_t i 0; i NUM; i){ThreadDate *td new ThreadDate();snprintf(td-namebuffer, sizeof(td-namebuffer), %s:%d, thread, i);pthread_create(td-tid, nullptr, start_routine, td); // 这里传td}while (true){cout main thread endl;sleep(1);}return 0;
}所以start_routine函数要被10个线程调用这个函数现在就是重入状态是一个可重入函数
在函数内定义的变量都叫做局部变量具有临时性 —— 在多线程的情况下也没有问题 —— 每一个线程都有独立的栈结构
3-5、线程终止
3-5-1、return返回
线程函数调用完毕return的时候就终止了
3-5-2、pthread_exit函数 需要注意,pthread_exit或者return返回的指针所指向的内存单元必须是全局的或者是用malloc分配的,不能在线程函数的栈上分配,因为当其它线程得到这个返回指针时线程函数已经退出了 任何一个执行流调用exit都会让整个进程退出
pthread_exit函数只终止线程不终止进程
3-5-3、pthread_cancel线程取消
线程是可以被cancel取消的 前提线程要运行起来了才能被取消 一个线程被取消它的退出码就是-1
3-6、线程等待
线程也是要被等待的—— 如果不等待会造成类似于僵尸进程的问题 —— 内存泄漏 为什么要线程等待 1、获取新线程/从线程的退出信息可以不关心但是必须要回收线程资源 2、创建新的线程不会复用刚才退出线程的地址空间回收新线程对应的pcb等内核资源防止泄露 功能等待线程结束
原型int pthread_join(pthread_t thread, void **value_ptr);
参数thread:线程IDvalue_ptr:它指向一个指针后者指向线程的返回值
返回值成功返回0失败返回错误码调用该函数的线程将挂起等待,直到id为thread的线程终止。thread线程以不同的方法终止,通过pthread_join得到的终止状态是不同的总结如下: 如果thread线程通过return返回,value_ ptr所指向的单元里存放的是thread线程函数的返回值。如果thread线程被别的线程调用pthread_ cancel异常终掉,value_ ptr所指向的单元里存放的是常数PTHREAD_ CANCELED。如果thread线程是自己调用pthread_exit终止的,value_ptr所指向的单元存放的是传给pthread_exit的参数。如果对thread线程的终止状态不感兴趣,可以传NULL给value_ ptr参数。 #include iostream
#include string
#include vector
#include pthread.h
#include unistd.h
#include cstdlib
#include cassert
using namespace std;class ThreadDate
{
public:pthread_t tid;char namebuffer[64];
};void *start_routine(void *argv) // 拷贝td形参是实参的一份临时拷贝
{// sleep(1);ThreadDate *td static_castThreadDate *(argv);int cnt 10;while (cnt){// cout thread name is : td-namebuffer cnt: cnt-- endl;cout cnt : cnt-- cnt : cnt endl;sleep(1);// return nullptr;// pthread_exit(nullptr);}//delete td;return nullptr;
}int main()
{vectorThreadDate * vp;
#define NUM 10for (size_t i 0; i NUM; i){ThreadDate *td new ThreadDate();snprintf(td-namebuffer, sizeof(td-namebuffer), %s:%d, thread, i);pthread_create(td-tid, nullptr, start_routine, td); // 这里传tdvp.push_back(td);}for (auto iter : vp){cout create thread: iter-namebuffer : iter-tid suceesss endl;}for (auto iter : vp){int n pthread_join(iter-tid, nullptr);assert(n 0);cout join: iter-namebuffer suceesss endl;delete iter;}cout join end endl;return 0;
} 所以线程是可以等待的等待的时候是join等待的 —— 阻塞式等待
3-7、线程执行函数的返回值 一句话 pthread_join函数的第二个参数就是线程执行函数的返回结果 因为用了pthread库所以我们不能直接从pthread库里面拿到返回值要使用pthread_join函数来拿返回值
void * ret nullptr void * 106
void ** p ret
*p 106 —— *p就是retreturn void*106 将106强转为指针类型了而106就是一个指针地址4字节数据填到8字节地址中。 pthread库里面存放void * 106这个地址 ret是一个void的地址里面存放一个8字节的地址 ret就是取出一个指针变量的地址(ret) —— 就拿到了ret这个指针变量也就是void * ret然后将void * 106赋值给ret这样我们就将线程执行函数的返回值拿到pthread_join函数的第二个参数里面了 返回值会被拷贝一份
为什么没有见到线程退出时候对应的退出信号呢 线程出异常收到信号整个进程都会退出
pthread_join默认函数会调用成功。不考虑异常问题异常问题是进程考虑的
3-8、语言层的pthread库 #include iostream
#include unistd.h
#include thread
using std::cout;
using std::endl;
void thread_run()
{while (true){cout 我是新线程... endl;sleep(1);}
}
int main()
{std::thread t1(thread_run);while (true){cout 我是主线程... endl;sleep(1);}t1.join();return 0;
}任何语言在linux中如果要实现多线程必定要是用pthread库。如何看待C11中的多线程呢C11 的多线程在Linux环境中本质是对pthread库的封装
pthread线程库意义给语言级别的线程提供底层的接口支持
3-9、线程分离
我们上面学到了主线程调用pthread_join函数进行阻塞式等待从线程而线程是没有非阻塞等待的 默认情况下新创建的线程是joinable的线程退出后需要对其进行pthread_join操作否则无法释放资源从而造成系统泄漏 如果不关心线程的返回值join是一种负担这个时候我们可以告诉系统当线程退出时自动释放线程资源 int pthread_detach(pthread_t thread);可以是线程组内其他线程对目标线程进行分离也可以是线程自己分离: pthread_detach(pthread_self());joinable和分离是冲突的一个线程不能既是joinable又是分离的
我们来看看正常情况下的代码
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
using std::cout;
using std::endl;void *start_toutine(void *args)
{std::string name static_castconst char *(args);while (1){//这里和上面不同采用printf向一个流(stdout)打印结果是原子性的printf(%s\n, name.c_str());//字符串要记得转换称为c类型的头文件cstringsleep(1);}
}
int main()
{pthread_t tid;pthread_create(tid, nullptr, start_toutine, (void *)thread 1);pthread_join(tid, nullptr);//进行等待return 0;
}我们线程可以通过主线程分离指定线程也可以新线程分离自己不需要进行join等待 那么我们对上面的代码进行加工打印出线程的tid
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
using std::cout;
using std::endl;std::string ThreadId(const pthread_t pthtred_id)
{char buffer[128];snprintf(buffer, sizeof(buffer), 0x%x, pthread_self());return buffer;
}
void *start_toutine(void *args)
{std::string name static_castconst char *(args);while (1){// 这里pthread_self获取到的id必须和下面主函数ThreadId获取到的id是一样的printf(%s running... %s\n, name.c_str(), ThreadId(pthread_self()).c_str());sleep(1);}
}
int main()
{pthread_t tid;pthread_create(tid, nullptr, start_toutine, (void *)thread 1);printf(main thread running... new thread id : %s\n, ThreadId(tid).c_str());pthread_join(tid, nullptr);return 0;
}接下来我们就来进行线程分离
3-9-1、pthread_detach 先来看看pthread_jion等待值的情况
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
using std::cout;
using std::endl;std::string ThreadId(const pthread_t pthtred_id)
{char buffer[128];snprintf(buffer, sizeof(buffer), 0x%x, pthread_self());return buffer;
}
void *start_toutine(void *args)
{std::string name static_castconst char *(args);// pthread_detach(pthread_self()); // 线程将自己分离int cnt 5;while (cnt--)//先不分离线程先检查5s之后线程等待的值是不是0{printf(%s running... %s\n, name.c_str(), ThreadId(pthread_self()).c_str());sleep(1);}return nullptr;
}
int main()
{pthread_t tid;pthread_create(tid, nullptr, start_toutine, (void *)thread 1);std::string main_tid ThreadId(pthread_self());printf(main thread running... new thread id : %s —— main threead id : %s\n, ThreadId(tid).c_str(), main_tid.c_str());// 一个线程创建出来默认是joinable的如果设置了分离状态就不能再对该线程进行等待了int n pthread_join(tid, nullptr);printf(result : %d : %s\n, n, strerror(n));//5s之后n应该是0return 0;
}线程分离代码
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
using std::cout;
using std::endl;
std::string ThreadId(const pthread_t pthtred_id)
{char buffer[128];snprintf(buffer, sizeof(buffer), 0x%x, pthread_self());return buffer;
}
void *start_toutine(void *args)
{std::string name static_castconst char *(args);int cnt 5;while (cnt--){printf(%s running... %s\n, name.c_str(), ThreadId(pthread_self()).c_str());sleep(1);}return nullptr;
}
int main()
{pthread_t tid;pthread_create(tid, nullptr, start_toutine, (void *)thread 1);std::string main_tid ThreadId(pthread_self());pthread_detach(tid);printf(main thread running... new thread id : %s —— main threead id : %s\n, ThreadId(tid).c_str(), main_tid.c_str());while (1){//主线程做自己的事情...printf(result : %d : %s\n, n, strerror(n));sleep(1);}return 0;
}3-10、pthread库的深入研究 我们使用了pthread库原生线程库创建了线程那么其他人也可以使用这个pthread库创建线程其他人在我们电脑上面创建他们所需要的线程 —— 原生线程库存在多个线程 那么原生线程库就要对线程做管理不然怎么知道线程对应的id栈在哪里大小是多少… 管理 —— 先描述再组织 —— 描述就是线程的属性只不过属性比较少线程id值栈区的地址等等 每一个线程都要有对应的属性pthread要管理好每一个线程 3-10-1、用户级线程id 我们通过线程的id也就是线程在库里面的起始地址向后就能找到线程在库里面的资源了 所以我们拿着线程id就能够对线程进行各种操作了 所以新线程的栈都是独立的存在共享区当中 主线程用的是地址空间中的栈而新线程用的是共享区中pthread原生线程库里面的栈
3-10-2、线程的局部存储
我们上面看到了每一个线程在库中都有一个线程局部存储
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
using std::cout;
using std::endl;int g_val 100;std::string ThreadId(const pthread_t pthtred_id)
{char buffer[128];snprintf(buffer, sizeof(buffer), 0x%x, pthread_self());return buffer;
}
void *start_toutine(void *args)
{std::string name static_castconst char *(args);while (1){printf( new thread : g_val : %d g_val : %p\n, g_val, g_val);sleep(1);g_val;}return nullptr;
}
int main()
{pthread_t tid;pthread_create(tid, nullptr, start_toutine, (void *)thread 1);std::string main_tid ThreadId(pthread_self());pthread_detach(tid);while (1){printf(main thread : g_val : %d g_val : %p\n, g_val, g_val);sleep(1);}return 0;
}添加__thred可以将一个内置类型int,double…设置为线程局部存储 4、线程封装重点
我们接下来就对linux中的轻量级进程进行封装从而达到C中直接使用线程的效果
我们这里实现的就简单一点能达到C中直接创建线程线程调用函数然后join等待线程就行了
Thread.hpp:
#pragma once#include iostream
#include pthread.h
#include string
#include cstring
#include functional
#include cassertclass Thread;class Context // 上下文将pthread_create中类内的第4个参数和this合并
{
public:Context(): _this(nullptr),_args(nullptr){}~Context(){}Thread *_this; // 调用函数的this指针pthread_create创建线程之后线程执行函数的参数的thisvoid *_args; // 保存线程执行函数的参数
};class Thread
{
public:// using func_t std::functionvoid*(void*);作用同下typedef std::functionvoid *(void *) func_t;const int num 1024;//这样改就可以像C一样直接构造一个线程然后传线程执行函数就行不需要传线程函数参数和线程编号了//Thread(func_t func, void *args nullptr, int number 0)Thread(func_t func, void *args, int number): _func(func),_args(args){// _name thread : ;// _name std::to_string(number);//和下面snprintf作用相同char buffer[num];snprintf(buffer, sizeof buffer, thread : %d, number);_name buffer;//void start()//可以直接把start函数拿进来//{Context *cnt new Context();cnt-_args _args;cnt-_this this;// 这里直接把cnt进行传参直接把start_routine的参数和this指针都传过去了int n pthread_create(_tid, nullptr, start_routine, cnt); // 这里_func会报错C接口不能直接掉C的函数对象assert(0 n); // 线程创建成功函数返回值为0(void)n; // 编译器在release版本会注释掉assert。会发现我们没有使用n有的编译器就会报存在未使用变量n的警告我们这里取消这个警告//}}// 在类内创建线程想让线程执行对应的方法需要将方法设置为static(静态方法) —— 因为static类内函数没有this指针static void *start_routine(void *args) // 写一个函数方便我们下面pthread_create第3个参数使用{// 很不幸下面还是不能直接使用start_routine函数因为start_routine是类内函数有缺少参数// 也就是说start_routine有两个参数第一个参数是Thread* this指针第二个参数才是args// return _func(args);// 这里就又出问题了静态函数只能调用静态方法和静态成员不能调用类内成员方法和成员变量// 所以得换一种写法Context *cnt static_castContext *(args);void *ret cnt-_this-run(cnt-_args); // 这里调用下面的run函数delete cnt;return ret;}// void start()//这里把start放外面调用的时候要让线程调用start// {// Context *cnt new Context();// cnt-_args _args;// cnt-_this this;// // 这里直接把cnt进行传参直接把start_routine的参数和this指针都传过去了// int n pthread_create(_tid, nullptr, start_routine, cnt); // 这里_func会报错C接口不能直接掉C的函数对象// assert(0 n); // 线程创建成功函数返回值为0// (void)n; // 编译器在release版本会注释掉assert。会发现我们没有使用n有的编译器就会报存在未使用变量n的警告我们这里取消这个警告// }void join(){int n pthread_join(_tid, nullptr);assert(n 0);(void)n;printf(%s\n, strerror(n));}void *run(void *args) // 给上面start_routine来用的{return _func(args);}~Thread(){}private:std::string _name; // 我们想直接看线程名字比如线程1线程2这种pthread_t _tid;func_t _func; // 线程未来执行的函数void *_args; // 线程执行函数的参数
};
mythread.cpp:
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
#include memory //智能指针#include Thread.hppusing std::cout;
using std::endl;void *thread_run(void *args) // 线程执行函数
{std::string work_type static_castconst char *(args);while (1){printf(我是一个新线程,我正在做 : %s\n, work_type.c_str());sleep(1);}
}int main()
{std::unique_ptrThread thread1(new Thread(thread_run, (void *)thread 1, 1));std::unique_ptrThread thread2(new Thread(thread_run, (void *)thread 2, 2));std::unique_ptrThread thread3(new Thread(thread_run, (void *)thread 3, 3));// thread1-start();//如果把Thread.hpp的start放到构造函数外面这里就要调start函数// thread2-start();// thread3-start();thread1-join();thread2-join();thread3-join();return 0;
}文章目录 前言1、Linux线程概念1-1、什么是线程1-1-1、如何看待页表1-1-2、回顾进程地址空间1-1-3、页表怎么进行虚拟地址到物理地址的映射的1-1-4、Linux中线程的概念重点1-1-5、原生线程库1-1-6、代码测试1-1-7、知识点重点1-1-8、小结1-1-8-1、什么是线程1-1-8-2、线程的优点1-1-8-3、线程的缺点1-1-8-4、线程异常1-1-8-5、线程用途 2、Linux进程VS线程2-1、轻量级进程接口 3、Linux线程控制3-1、POSIX线程库3-2、创建线程3-3、错误检查3-4、创建一批线程3-5、线程终止3-5-1、return返回3-5-2、pthread_exit函数3-5-3、pthread_cancel线程取消 3-6、线程等待3-7、线程执行函数的返回值3-8、语言层的pthread库3-9、线程分离3-9-1、pthread_detach 3-10、pthread库的深入研究3-10-1、用户级线程id3-10-2、线程的局部存储 4、线程封装重点5、Linux线程互斥5-1、进程线程间的互斥相关背景概念5-2、提出问题5-3、解决问题 锁5-3-1、如何理解锁5-3-2、如何理解加锁和解锁的本质5-3-3、对锁进行封装设计 5-4、可重入VS线程安全5-4-1、常见的线程不安全的情况5-4-2、常见的线程安全的情况5-4-3、常见不可重入的情况5-4-4、常见可重入的情况5-4-5、可重入与线程安全联系5-4-6、可重入与线程安全区别5-4-7、小结 6、常见锁概念6-1、死锁6-2、死锁的4个必要条件6-3、避免死锁6-4、避免死锁算法 7、Linux线程同步7-1、同步的概念7-2、条件变量7-3、理解条件变量7-4、条件变量函数 8、生产者消费者模型8-1、举例说明8-2、深入研究8-3、小结8-4、生产消费模型的特点8-5、基于BlockingQueue的生产者消费者模型 9、总结 5、Linux线程互斥
5-1、进程线程间的互斥相关背景概念 临界资源多线程执行流共享的资源就叫做临界资源 临界区每个线程内部访问临界资源的代码就叫做临界区 互斥任何时刻互斥保证有且只有一个执行流进入临界区访问临界资源通常对临界资源起保护作用 原子性后面讨论如何实现不会被任何调度机制打断的操作该操作只有两态要么完成要么未完成 5-2、提出问题
我们前面知道了全局变量是被所有线程所共享的那么我们多个线程对这一个全局变量进行–操作会发生什么呢
这里我们要看到的现象是 —— 票有可能被抢为负数 我们需要让多个线程进行并行交叉执行usleep休眠切换进程 多个线程并行交叉执行的本质cpu内的调度器频繁的发生线程进行切换和调度 线程一般什么时候发生切换呢 —— 1、时间片到了2、来了更高优先级的线程3、线程等待的时候 线程什么时候检测时间片到了来了优先级更高的线程线程等待这些问题呢 线程从内核态返回用户态的时候线程要对调度状态进行检测如果可以就直接发生线程切换
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
#include memory
#include Thread.hpp
using std::cout;
using std::endl;// 这里我们要看到的现象是 —— 票有可能被抢为负数
// 我们需要让多个线程进行并行交叉执行usleep休眠切换进程
// 多个线程并行交叉执行的本质cpu内的调度器频繁的发生线程进行切换和调度
// 线程一般什么时候发生切换呢 —— 1、时间片到了2、来了更高优先级的线程3、线程等待的时候
// 线程什么时候检测时间片到了来了优先级更高的线程线程等待这些问题呢
// 线程从内核态返回用户态的时候线程要对调度状态进行检测如果可以就直接发生线程切换int train_tickets 1000; // 火车票
void *get_train_tickets(void *args)
{std::string user_name static_castconst char *(args);while (1){if (train_tickets 0) // 有票才能抢{// 用这段时间来模拟抢票真实需要花费的时间usleep(1000); // 1s 1000毫秒 1000 000微秒 1000 000 000纳秒printf(%s正在进行抢票 : %d\n, user_name.c_str(), train_tickets);train_tickets--;}elsebreak;}
}
int main()
{std::unique_ptrThread thread1(new Thread(get_train_tickets, (void *)user 1, 1));std::unique_ptrThread thread2(new Thread(get_train_tickets, (void *)user 2, 2));std::unique_ptrThread thread3(new Thread(get_train_tickets, (void *)user 3, 3));thread1-join();thread2-join();thread3-join();return 0;
}为什么会出现抢到负数这种情况呢 所以最终票被抢为了负数
对一个全局变量进行多线程修改是安全的吗 答案是会的! 哪怕只有一个主线程一个从线程都是不安全的除非进程内部只有一个执行流 所以我们上面抢票的代码就是没有if判断语句只有train_tickets–这个操作抢票也会抢到负数只不过因为cpu太高效了我们不好模拟出来罢了 我们定义的全局变量在没有保护的情况下往往是不安全的 像上面多个线程并行执行造成的数据安全问题 —— 我们称之为数据不一致问题 那么接下来我们就要提出解决数据不一致问题的方案了
要解决以上问题需要做到三点 1、代码必须要有互斥行为当代码进入临界区执行时不允许其他线程进入该临界区。 2、如果多个线程同时要求执行临界区的代码并且临界区没有线程在执行那么只能允许一个线程进入该临界区。 3、如果线程不在临界区中执行那么该线程不能阻止其他线程进入临界区 要做到这三点本质上就是需要一把锁。Linux上提供的这把锁叫互斥量
5-3、解决问题 锁 1、多个执行流进行安全访问的共享资源我们称之为 —— 临界资源上面的共享资源tickets不是安全的所以不是临界资源 2、我们把多个执行流中访问临界资源的代码称之为 —— 临界区 —— 往往是线程代码很小的一部分 3、想让多个线程串形访问共享资源 —— 互斥上面3个线程抢票一个一个的按顺序来不能并行抢票 4、对一个资源进行操作的时候要么不做要么做完 —— 原子性 对资源进行操作如果只有一条汇编指令那么该操作就是原子的这种情况是原子性的一小部分 所以经过上面的总结我们提出的解决方案就是加锁 pthread_mutex_init()函数 功能初始化一个互斥锁 pthread_mutex_destroy()函数 功能销毁一个互斥锁 pthread_mutex_lock()函数 功能加锁 pthread_mutex_trylock()函数 功能尝试加锁 pthread_mutex_unlock()函数 功能解锁 以上5个函数的返回值都是成功返回0 失败返回错误号。 pthread_mutex_t 类型其本质是一个结构体。为简化理解应用时可忽略其实现细节简单当成整数看待。如 pthread_mutex_t mutex; 变量mutex只有两种取值1、0。 我们定义锁可以定义为全局的也可以定义为局部的 但是如果锁是局部的我们要调用pthread_mutex_init()和pthread_mutex_destroy()对锁进行初始化和销毁 如果锁的全局的pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;//全局锁直接使用PTHREAD_MUTEX_INITIALIZER初始化就行
我们见见猪跑
#include iostream
#include pthread.h
#include string
#include unistd.h
#include cstring
#include cstdio
#include memory
#include Thread.hpp
using std::cout;
using std::endl;pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;//全局锁直接使用PTHREAD_MUTEX_INITIALIZER初始化就行int train_tickets 1000;
void *get_train_tickets(void *args)
{std::string user_name static_castconst char *(args);while (1){pthread_mutex_lock(mutex);if (train_tickets 0) {usleep(1000); printf(%s正在进行抢票 : %d\n, user_name.c_str(), train_tickets);train_tickets--;pthread_mutex_unlock(mutex);}else{pthread_mutex_unlock(mutex);//如果if为假直接走else那么锁没有被解锁所以这里要解锁break;}}
}
int main()
{std::unique_ptrThread thread1(new Thread(get_train_tickets, (void *)user 1, 1));std::unique_ptrThread thread2(new Thread(get_train_tickets, (void *)user 2, 2));std::unique_ptrThread thread3(new Thread(get_train_tickets, (void *)user 3, 3));thread1-join();thread2-join();thread3-join();return 0;
}上面抢票确实是安全的但是都是由一个线程抢完了也就是说哪一个线程先被加锁了它就会一直抢票直到结束然后解锁 这个情况我们等下再解释我们先来看看局部变量的锁怎么使用
#include iostream
#include pthread.h
#include string
#include vector
#include unistd.h
#include cstring
#include cstdio
#include memory
#include Thread.hpp
using std::cout;
using std::endl;int train_tickets 1000;
class ThreadData
{
public:ThreadData(std::string threadname, pthread_mutex_t *lock): _threadname(threadname),_lock(lock){}~ThreadData(){}public: // 为了方便下面使用就不定义为private了std::string _threadname;pthread_mutex_t *_lock;
};void *get_train_tickets(void *args)
{ThreadData *td static_castThreadData *(args);while (1){pthread_mutex_lock(td-_lock);if (train_tickets 0){usleep(1000);printf(%s正在进行抢票 : %d\n, td-_threadname.c_str(), train_tickets);train_tickets--;pthread_mutex_unlock(td-_lock);}else{pthread_mutex_unlock(td-_lock);break;}//不能在这里解锁不然上面else的break直接跳出循环了锁没有被解开}
}int main()
{
#define NUM 4pthread_mutex_t lock;//这把锁是公共的锁pthread_mutex_init(lock, nullptr);std::vectorpthread_t tids(NUM);for (size_t i 0; i NUM; i){char buffer[128];snprintf(buffer, sizeof buffer, thread %d, i 1); // i1使得新线程id不从0开始ThreadData *td new ThreadData(buffer, lock);pthread_create(tids[i], nullptr, get_train_tickets, td);}for (const auto tid : tids){pthread_join(tid, nullptr);}pthread_mutex_destroy(lock);// pthread_t t1, t2, t3, t4;// pthread_create(t1, nullptr, get_train_tickets, (void *)thread 1);// pthread_create(t2, nullptr, get_train_tickets, (void *)thread 2);// pthread_create(t3, nullptr, get_train_tickets, (void *)thread 3);// pthread_create(t4, nullptr, get_train_tickets, (void *)thread 4);// pthread_join(t1, nullptr);// pthread_join(t2, nullptr);// pthread_join(t3, nullptr);// pthread_join(t4, nullptr);// pthread_mutex_destroy(lock);return 0;
}加锁和解锁的过程是被多个线程串行执行的这就导致程序变慢了 锁只规定互斥访问没有规定必须让谁优先执行 所以锁就是多个执行流竞争的结果谁竞争到了锁就是谁的
当然我们线程抢完票就什么都不做了吗 当然不是的我们线程抢到票了还要给用户汇报抢到票的消息以及处理其他工作 5-3-1、如何理解锁
线程要使用锁达到安全访问共享资源的目的那么线程要先看到锁才行 所以锁被每一个线程都看到了锁就是一个共享资源 锁是用来保护全局资源的共享资源锁本身也是一个全局资源共享资源那么谁来保护锁呢 所以pthread_mutex_lock/pthread_mutex_unlock 加锁和解锁操作必须是安全的加锁操作是原子的
如果申请锁成功了那么继续向后执行代码那么如果申请没有成功呢 如果申请锁没有成功执行流会被阻塞线程进入休眠状态 这种申请失败使执行流阻塞的锁我们称为 —— 挂起等待锁 谁持有锁谁进入临界区 未来我们写代码的时候如果要多线程访问公共资源共享资源我们要么对线程全部加锁要么就全部不加锁 不能一部分线程加锁一部分线程不加锁这算我们代码是有bug的 所以我们申请锁的操作一定要是安全的原子的因为锁是一个共享资源还要起到保护共享资源的作用如果锁都是一个不安全的操作怎么保护共享资源呢所以锁也是要被保护的而保护锁的操作就是 —— 申请锁是原子操作 5-3-2、如何理解加锁和解锁的本质 加锁的过程是原子的而我们对于解锁其实没有太多的要求因为就你一个线程拿到锁了其他人都没有正常进行解锁操作把锁让出来就行 接下来我们来研究一下互斥量实现原理也就是互斥锁是怎么样保证加锁是原子操作的 经过上面的例子大家已经意识到单纯的 i 或者 i 都不是原子的有可能会有数据一致性问题 为了实现互斥锁操作,大多数体系结构(cpu架构比如x86_32)都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性,即使是多处理器平台,访问内存的 总线周期也有先后,一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。 在硬件层面上面我们可以通过地址总线使线程/进程不会被切换的操作让进程/线程达到原子性。比如时间片到了优先级更高的进程/线程来了进程/线程要进行等待等一系列要切换进程/线程的操作通过地址总线对切换操作进行忽略 当然一般除非遇到一个进程/线程特别特别重要的时候才会作用操作大部分情况下都不会这么做
现在我们把lock和unlock的伪代码改一下 1、cpu内部只有一套寄存器被所有执行流共享 2、cpu内部寄存器存放的内容是每一个执行流私有的是每一个执行流运行时的上下文结构 5-3-3、对锁进行封装设计
Mutex.hpp:
#pragma once#include iostream
#include pthread.husing std::cout;
using std::endl;class Mutex
{
public:Mutex(pthread_mutex_t *lock_p nullptr): _lock_p(lock_p){}void lock(){if (_lock_p)//锁不为空才表示要设置锁pthread_mutex_lock(_lock_p);}void unlock(){if (_lock_p)//锁不为空表示有锁需要我们解锁pthread_mutex_unlock(_lock_p);}~Mutex() {}private:pthread_mutex_t *_lock_p;
};class LockGuard
{public:LockGuard(Mutex mutex):_mutex(mutex){_mutex.lock();//在构造函数中加锁}~LockGuard(){_mutex.unlock();//在析构函数中解锁}private:Mutex _mutex;
};#include iostream
#include pthread.h
#include string
#include vector
#include unistd.h
#include cstring
#include cstdio
#include memory
#include Thread.hpp
#include Mutex.hppusing std::cout;
using std::endl;pthread_mutex_t lock PTHREAD_MUTEX_INITIALIZER;
int train_tickets 1000;
class ThreadData
{
public:ThreadData(std::string threadname, pthread_mutex_t *lock): _threadname(threadname),_lock(lock){}~ThreadData(){}public:std::string _threadname;pthread_mutex_t *_lock;
};
void *get_train_tickets(void *args)
{std::string user_name static_castconst char *(args);while (1){{//不想把usleep也加锁加一个花括号就相当于一个作用域了// 加锁LockGuard lockguard(lock);//RAII操作if (train_tickets 0){usleep(1000);printf(%s正在进行抢票 : %d\n, user_name.c_str(), train_tickets);train_tickets--;}else{break;//这里break也不怕了因为出了作用域lockguard对象自动销毁调用析构}}usleep(1000);}
}
int main()
{pthread_t t1, t2, t3, t4;pthread_create(t1, nullptr, get_train_tickets, (void *)thread 1);pthread_create(t2, nullptr, get_train_tickets, (void *)thread 2);pthread_create(t3, nullptr, get_train_tickets, (void *)thread 3);pthread_create(t4, nullptr, get_train_tickets, (void *)thread 4);pthread_join(t1, nullptr);pthread_join(t2, nullptr);pthread_join(t3, nullptr);pthread_join(t4, nullptr);return 0;
}5-4、可重入VS线程安全 线程安全多个线程并发同一段代码时不会出现不同的结果。常见对全局变量或者静态变量进行操作并且没有锁保护的情况下会出现该问题。 重入同一个函数被不同的执行流调用当前一个流程还没有执行完就有其他的执行流再次进入我们称之为重入。一个函数在重入的情况下运行结果不会出现任何不同或者任何问题则该函数被称为可重入函数否则是不可重入函数 5-4-1、常见的线程不安全的情况 不保护共享变量的函数 函数状态随着被调用状态发生变化的函数 返回指向静态变量指针的函数 调用线程不安全函数的函数 5-4-2、常见的线程安全的情况 每个线程对全局变量或者静态变量只有读取的权限而没有写入的权限一般来说这些线程是安全的 类或者接口对于线程来说都是原子操作 多个线程之间的切换不会导致该接口的执行结果存在二义性 多线程访问共享资源全局变量不加保护会引发线程安全问题但是多线程访问局部资源就不会
5-4-3、常见不可重入的情况 调用了malloc/free函数因为malloc函数是用全局链表来管理堆的 调用了标准I/O库函数标准I/O库的很多实现都以不可重入的方式使用全局数据结构 可重入函数体内使用了静态的数据结构 5-4-4、常见可重入的情况 不使用全局变量或静态变量 不使用用malloc或者new开辟出的空间 不调用不可重入函数 不返回静态或全局数据所有数据都有函数的调用者提供 使用本地数据或者通过制作全局数据的本地拷贝来保护全局数据 5-4-5、可重入与线程安全联系 函数是可重入的那就是线程安全的 函数是不可重入的那就不能由多个线程使用有可能引发线程安全问题 如果一个函数中有全局变量那么这个函数既不是线程安全也不是可重入的。 5-4-6、可重入与线程安全区别 可重入函数是线程安全函数的一种 线程安全不一定是可重入的而可重入函数则一定是线程安全的 如果将对临界资源的访问加上锁则这个函数是线程安全的但如果这个重入函数若锁还未释放则会产生死锁因此是不可重入的 可重入函数是线程安全的一种一个函数是可重入函数那么这个函数就算线程安全的 但是线程安全的函数不一定是可重入函数
5-4-7、小结 一个函数能被多个执行流执行切不会出错那么这个函数就算可重入函数出错就是不可重入函数 一个代码片段被多个线程调用如果没有引发数据不安全问题如果有那么这个线程就算不安全的没有就是线程安全的 6、常见锁概念
6-1、死锁 死锁是指在一组进程中的各个线程轻量级进程均占有不会释放的资源但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态 在多把锁的刺激下我们持有自己的锁不释放还要对方的锁对方也是如此此时就容易造成死锁 一把锁可能产生死锁吗是可以的我们重复申请一把锁两次/多次
为什么会有死锁呢
我们后面会再讲这里我们先来讲讲逻辑链条
为什么会有死锁呢因为我们用了锁 ——为什么我们要用锁呢保证临界资源的安全——为什么要保证临界资源的安全呢因为多线程访问我们可能出现数据不一致问题——为什么会产生数据不一致问题呢因为我们用了多线程并且访问了全局资源——多线程大部分资源全局资源是共享的——多线程的特性
任何技术都有自己的边界是解决问题的但有可能在解决问题的同时一定会引入新的问题
6-2、死锁的4个必要条件 互斥条件一个资源每次只能被一个执行流使用 请求与保持条件一个执行流因请求资源而阻塞时对已获得的资源保持不放 不剥夺条件:一个执行流已获得的资源在末使用完之前不能强行剥夺 循环等待条件环路等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系 如果我们要得到一个死锁那么这4个条件必须满足
我们来仔细分析
1、互斥我们访问某些资源的时候必须是互斥的线程串行访问没有互斥就代表没有加锁锁都没有还谈什么死锁
2、请求与保持我要你的锁请求我不释放我的锁保持
3、不剥夺我要你的但是我不抢你的要让你自愿释放锁这叫不剥夺【我要你的但是我通过优先级或者手动设置的状态允许我过来抢过你的这就是剥夺】
4、环路等待条件A有自己的锁不释放还要B的锁B有自己的锁不释放还要C的锁C有自己的锁不释放还要A的锁。这样就形成了一个环路
6-3、避免死锁
那么我们要怎么来破坏死锁呢 破坏死锁的四个必要条件 破坏一个必要条件就行了 1、破坏互斥 互斥是锁的特性这个肯定是不能被破坏的 2、破坏请求与保持我们有一个锁了再申请锁的时候就会申请失败这个时候我们如果申请失败我们把锁给释放掉 3、破坏不剥夺我们讲锁设置为可以剥夺的就行我们通过线程的优先级或者手动设置的状态让你被迫放弃你的锁然后将锁给我 4、破坏环路等待条件 让线程申请锁的顺序一致多线程同时先申请A锁然后B锁然后C锁 加锁顺序一致破坏环路等待条件 避免锁未释放的场景破坏请求与保持 资源一次性分配要加锁的地方直接一次性全部加锁不要打散式的加锁 6-4、避免死锁算法 死锁检测算法(了解) —— 锁里面有一个计数器如果线程检测这个计数器长时间没有发生变化线程就会自动进行解锁 银行家算法了解 一个线程申请锁可以有另一个线程来释放锁
加锁也是要根据实际情况来使用的我们上面的抢票中票全被一个线程抢完了这有错误吗没有错误但是这是不合理的
7、Linux线程同步
7-1、同步的概念
当我们安全访问临界资源的条件下让多个线程按照顺序来进行访问从而避免饥饿问题这就叫同步
7-2、条件变量
条件变量是pthread给我们提供的一种数据类型我们定义了条件变量之后阻塞等待的线程在条件变量下面进行等待阻塞线程一个个链接在条件变量下面如同单链表一样 1、当一个线程互斥地访问某个变量时它可能发现在其它线程改变状态之前它什么也做不了。 2、例如一个线程访问队列时发现队列为空线程一直加锁判断队列有没有节点没有然后解锁。一直重复这个工作这个过程没有错误但是严重不合理 我们要做到一个线程访问队列时发现队列为空它只能等待直到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量 抢票也是我今天发布了1000张票黄牛通过线程抢完之后黄牛只能通过线程拿到锁然后检测是否有票有就抢没有就什么都做不了【因为获取票的条件没有满足 没有票】 将阻塞挂起的线程放到条件变量下面进行等待然后被唤醒再拿走 7-3、理解条件变量
样例 面试官在一个酒店的房间里面进行招聘那么这个面试官就是一个共享资源并且因为有房间锁的保护一次只能有一个人进行面试。但是房间外面的人可能不讲武德有一个人就进去面试感觉面的不好又进去面试了一遍面试官还没有认出来这样循环式的让一个人一直面试就导致其他人的饥饿问题了其他人都不能参加面试了 这个时候来了一个比较强的面试官 当条件不满足的时候多线程必须去某些定义好的条件变量进行等待。等条件满足了再从条件变量下面唤醒线程
画一张图
7-4、条件变量函数
初始化
int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrictattr);参数
cond要初始化的条件变量
attrNULL销毁
int pthread_cond_destroy(pthread_cond_t *cond)等待条件满足
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
参数
cond要在这个条件变量上等待
mutex互斥量后面详细解释唤醒等待
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);条件变量本身不具备互斥的功能所以要配合互斥锁来使用
老规矩见见猪跑
#include iostream
#include pthread.h
#include string
#include unistd.husing std::cout;
using std::endl;pthread_mutex_t lock PTHREAD_MUTEX_INITIALIZER; // 这里就将条件变量和锁定义为全局的后面会使用init和destroy函数
pthread_cond_t cond PTHREAD_COND_INITIALIZER; // 条件变量本身不是互斥的所以要配和互斥锁来使用int tickets 1000;void *start_routine(void *args)
{std::string name static_castconst char *(args);while (1){pthread_mutex_lock(lock);// if (tickets 0)//判断暂时省略pthread_cond_wait(cond, lock); // 这里为什么要有lock锁我们后面说cout name - tickets endl;cout.flush();tickets--;pthread_mutex_unlock(lock);}
}
int main()
{// 通过条件变量控制线程的执行pthread_t t1, t2;pthread_create(t1, nullptr, start_routine, (void *)thread 1);pthread_create(t2, nullptr, start_routine, (void *)thread 2);while (1){sleep(1);pthread_cond_signal(cond);cout main thread wakeup one thread... endl;cout.flush();}pthread_join(t1, nullptr);pthread_join(t2, nullptr);return 0;
}8、生产者消费者模型
8-1、举例说明 生产消费模型是一种多执行流协同的方式 为什么要协同呢多个执行流各自执行各自的不行吗
我们多线程在访问共享资源的时候如果不对共享资源进行加锁保护会引发数据不一致问题。加锁之后我们确实保证了数据安全的问题但是像我们上面抢票那样一个线程直接把票抢完了导致其他线程饥饿问题这有错吗没有错但是这合理吗不合理
所以我们在多线程工作的场景下面既要保证数据的安全又要保证多线程在工作的时候按照特定的顺序来访问。这种顺序可以不是一种绝对顺序但是一定要有顺序尽可能保证每个线程都能合适合理的访问某种资源。所以我们需要一种多线程在这种场景下面工作的模式这种模式最经典最常见的就叫做 —— 生产消费模型 拿火腿肠举例。我们消费者一般不能直接去找供货商买火腿肠因为我们吃火腿肠一次顶多吃几串就算一个人去买两箱但是又许多人都去供货商哪里买供货商卖给我们火腿肠的成本还不如开机器养员工的成本所以我们去找供货商买货品供货商卖给我们是亏本生意一般不买其次就是供货商的工厂一般在远离市中心的地方我们去供货商路途较远我们买货物所消耗的钱可能还没有路费花的多。所以就算我们跑到供货商门口供货商也不会卖给我们因为这是亏本生意
那么供货商将一批大量的货物生产好之后怎么将货物销售给我们消费者呢这个时候就要通过中间人来向我们消费者提供货物了这个中间人就是 —— 超市。超市集中需求分发产品 —— 超市将我们各个消费者的需求集中起来然后提供供货商提供的货物将这些货物销售给消费者供货商一次性将一批产品出售给各地区的超市然后再由超人作为中间人出售给我们消费者这样供货商既不会亏本消费者也可以轻松进行消费
所以现实生活中学生等购买货物的人是消费者供货商等提供货物的人是生产者而超市这个中间人就是生产者和消费者的交易场所 而有了这个超市交易场所之后供货商的人有可能在进行加工生成商品也可能过年在家休息而我们学生消费者可能在玩游戏刷抖音。
有了超市之后我们就【把这种生产者和消费者互不干扰的行为用计算机的语言就称为 生产者过程和消费者过程的 —— 解耦】
而当我们消费者一直消费的时候把超市的东西买完了生产者一直不向超市提供货物这就没有货物供我们购买了
当生产者一直生产向超市提供货物而超市都放不下生产者的货物了但是没有一个消费者来购买这就导致超市没有空间来存放货物了
所以【超市只不过是临时保存货物的场所用计算机语言就称为 缓冲区】有了缓冲区之后在缓冲区有足够空间的前提下通过这个缓冲区生产者可以一直向缓冲区提供数据而消费者也可以一直向缓冲区消费数据。这就使得 生产者和消费者的步调并不怎么一致从而达到生产者过程和消费者过程的 —— 解耦
而上面的缓冲区起到了生产者过程和消费者过程的解耦那么我们举一个没有解耦样例
我们在main函数中调用fun函数给fun函数传参 其中 调用fun函数的一方生成了数据 形成的变量 变量暂时保存数据 目标函数 消费了数据 而当我们main函数调用fun函数的时候main函数此时什么都不能做要等fun函数调用完返回main函数才能继续向下执行
所以main函数和fun函数就是一种强耦合关系我们上面买火腿肠的例子中一个消费者去供货商买一根火腿肠这个时候供货商对这个消费者说你在这里等着我去开机器给你做一根火腿肠这个时候消费者只能等着这就是生产者和消费者的强耦合
8-2、深入研究 我们通过上面知道了 —— 生产者生产货物到超市而消费者要在超市进行消费 —— 所以生产者和消费者要都能看到超市才行 —— 那么这个超市就是一个共享资源 那么就会出现这种情况 当我们买火腿肠的时候超市没有了而生产者正在往超市架子上面放这个时候我们能不能买火腿肠成功呢是不确定的因为生产者向超市架子摆放火腿肠这个操作是原子的要么摆上去了要么没有摆上去这个时候我们能不能买到火腿肠取决于生产者有没有把火腿肠放到超市架子上面
所以如果不对共享资源进行保护超市那么就会引发数据不一致问题所以共享资源超市要被保护起来 我们的生产者在计算机中是一个或者多个线程 我们的消费者在计算机中也是一个或者多个线程 共享资源超市要怎么被被保护起来呢 我们就先要讨论一下生产者与生产者消费者与消费者生产者与消费者之间的关系了 1、生产者和生产者之间的关系 互斥关系两家火腿肠供货商超市要么摆你的火腿肠要么摆我的火腿肠。可以先摆你的火腿肠卖完之后再摆我的但是不能一起摆不然混在一起消费者不好购买所以对于超市固定的框架来说我们处于竞争关系用计算机语言来说我们处于互斥有我没你有你没我 2、消费者与消费者之间的关系 互斥关系两个消费者看上了同一份数据互不相让通过竞争之后才能知道哪一个消费者拿到了这份数据。硬件设计语言来说两个消费者处于互斥要么我拿到了这份数据要么你拿到了这份数据 3、生产者和消费者之间的关系 同步与互斥关系 互斥关系 我们生产者生产了一份hello world消费者要读出这份hello world。消费者刚刚读出hello生产者就直接将剩下的world改成了bit这就导致了消费者读取出来的数据是hello bit。这就引发了数据不一致问题了所以生产者和消费者是互斥关系只允许一方对数据进行访问 同步关系当超市没有货物的时候一个消费者来问有没有货物超市工作人员说没有但是今天或者这段时间这个消费者一直来问工作人员有没有货物工作人员一直回答没有。而我们不止一个消费者肯定有很多消费者如果每一个消费者都这么一直问那么既浪费了每一个消费者的时间又浪费了工作人员的时间本来超市工作人员可以用这些时间来催促供货商提供货物的 同理当我们超市货物摆满了的时候没有人来消费而生产者一直询问超市工作人员超市有没有空间工作人员也回答说没有这个时候多个生产者一直询问而工作人员也一直回答没有这就又浪费生产者的时间又浪费了工作人员的时间本来工作人员可以用这些时间来通知消费者过来消费的 所以我们可以让超市工作人员加上消费者和生产者的微信有来货物通知消费者来消费货物不足的时候通知生产者来生产货物 —— 通过这种策略保证消费者过程和生产者过程协同起来。维护生产者与消费者之间的同步关系从而提高整个生产消费模型的效率 8-3、小结
我们生产消费模型遵循一个原则 321原则 3种关系 生产者和生产者之间的关系互斥消费者和消费者之间的关系互斥生产者和消费者之间的关系同步【保证共享资源的安全性】与互斥 —— 产品本质是数据 2种角色生产者线程和消费者线程 1个交易场所一段特定结构的缓冲区 我们想写生产消费模型本质工作就是维护321原则
8-4、生产消费模型的特点 1、生产线程和消费线程进行解耦缓冲区可以缓存数据消费线程不用一直等生产线程通过缓冲区可以各做各的 2、支持生产和消费在一段时间内忙闲不均的问题消费者消费的很快生产者生产的很慢不用担心缓冲区缓存了一批数据消费者直接来消费就行等消费者消费慢下来了再找生产者加载数据到缓冲区里面消费者消费的很慢生产者生产的很快也不用担心缓冲区会将生产者生产的数据都保存起来等生产者生产慢下来的时候再通知消费者来进行消费 这样就使得生产者生产的快慢和消费者消费的快慢没有互相影响 3、提高效率生产者专门生产消费者专门消费因为有缓冲区的存在不需要消费者直接找生产者等生产者生产生产者也不需要等消费者到了然后进行生产 但是还是有问题 生产消费模型的特点有一个提高效率但是如果我缓冲区已经满了但是生产者还向缓冲区打数据这个时候就不行得等消费者来消费而缓冲区没有数据了消费者也不能进行消费要得生产者向缓冲区打数据这不还是要等吗这么就提高效率了呢
再者就是我们如果只维护生产消费模型互斥关系是肯定不行的因为我们对共享资源缓冲区要进行保护加锁和解锁如果生产者拿到锁之后进行判断缓冲区有没有满没有满就向缓冲区输入数据然后解锁。当缓冲区满了的时候生产者线程拿到锁然后判断缓冲区满了直接解锁。然后生产者线程又拿到锁判断缓冲区满了然后解锁…像这样生产者线程拿到锁之后判断然后解锁什么都不做一直占用锁资源 所以我们生产消费模型不能只维护互斥关系还要维护同步关系
8-5、基于BlockingQueue的生产者消费者模型
我们上面7-4学习了pthread_cond_wait和pthread_cond_signal函数这里我们来以这上面的知识为基点继续深入学习
BlockingQueue 在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于当队列为空时从队列获取元素的操作将会被阻塞直到队列中被放入了元素当队列满时往队列里存放元素的操作也会被阻塞直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的线程在对阻塞队列进程操作时会被阻塞) 这个BlockingQueue就是我们上面说的一段特定结构的缓冲区超市 这个BlockQueue就是共享资源它可以为空可以为满有最大上限限制这就约束了我们的生产者和消费者在特定情况下应该阻塞住
我们这里直接用C中stl容器的queue来充当BlockingQueue
先来看看最基础的版本
BlockingQueue.hpp
#pragma once#include iostream
#include queue
#include pthread.hstatic const int gmaxcap 5;
template class T
class BlockingQueue
{
public:BlockingQueue(const int maxcap gmaxcap): _maxcap(maxcap){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_pcond, nullptr);pthread_cond_init(_ccond, nullptr);}void push(const T in) // 输入型参数 —— const {pthread_mutex_lock(_mutex);// 细节2: 充当条件判断的语法必须是while不能用ifwhile(is_full())//if (is_full())// 细节1pthread_cond_wait这个函数的第二个参数必须是我们正在使用的互斥锁不然我们线程在条件变量下等待的时候抱着锁直接跑了其他线程都不能申请成功锁了// a. pthread_cond_wait: 该函数调用的时候会以原子性的方式将锁释放并将自己挂起// b. pthread_cond_wait: 该函数在被唤醒返回继续向下执行代码的时候会自动的重新获取你传入的锁pthread_cond_wait(_pcond, _mutex); // 生产条件不满足无法生产生产者进行等待// 走到这里阻塞队列里面一定没有满_q.push(in);// 阻塞队列里面一定有数据pthread_cond_signal(_ccond); // 细节3pthread_cond_signal这个函数可以放在临界区内部也可以放在外部pthread_mutex_unlock(_mutex);//pthread_cond_signal(_pcond);也可以放在这里}void pop(T *out) // 输出型参数 —— * //输入输出型参数 —— {pthread_mutex_lock(_mutex);while(is_empty())//if (is_empty())pthread_cond_wait(_ccond, _mutex);// 走到这里阻塞队列里面一定不为空*out _q.front();_q.pop();// 阻塞队列里面一定有一个空位置pthread_cond_signal(_pcond);// 细节3pthread_cond_signal这个函数可以放在临界区内部也可以放在外部pthread_mutex_unlock(_mutex);//pthread_cond_signal(_pcond);也可以放在这里}~BlockingQueue(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_pcond);pthread_cond_destroy(_ccond);}private:bool is_empty(){return _q.empty();}bool is_full(){return _q.size() _maxcap;}private:std::queueT _q;int _maxcap; // 队列最大上限pthread_mutex_t _mutex;pthread_cond_t _pcond; // 生产者的条件变量pthread_cond_t _ccond; // 消费者的条件变量
};Main.cc:
#include iostream
#include pthread.h
#include ctime
#include unistd.h
#include sys/types.h
#include BlockingQueue.hpp
using std::cout;
using std::endl;void *consumer(void *args)
{BlockingQueueint *bq static_castBlockingQueueint *(args);while (1){// 消费int data;bq-pop(data);cout 消费数据 : data endl;// sleep(1);}return nullptr;
}
void *producter(void *args)
{BlockingQueueint *bq static_castBlockingQueueint *(args);while (1){// 生产int data rand() % 10 1;bq-push(data);cout 生产数据 : data endl;sleep(1);//生产速度慢下来消费数据也必须慢因为这是阻塞队列}return nullptr;
}
int main()
{srand((unsigned long)time(nullptr) ^ getpid());BlockingQueueint *bq new BlockingQueueint();pthread_t c, p;pthread_create(c, nullptr, consumer, bq);pthread_create(p, nullptr, producter, bq);pthread_join(c, nullptr);pthread_join(p, nullptr);delete bq;return 0;
}如果不用signal函数唤醒而是使用broadcast函数唤醒一批线程如果使用if判断那么我们的_q.push和_p.pop操作可能执行了多次当我们阻塞队列如果为空或者为满的时候只有一个线程执行后面的代码其他线程都在while里面循环再次执行wait函数到对应的条件变量上面等待
当然了我们都用模板了就为了搞个int类型这么简单吗 当然不是我们是可以向阻塞队列里面放任务的 我们来继续改进代码
BlockQueue.hpp上面都没有没有改但还是放在这里好统一观看
#pragma once#include iostream
#include queue
#include pthread.hstatic const int gmaxcap 5;
template class T
class BlockingQueue//生产和消费必须都能够先看到这个阻塞队列
{
public:BlockingQueue(const int maxcap gmaxcap): _maxcap(maxcap){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_pcond, nullptr);pthread_cond_init(_ccond, nullptr);}void push(const T in) // 输入型参数 —— const {pthread_mutex_lock(_mutex);// if (is_full())while (is_full()){pthread_cond_wait(_pcond, _mutex); // 生产条件不满足无法生产生产者进行等待}// 走到这里阻塞队列里面一定没有满_q.push(in);// 阻塞队列里面一定有数据pthread_cond_signal(_ccond); // 阻塞队列有数据了我们唤醒消费者开始进行消费pthread_mutex_unlock(_mutex);}void pop(T *out) // 输出型参数 —— * //输入输出型参数 —— {pthread_mutex_lock(_mutex);// if (is_empty())while (is_empty()){pthread_cond_wait(_ccond, _mutex);}// 走到这里阻塞队列里面一定不为空*out _q.front();_q.pop();// 阻塞队列里面一定有一个空位置pthread_cond_signal(_pcond);pthread_mutex_unlock(_mutex);}~BlockingQueue(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_pcond);pthread_cond_destroy(_ccond);}private:bool is_empty(){return _q.empty();}bool is_full(){return _q.size() _maxcap;}private:std::queueT _q;int _maxcap; // 队列最大上限pthread_mutex_t _mutex;pthread_cond_t _pcond; // 生产者的条件变量pthread_cond_t _ccond; // 消费者的条件变量
};
Main.cc:
#include iostream
#include pthread.h
#include ctime
#include cstring
#include unistd.h
#include sys/types.h
#include BlockingQueue.hpp
#include Task.hpp
using std::cerr;
using std::cout;
using std::endl;const std::string oper -*/%;
int my_math(int x, int y, char op)
{int result 0;switch (op){case :// return x y;//return 不需要breakresult x y;break;case -:// return x - y;result x - y;break;case *:// return x * y;result x * y;break;case /:{if (y 0){cerr div zero error! endl;result -1;}elseresult x / y;}break;case %:{if (y 0){cerr mod zero error! endl;result -1;}elseresult x % y;// return x % y;}break;default:break;}return result;
}
int myadd(int x, int y)
{return x y;
}
void *consumer(void *args)
{BlockingQueueTask *bq static_castBlockingQueueTask *(args);while (1){// 消费Task t;bq-pop(t);cout 消费任务 : t() endl; // 仿函数 -// cout 消费任务 : ;// cout t() endl;sleep(1);// int data;// bq-pop(data);// cout 消费数据 : data endl;// sleep(1);}return nullptr;
}
void *producter(void *args)
{BlockingQueueTask *bq static_castBlockingQueueTask *(args);while (1){// 生产int x rand() % 10 1;// int y rand() % 5 1;int y rand() % 5; // 不加1可能会%到0char op rand() % oper.size();Task t(x, y, oper[op], my_math);bq-push(t);cout 生产任务 : t.TaskToString() endl;sleep(1);// cout 生产数据 : t endl; sleep(1);//生产速度慢下来消费数据也必须慢因为这是阻塞队列}return nullptr;
}
int main()
{srand((unsigned long)time(nullptr) ^ getpid());BlockingQueueTask *bq new BlockingQueueTask();pthread_t c, p;pthread_create(c, nullptr, consumer, bq);pthread_create(p, nullptr, producter, bq);pthread_join(c, nullptr);pthread_join(p, nullptr);delete bq;return 0;
}Task.hpp:
#pragma once#include iostream
#include queue
#include pthread.h
#include functional
#include cstdioclass Task
{using func_t std::functionint(int, int, char); // 这里也要加上一个char不然回调函数和Main文件里面的my_math函数参数对不上// typedef functionint(int, int) func_t;public:Task(){}Task(int x, int y, char op, func_t func): _x(x),_y(y),_op(op),_callback(func){}// int operator()() // 仿函数std::string operator()(){int result _callback(_x, _y, _op);char buffer[1024];snprintf(buffer, sizeof(buffer), %d %c %d %d, _x, _op, _y, result);return buffer;// return result;}std::string TaskToString(){char buffer[1024];snprintf(buffer, sizeof(buffer), %d %c %d ?, _x, _op, _y);return buffer;}~Task(){}private:int _x;int _y;char _op;func_t _callback;
};这里出现乱序的原因是我们的生产和消费的打印没有加锁while是死循环打印难免出现乱序问题 我们接下来继续改进我们有了一个生产一个消费
我们还可以这样改 1号线程生产派发任务通过阻塞队列12号线程消费处理任务然后2号线程再通过阻塞队列23号线程将2号线程的处理结果保存在文件中 BlockingQueue.hpp还是没改
#pragma once#include iostream
#include queue
#include pthread.hstatic const int gmaxcap 500;
template class T
class BlockingQueue//生产和消费必须都能够先看到这个阻塞队列
{
public:BlockingQueue(const int maxcap gmaxcap): _maxcap(maxcap){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_pcond, nullptr);pthread_cond_init(_ccond, nullptr);}void push(const T in) // 输入型参数 —— const {pthread_mutex_lock(_mutex);// if (is_full())while (is_full()){pthread_cond_wait(_pcond, _mutex); // 生产条件不满足无法生产生产者进行等待}// 走到这里阻塞队列里面一定没有满_q.push(in);// 阻塞队列里面一定有数据pthread_cond_signal(_ccond); // 阻塞队列有数据了我们唤醒消费者开始进行消费pthread_mutex_unlock(_mutex);}void pop(T *out) // 输出型参数 —— * //输入输出型参数 —— {pthread_mutex_lock(_mutex);// if (is_empty())while (is_empty()){pthread_cond_wait(_ccond, _mutex);}// 走到这里阻塞队列里面一定不为空*out _q.front();_q.pop();// 阻塞队列里面一定有一个空位置pthread_cond_signal(_pcond);pthread_mutex_unlock(_mutex);}~BlockingQueue(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_pcond);pthread_cond_destroy(_ccond);}private:bool is_empty(){return _q.empty();}bool is_full(){return _q.size() _maxcap;}private:std::queueT _q;int _maxcap; // 队列最大上限pthread_mutex_t _mutex;pthread_cond_t _pcond; // 生产者的条件变量pthread_cond_t _ccond; // 消费者的条件变量
};
Task.hpp:
#pragma once#include iostream
#include queue
#include pthread.h
#include functional
#include cstdio
#include cstringusing std::cerr;
using std::cout;
using std::endl;const std::string oper -*/%;
int my_math(int x, int y, char op)
{int result 0;switch (op){case :// return x y;//return 不需要breakresult x y;break;case -:// return x - y;result x - y;break;case *:// return x * y;result x * y;break;case /:{if (y 0){cerr div zero error! endl;result -1;}elseresult x / y;}break;case %:{if (y 0){cerr mod zero error! endl;result -1;}elseresult x % y;// return x % y;}break;default:break;}return result;
}class CalTask
{using func_t std::functionint(int, int, char); // 这里也要加上一个char不然回调函数和Main文件里面的my_math函数参数对不上// typedef functionint(int, int) func_t;public:CalTask(){}CalTask(int x, int y, char op, func_t func): _x(x),_y(y),_op(op),_callback(func){}// int operator()() // 仿函数std::string operator()(){int result _callback(_x, _y, _op);char buffer[1024];snprintf(buffer, sizeof(buffer), %d %c %d %d, _x, _op, _y, result);return buffer;// return result;}std::string TaskToString(){char buffer[1024];snprintf(buffer, sizeof(buffer), %d %c %d ?, _x, _op, _y);return buffer;}~CalTask(){}private:int _x;int _y;char _op;func_t _callback;
};void savethread(const std::string massage)
{const std::string target ./log.tex;FILE *fp fopen(target.c_str(), a);if (fp nullptr){cerr fopen file error! endl;return;}fputs(massage.c_str(), fp);fprintf(fp, \n);fclose(fp);
}class SaveTask
{typedef std::functionvoid(const std::string ) func_t;public:SaveTask(){}SaveTask(const std::string massage, func_t func): _massage(massage),_func(func){}~SaveTask(){}void operator()(){_func(_massage);}private:std::string _massage;func_t _func;
};Main.cc:
#include iostream
#include pthread.h
#include ctime
#include cstring
#include unistd.h
#include sys/types.h
#include BlockingQueue.hpp
#include Task.hpp
using std::cerr;
using std::cout;
using std::endl;pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;template class C, class S // C:计算S存储
class BlockQueues
{
public:BlockingQueueC *c_bp;BlockingQueueS *s_bp;
};void *producter(void *args)
{// BlockQueuesCalTask, SaveTask *bq static_castBlockQueuesCalTask, SaveTask *(args);BlockingQueueCalTask *bq (static_castBlockQueuesCalTask, SaveTask *(args))-c_bp;while (1){// 生产int x rand() % 10 1;// int y rand() % 5 1;int y rand() % 5; // 不加1可能会%到0char op rand() % oper.size();CalTask t(x, y, oper[op], my_math);bq-push(t);pthread_mutex_lock(mutex);cout producter thread 生产计算任务 : t.TaskToString() endl;pthread_mutex_unlock(mutex);sleep(1);// cout 生产数据 : t endl; sleep(1);//生产速度慢下来消费数据也必须慢因为这是阻塞队列}return nullptr;
}void *consumer(void *args)
{// BlockQueuesCalTask, SaveTask *bq static_castBlockQueuesCalTask, SaveTask *(args);BlockingQueueCalTask *bp (static_castBlockQueuesCalTask, SaveTask *(args))-c_bp; // 2号线程从这个队列拿数据进行消费处理BlockingQueueSaveTask *save_bp (static_castBlockQueuesCalTask, SaveTask *(args))-s_bp; // 2号线程向这个队列放数据给3号线程进行消费while (1){// 消费 —— 计算任务CalTask t;// bq-c_bp-pop(t);//对应BlockQueuesCalTask, SaveTask *bq static_castBlockQueuesCalTask, SaveTask *(args);bp-pop(t);std::string result t(); // 将消费处理结果赋值给resultpthread_mutex_lock(mutex);cout consumer thread 消费计算任务 : result endl; // 仿函数 -pthread_mutex_unlock(mutex);// 生产 —— 存储任务SaveTask save(result, savethread);save_bp-push(save);pthread_mutex_lock(mutex);cout 推送保存任务完成... endl; // 仿函数 -pthread_mutex_unlock(mutex);// sleep(1);}return nullptr;
}void *saver(void *args)
{BlockingQueueSaveTask *save_bp (static_castBlockQueuesCalTask, SaveTask *(args))-s_bp;while (1){SaveTask t;save_bp-pop(t);t();cout saver thread 保存任务完成... endl; // 仿函数 -}return nullptr;
}int main()
{srand((unsigned long)time(nullptr) ^ getpid());BlockQueuesCalTask, SaveTask bps;bps.c_bp new BlockingQueueCalTask();bps.s_bp new BlockingQueueSaveTask();// BlockingQueueCalTask *task_bq new BlockingQueueCalTask();//两个阻塞队列// BlockingQueueSaveTask *save_bq new BlockingQueueSaveTask();pthread_t c, p, s;pthread_create(p, nullptr, producter, bps);pthread_create(c, nullptr, consumer, bps);pthread_create(s, nullptr, saver, bps);pthread_join(c, nullptr);pthread_join(p, nullptr);pthread_join(s, nullptr);delete bps.c_bp;delete bps.s_bp;return 0;
}我们可以直接改成多线程模式这里就不贴代码了因为push和pop里面有锁所以多个生产者和多个消费者也是竞争关系有锁了就不会引发数据不一致等错误
所以就提出来了两个问题
1、创建多线程生产和消费的意义 在多线程环境下面避免不了多个执行流访问同一个共享资源的情况但是如果一个执行流长时间持有锁或者它抢占锁的能力更强就会导致其他线程出现饥饿问题 —— 而生产消费模型是最常见解决这种问题的方法 2、生产消费模型高效在哪里 对于生产者 生产者的任务从哪里来呢 - 从数据库从网络从外设等等地方拿来的用户数据 - 所以生产者获取任务和构建任务是要花时间的 对于消费者 消费者拿到任务之后后续有很大概率处理任务非常非常耗时
所以生产消费模型高效并不是体现在阻塞队列拿任务/数据高效。而是让多个线程可以同时的去执行/计算多个任务/数据并且还不会影响到其他的线程继续从任务队列里面拿任务同理生产者拿数据的时候获取任务和构建任务要花费时间而这个时间内其他的线程也可以拿数据到队列里面并且不会影响到其他的生产者线程 生产消费模型高效在哪里 可以在生产之前消费之后让线程并行执行 9、总结
本节内容比较多所以分了几篇文章来进行讲解后面也都是重要的内容所以要一步步把基础的知识学好理解清楚不然后面学起来很吃力
