博罗网站设计,wordpress文章视频,公司网站推广方式,设计广告图用什么软件好用前言 本文主要介绍多线程基础知识#xff0c;以及使用多线程技术进行并发编程#xff1b;最后会介绍生产者消费者模型#xff1b; 一、线程基本认识
1、什么是线程 如果你是科班出生#xff0c;你肯定听过线程相关概念#xff1b;但是你可能没有真正搞懂什么是线程#…前言 本文主要介绍多线程基础知识以及使用多线程技术进行并发编程最后会介绍生产者消费者模型 一、线程基本认识
1、什么是线程 如果你是科班出生你肯定听过线程相关概念但是你可能没有真正搞懂什么是线程在认识线程之前你得知道什么是进程进程我们已经前面介绍过了这里只是简单阐述进程是操作系统分配资源的基本单位我们通常将进程相关内核数据结构 内存中该程序的代码称作进程也有好多课本说进程是程序运行时某一时刻的快照我们可以理解为进程在运行过程中其内核数据会不断发生变化的关于进程相关内核数据前面我们已经提及如PCB控制块、页表、进程地址空间、描述当前进程已经打开的文件结构体 files_struct 等这些都是我们进程的内核数据代码就更好理解了这里的代码是我们写的程序经过编译后形成的二进制机器指令 有了上面概念的铺垫我们对线程的理解就很简单了首先我们来看看课本上的概念线程是处理机调度的基本单位线程是进程的一部分听了这些概念肯定还是模模糊糊这是因为我们无法通过这些生硬的概念来理解下面我用几张图带着大家逐步理解我们的线程 上图是我们之前将的进程相关内核数据Linux我们学习完进程和文件应该脑海有上面这样的一个整体结构 关于线程我们是否也要拿一种数据结构把他描述起来呢当然是肯定的我们可能需要保存线程的编号线程的私有上下文等等信息我们计算机中当然也不可能只有一个线程肯定有大量的线程既然有大量的线程我们的操作系统是否需要将其组织起来呢当然也是肯定的既然要组织起来肯定要拿某种数据结构将其组织起来我们的window就是采用上述这种方案为线程设计一种独立的结构体然后将其组织起来而我们的Linux则提出一种更巧妙的思路 Linux系统下并没有单独的线程结构体那么线程又是如何描述组织起来的呢Linux下每当我们在当前进程下创建一个线程时操作系统会为当前进程创建一个PCB控制块且并不会申请新的虚拟地址空间等内核数据而是与原来的进程共用一个虚拟地址空间、页表、已打开文件结构体等内核数据结构如下图所示 Linux下并不称这些一个又一个新的执行流为线程而是称作轻量级进程因此操作系统也不会提供线程相关接口而是提供轻量级进程相关接口那我们平常在操作系统课本学的都是线程啊可并没有什么轻量级进程相关概念那我们也不会使用这些轻量级进程相关接口啊我们也不懂怎么使用此时Linux为我们提供了一个第三方库这个库中就是对轻量级进程进行包装包装成我们今天看到的线程 我们再从概念上理解线程首先概念上说线程是处理机调度的基本单位我们要马上反应过来我们的CPU是如何看待一个进程的我们CPU并不关心进程内部是怎么样的它仅仅是通过PCB控制块来决定待会儿执行什么指令因此在我们CPU看来并不认识什么线程、还是进程它只是通过读取PCB控制块数据来找到我们的虚拟地址空间、页表等数据然后找到代码起始地址然后逐句执行故我们的线程是处理机调度的基本单位 线程在进程内这句话就更好理解了我们线程只是进程的一部分它与整个进程共享资源 2、再次理解进程 很多人就对我们进程的理解就很模糊了上面一个又一个执行流是进程而我们之前理解的进程呢首先我们之前理解的进程为内部只有一个执行流的进程今天我们知道了进程内部可能有很多执行流每一个执行流都可称作一个单独的线程而这些所有线程 进程内核数据结构 代码 才被称作一个进程今天我们学习的线程是对进程概念的进一步补充并不违背我们之前学过的进程概念 我们还讲过进程是资源分配的基本单位我们向操作系统申请资源是以进程为基本单位的当我们创建一个进程时会创建虚拟地址空间、页表等内核数据这就是操作系统进行的资源分配而我们的线程并不会创建这些内核数据只是创建一个PCB控制块与我们进程共享这些内核数据罢了 综上所述我们不难理解上面我们的线程是内核给我们提供的轻量级进程的封装我们在用户层将这个轻量级进程封装起来这时我们才成这个是线程那所谓的轻量级轻量在哪里呢首先我们进行线程切换时不需要切换内核数据缓存比如我们虚拟地址空间、页表等内核数据都会在CPU里的寄存器中缓存起来而我们的进程切换不同这些数据肯定是不同的因此需要进行切换 3、线程的优缺点
1优点 创建一个新线程的代价要比创建一个新进程的代价小得到线程切换比进程切换所需操作系统做的事要少很多线程占用资源比进程少提高并发度 2缺点 缺乏访问控制某个线程调用某些系统调用可能对整个进程造成影响编写难度提高调试不易 4、线程的异常 在多线程程序中若有一个线程因为除零、野指针等问题崩溃会造成整个进程崩溃也就会引起所有线程退出 5、线程的数据 关于线程有自己独立数据也有与其他线程共享数据线程间通信相比进程容易很多因为线程共享同一进程地址空间下面来介绍线程私有数据与共享数据 私有数据 线程ID 一组寄存器寄存器中数据值 栈 errno 信号屏蔽字 调度优先级 注意其中一组寄存器强调寄存器里的数据也就是上下文数据 共享数据 文件描述符表 信号处理动作 当前工作目录 用户id与组id 堆 代码段数据 二、线程相关接口
1、线程控制 前面我们提过Linux并未为我们提供线程相关接口只是为我们提供了轻量级进程接口故Linux又为我们用户提供了第三方库的形式来使用线程这个线程库叫 pthread 通过我们Linux会自带这个库文件既然是第三方库文件我们在编译时必须告诉 gcc/g 编译器我们要用的库名称这里所以要加上 -l 库名 这个选项 1创建线程 关于创建一个线程接口声明如下所示 int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg); 参数一类型为 pthread_t 是一种自定义类型我们称这个为线程id是一个输出型参数 参数二此参数为我们线程的属性我们填NULL表示我们使用默认属性即可 参数三此参数为函数指针线程启动后执行的函数 参数四此参数为我们为参数三函数指针传入的参数无参可填写NULL 返回值调用成功返回0失败返回错误码 注意 1、这里返回值特别注意一般系统调用成功返回0失败返回-1错误码被设置而多线程系类函数不同后面所有函数返回值没有特别提及都是这样 2、关于线程参数一返回的线程id我们后面会做详细讲解这里当作一个线程句柄即可 3、pthread系类都需要带额外编译选项-l pthread因为这是第三方库 2获得自己的线程id pthread_t pthread_self(void); 该函数无参数返回值为当前线程的id该函数总是调用成功的 3线程终止 关于线程终止我们有三种方案如下所示 方案一调用return语句直接返回主函数也就是线程终止 方案二调用 pthread_exit 函数终止当前线程具体声明如下所示 void pthread_exit(void *retval); 参数一该参数为返回给主线程的返回值就与子进程返回给父进程的退出码类似 注意 1、线程返回的值必须是全局的或者是malloc等调用在堆上申请的空间若是栈上的当线程退出后线程的栈上的值也随之失效 2、不可用exit系列函数替代因为exit是退出进程的函数一旦调用该进程下的其他线程也会跟着退出 方案三调用 pthread_cancel 函数取消一个正在执行的线程具体声明如下所示 int pthread_cancel(pthread_t thread); 参数一该参数就是我们之前创建线程输出型参数返回的线程id 注意该函数是取消一个执行中的线程调用要确保线程执行中 4线程等待 首先我们要搞清楚一个问题我们为什么要进行线程等待不等待不可以吗这个问题的答案与我们前面讲解的进程等待的原因是类似的 1、已经退出的线程会处于一种类似僵尸状态我们必须进行回收不然会造成资源泄漏 2、我们可以通过线程等待获取线程退出值 关于线程等待相关接口声明如下所示 int pthread_join(pthread_t thread, void **retval); 参数一参数一为进程的id 参数二参数二为一个二级指针主要接收线程返回的一级指针、 注意关于参数二的返回值有以下几种可能 1、线程调用return语句正常返回则该参数的值为返回值的地址 2、线程是因为别的线程调用pthread_cancel而退出的此时该参数指向的单元存放的是常数PTHREAD_ CANCELED 3、线程是因为自己调用 pthread_exit 而退出的此时该参数指向的单元存放的是 pthread_exit 的参数 4、若对线程的返回值不关心参数二可以直接传入NULL 5线程分离 通常情况下一个线程默认是joinable的线程退出后若不join则会造成资源泄漏但是我们有时不想关心这个线程不想join我们可以将该线程进行分离操作线程退出后无需等待自动释放线程资源具体接口如下所示 int pthread_detach(pthread_t thread); 该函数只有一个参数为线程的id除了其他线程可以调用将该线程分离自己也可以调用自己我们可以执行以下语句进行自己分离 pthread_detach(pthread_self()); 2、测试代码 下面为一段多线程测试代码初步了解多线程使用不要跳过这里讲解线程id #include iostream
#include pthread.h
#include sys/types.h
#include unistd.h#define NUM 3void * threadRoutine(void* args)
{int count 5;while(count--){printf(分支线程执行中, tid: %u[%p], pid: %d\n, pthread_self(), pthread_self(), getpid());sleep(1);}return nullptr;
}int main()
{// 定义线程句柄pthread_t tid[NUM];// 创建线程for(int i 0; i NUM; i){pthread_create(tid i, nullptr, threadRoutine, nullptr);}int count 3;while(count--){std::cout main thread, pid: getpid() std::endl;sleep(1); }// 线程等待for(int i 0; i NUM; i)pthread_join(tid[i], nullptr); // 对线程返回结果不关心return 0;
} 注意编译时一定要加 -l 库文件选项不然会有以下报错 当我们加上编译时加上参数后可以正常编译如下所示 运行结果如下图所示 观察上图我们不难发现主进程的pid和分支线程的pid相同说明子线程在进程内部运行时我们还可以在命令行使用 ps -aL | head -1 ps -aL | grep ./main 确实有四个进程在跑其中LWP就是线程号并不是我们之前的 线程id其中主线程的pid与lwp相同那么我们前面代码中的线程id是什么呢 仔细观察上图我们打印tid更像一个地址实际上他就是地址当我们创建一个线程时pthread库会在用户层替我们维护一个线程结构该结构保存每个线程的线程栈以及线程相关数据每创建一个线程就会申请一个线程结构而这个线程id就是这个结构的起始地址 三、线程互斥
1、概念补充 临界资源多线程执行流共享的资源 临界区访问临界资源的代码 互斥任何时刻都只能有一个执行流访问临界资源 原子性不可被打断对于一种操作只有两种结果要么没做要么已经完成了我们称这种操作具有原子性、 可重入函数多个线程访问这个函数的情况下不会产生数据不一致等结果下面这个图来解释关于数据不一致 稍微解释一下上面四张图其中图一和图二表示线程一调用链表头插函数而线程一调用头插函数还没调用完时此时时间片到了然后线程二继续执行链表头插函数线程二很幸运执行完了头插函数若我们此时线程一继续执行后面的代码会造成newnode2丢失的场景我们称这种会造成数据不一致的函数为不可重入函数而对于不会造成数据不一致的为可重入函数 2、代码引入 我们使用接下来这段代码来演示多线程下并发访问共享资源带来的数据不一致问题 #include iostream
#include pthread.h
#include unistd.h// 定义抢票线程个数
#define NUM 3
// 定义全局变量, 表示票的总数
int tickets 10000;// 抢票线程执行函数
void* threadRoutine(void* args)
{int num *(int*)args;while(true){if(tickets 0){usleep(1000);printf(线程%d get ticket %d\n, num, tickets);tickets--;}else{break;}}
}int main()
{// 1、创建NUM个抢票线程pthread_t tids[NUM];for(int i 0; i NUM; i){pthread_create(tids i, nullptr, threadRoutine, (void*)(i));usleep(1000);}// 2、线程等待for(int i 0; i NUM; i){pthread_join(tids[i], nullptr);}return 0;
} 我们用多线程进行抢票操作可结果却不如我们所料每次运行的结果都不一样且居然还出现了负数的票 为什么会造成上面的现象呢就是由于我们的抢票函数是一个不可重入函数而我们用多个线程同时调用当然有可能产生数据不一致的问题下面我来演示一下是如何产生数据不一致的问题的 假设t1线程先得到了调度将我们的数据进行了一次减减也就是如上的三个小步骤接着当我们t1刚将 9999 放入CPU寄存器中并进行-1计算后发生了调度t1线程进行现场保护后被调出了内存换t2线程执行 t2开始执行自己得代码也是先将自己得ticket放入CPU寄存器然后减减然后放回内存如下所示 此时其实已经就出现问题了已经有两张9998得票了接着继续t2线程比较幸运将我们得9998减到了5000当t2线程将5000从内存取到寄存器并完成减减操作后突然又发生了调度t2要进行现场保护将自己得4999也要保存起来如下所示 t1运行后将自己上下文数据放回寄存器中从上次中断处的下一行代码接着执行将自己的数据放回内存中如下所示 好家伙t2线程好不容易将数据减到5000结果t1线程一下就改到了9998类似这种问题引发了上面我们看到的现象因此为了防止这种现象我们得使这些访问共享资源的进程互斥的进行访问 3、接口介绍 关于线程互斥我们是通过锁来完成的也称互斥量我们最常用的锁就是互斥锁互斥锁就像一把钥匙我们在访问临界资源前也就是进入临界区前我们为临界区加锁出临界区前我们要解锁用下面这个例子来介绍互斥锁 你们学习有一个VIP影院这个影院呢只有一个座位因此每次只能进去一个人钥匙放在影院门口每个人想要进入这个VIP影院得拿到钥匙规定必须有钥匙才可以进入影院如果出影院前必须要把钥匙挂到门口墙上因此有很多人一大早就来抢钥匙没有抢到的就在门口待着 上述故事模型中影院座位我们就可以看作临界资源而门口那个钥匙便是我们得互斥量在上述规则体系下任何时刻都只能有一个人能使用VIP影院也就是只有一个线程能访问临界资源且出临界区前必须归还钥匙也就是归还互斥量我们把进临界区那个动作称为加锁出临界区那个动作称为解锁 1互斥量初始化与销毁 互斥量的类型为 pthread_mutex_t我们要对其初始化有两种方案 1、若互斥量为全局变量 pthread_mutex_t mtx PTHREAD_MUTEX_INITIALIZER; 2、若为局部变量 pthread_mutex_t mtx; pthread_mutex_init(mtx, nullptr); 其中初始化函数有两个参数第一个参数是互斥量锁的地址第二个变量是所得属性我们通常设置成NULL表示默认属性返回值还是依照上面pthread系列返回值规则 通常我们局部变量的锁我们在使用完毕后我们要对其进行销毁动作我们需要调用上图中pthread_mutex_destroy接口其中参数为锁的地址 2互斥量的加锁与解锁 如下图所示其中前两个是加锁函数后面那个是解锁函数三个函数的参数都为锁的地址 这里我们介绍一下第二个 pthread_mutex_trylock 这个函数若没有申请到锁资源则立即返回一个非零错误码若申请成功则返回0而我们的 pthread_mutex_lock 函数若没有申请到锁则会阻塞直至能申请到锁资源为止 4、代码纠正 有了互斥量我们就可以通过互斥量的方案来实现线程互斥来保证不会出现数据不一致的问题如下所示 #include iostream
#include pthread.h
#include unistd.h// 定义抢票线程个数
#define NUM 3
// 定义全局变量, 表示票的总数
int tickets 10000;// 定义全局变量锁
pthread_mutex_t mtx PTHREAD_MUTEX_INITIALIZER;// 抢票线程执行函数
void* threadRoutine(void* args)
{int num *(int*)args;while(true){// 加锁pthread_mutex_lock(mtx);if(tickets 0){usleep(1000);printf(线程%d get ticket %d\n, num, tickets);tickets--;pthread_mutex_unlock(mtx); // 解锁}else{pthread_mutex_unlock(mtx); // 解锁break;}}
}int main()
{// 1、创建NUM个抢票线程pthread_t tids[NUM];for(int i 0; i NUM; i){pthread_create(tids i, nullptr, threadRoutine, (void*)(i));usleep(1000);}// 2、线程等待for(int i 0; i NUM; i){pthread_join(tids[i], nullptr);}return 0;
} 注意上面代码中我们写了两个解锁这是因为确保我们使用完临济资源后一定要进行解锁而在上述代码中执行流可能执行if语句也可能执行else语句所以这里两个语句块里都必须要有解锁动作运行结果如下 这时有小伙伴又有疑问了那为什么不直接放到while循环外面这样不就不用搞这么麻烦了吗 我们首先要搞清楚要是我们放在了while循环外面我们一旦进入临界区就是整个while循环结束才会出来而出while循环的条件是票被抢完了那么这么做就只有一个线程抢票了其他线程都申请不到所资源这样显然是不合理的 但实际上我们这代码还是有一点不合理的地方我们发现有时候总是只有一个线程申请得到锁关于这个问题后面我们介绍线程同步时再讲解 5、互斥量原理 互斥量锁的实现原理主要是利用一条汇编指令来实现的首先我们要清楚一条汇编指令一定是原子的其伪代码如下图 假设此时有两个线程我们分别称为a与b在访问临界资源前都需要加锁我们假设一种场景a先执行lock第一句汇编接着被切走运行bb执行前两条接着又被切走运行a此时是什么样的呢 因此无论如何切换只有一个1只有持有1的线程归还将mutex值设置为1此时才算归还锁资源别的线程才可以继续竞争当然自己也可以在继续竞争 总结这种由于多个线程共同访问共享资源而导致数据不一致的问题我们也称线程安全问题我们通常使用锁来解决当然还有信号量也可以解决后面会介绍 6、死锁 这是由于我们上面加锁方案繁衍出的一种问题当我们多个执行流相互等待对方资源释放而可能会产生死锁问题以下面为例 如上图线程A先使用共享资源B再使用共享资源A线程B反之要使用共享资源为了确保数据安全我们可采用加锁方案上述有两种共享资源因此我们访问资源A得申请锁A申请资源B得申请锁B若两个线程都顺序执行线程A得到共享资源B得使用权并加锁然后轮到线程B调度接着线程B也得到共享资源A的使用权正准备申请共享资源B时发现此时无论如何申请也不可能成功因为此时资源B的使用权在线程A身上而接着调度回线程A时线程A也阻塞了因为资源B的使用权在线程B身上因此两个线程都无法继续向后推进运行这样便造成了死锁的现象 死锁的四个必要条件 1、互斥条件一个共享资源只能被一个执行流访问 2、请求与保持每个执行流都不愿意释放自己锁资源而请求其他锁资源 3、不剥夺条件当前执行流无法剥夺其他执行流的锁资源 4、循环等待条件执行流循环等待对方锁资源 只要破坏上述条件中任意一个都可以解除死锁问题 四、线程同步
1、概念引入 在上述买票代码运行的结果中我们发现经常会产生一个线程买很多票的情况如下图所示 面对这种情况显然时很不合理的首先面对上面这种现象我们要搞懂以下几个问题 为什么会出现这样的现象呢 首先我们要清楚的是我们无法确保线程调度的顺序这是由CPU调度器决定的而当我们一个线程加锁后访问临界资源然后出临界区释放锁资源在这之后有没有可能又申请到锁资源了呢以上面VIP电影院的例子为例我有没有可能在刚出电影院然后将锁放在门旁边的时候突然发现门外这么多人排队然后我又继续持有锁打开门继续进入电影院的观影的情况呢因为我离钥匙最近因此我最有概率得到这把钥匙完全有可能我又再次进入VIP影院继续观看电影上面我们写的代码也是如此这种代码并没有错误只是不合理 为了解决上述问题我们引入线程同步的概念我们让这些执行流相互协调配合有序的推进执行而不会出现由于某个执行流因其他进程反复申请释放而导致的饥饿问题 2、接口介绍 关于实现线程同步我们通常采用条件变量方案下面正是介绍条件变量相关接口 1条件变量初始化与销毁 条件变量的类型是 pthread_cond_t 初始化同样也有两种方案与我们互斥锁类似 1、若条件变量为全局变量 pthread_cond_t cond PTHREAD_COND_INITIALIZER; 2、若条件变量为局部变量 pthread_cond_t cond; pthread_cond_init(cond, NULL); 此时我们需要对条件变量进行销毁接口如下 int pthread_cond_destroy(pthread_cond_t *cond); 其中初始化函数的参数一为条件变量的地址参数二为条件变量的属性这里我们常常设置成默认属性直接填写NULL即可 销毁条件变量的函数的参数使条件变量的地址 2条件变量的等待 线程等待主要的作用是使当前线程在某个条件变量去排队等待等待下一次被唤醒 int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex); 其中参数一为条件变量的地址参数二为所访问临界资源的互斥锁可能听到这就有点蒙圈了不要急看完后面条件变量的使用会清晰很多后面我们会加入条件变量的应用场景 3条件变量的唤醒 前面的接口可以使一个线程到指定的条件变量下排队接下来介绍的函数就是唤醒条件变量下的线程主要有如下两个接口 int pthread_cond_broadcast(pthread_cond_t *cond); // 唤醒条件变量下所有进程 int pthread_cond_signal(pthread_cond_t *cond); // 唤醒条件变量下的任一线程 上面函数中参数均为条件变量的地址 3、代码引入 接下来我们用两段代码来演示条件变量接口的使用 #include iostream
#include string
#include pthread.h
#include unistd.h// 定义线程数量
#define NUM 3// 定义线程数据
class ThreadDate
{
public:ThreadDate(){}ThreadDate(std::string name, pthread_mutex_t* pmtx, pthread_cond_t* pcond):_name(name),_pmtx(pmtx),_pcond(pcond){}
public:std::string _name;pthread_mutex_t* _pmtx;pthread_cond_t* _pcond;
};// 线程运行函数
void* threadRoutine(void* args)
{ThreadDate* pd (ThreadDate*)args;while(true){// 加锁准备进入临界区pthread_mutex_lock(pd-_pmtx);pthread_cond_wait(pd-_pcond, pd-_pmtx);std::cout pd-_name 正在运行..... std::endl;pthread_mutex_unlock(pd-_pmtx);}// 记得delete掉数据不然会造成内存泄漏delete pd;return nullptr;
}int main()
{// 初始化条件变量pthread_cond_t cond;pthread_cond_init(cond, nullptr);// 初始化互斥锁pthread_mutex_t mtx;pthread_mutex_init(mtx, nullptr);// 创建线程pthread_t tids[NUM];for(int i 0; i NUM; i){// 初始化线程数据char name[128];sprintf(name, thread-%d, i 1);ThreadDate* pd new ThreadDate(name, mtx, cond);pthread_create(tids i, nullptr, threadRoutine, pd);}// 唤醒任意线程while(true){// 唤醒该条件变量下的一个线程pthread_cond_signal(cond);sleep(1);}// 进程等待for(int i 0; i NUM; i){pthread_join(tids[i], nullptr);}// 销毁互斥锁和条件变量pthread_mutex_destroy(mtx);pthread_cond_destroy(cond);return 0;
} 我们在主线程随机唤醒等待在该条件变量队列中的任意线程这有点像我们后面要实现的线程池代码这里只是带着大家一起使用熟悉一下条件变量接下来的生产者消费者模型会带着大家一起应用条件变量 五、生产者消费者模型
1、模型介绍 我以我们日常生活来介绍生产者消费者模型想必大家平常都有线下购物的经历吧通常我们去超市进行购物可以说超市是我们的购物场所我们各位去购物都有一个身份那就是消费者那么是谁给我们提供的商品呢是超市吗并不是是生产商给我们提供的商品生产商也有一个身份就是生产者结合我们生活的例子我们发现对于一个超市来说肯定也不止一个生产者也不止一个消费者如下图所示 对于上述模型我们首先要记住三点 三个种关系生产者与消费者、生产者与生产者、消费者与消费者 二种角色生产者、消费者 一个交易场所超市 有了如上三点我们便能实现一个生产者消费者模型我们着重介绍三种关系如下 消费者与消费者互斥关系竞争 有时候超市经常会给我们做促销促销商品一般都是限量的因此我们的消费者可能会竞争同一种商品此时我们便要提供一种策略来限制非法竞争的消费者因此消费者之间是互斥关系 生产者与生产者互斥关系竞争 对于超市来说场所的大小是有限的且有的为止可能更容易让客户看到因此生产者们都想要一个好的位置让自己商品的销量更多因此我们生产者之间也存在着竞争既然存在我们就要进行管理防止恶意竞争所以我们生产者之间也是互斥的 生产者与消费者互斥与同步 对于超市每一个位置来说商品的生产与消费不是同时比如我们正在挑选某个位置的商品生产者就不能往我准备购买的商品的位置填充新的商品只有我拿走以后生产者才可以继续填充这就是两者之间的互斥关系 假如我们某一天特别想要超市的指定某样商品而我们去超市的时候发现卖光了结果我们第二天来发现还是每补货接着第三天、第四天、第五天以此类推我每天都来发现都没有这样显然是不合理的不仅浪费消费者时间也占用超市资源因此我们最好是给超市销售留下电话等生产者补充货物以后让销售给我们打电话通知我们来购物这样才是一种合理的方案同样反观生产者若其生产的货物还没有被消费也需要等消费者消费以后然后超市工作人员打电话给生产者让生产者提供商品这才是一种有效方案这也就是两者之间的同步关系没错但不合理 2、基于阻塞队列的生产者消费者模型 我们可以设计一种阻塞队列作为我们的交易场所我们消费者在消息队列里拿数据生产者在消息队列里放数据如下所示 #include iostream
#include string
#include time.h
#include unistd.h
#include BlockQueue.hpp// 生产者线程数量
#define PRO_NUM 3
// 消费者线程数量
#define CON_NUM 3// 线程所传参数数据类
class ThreadDate
{
public:ThreadDate(std::string name, BQ::BlockQueueint* bq):_name(name),_bq(bq){}
public:std::string _name;BQ::BlockQueueint* _bq;
};void* productorRoutine(void* args)
{ThreadDate* pd (ThreadDate*)args;while(true){// 模拟生产数据int num rand() % 10; pd-_bq-push(num, pd-_name);//sleep(1);}// 释放堆上申请空间delete pd;return nullptr;
}void* consumerRoutine(void* args)
{ThreadDate* pd (ThreadDate*)args;while(true){int num; // 接受要消费的数据pd-_bq-pop(num, pd-_name);sleep(1);}// 释放堆上申请空间delete pd;return nullptr;
}int main()
{// 种下随机数种子srand((unsigned int)time(nullptr));// 初始化阻塞队列BQ::BlockQueueint* bq new BQ::BlockQueueint();// 初始化生产者消费者线程pthread_t tids_pro[PRO_NUM];pthread_t tids_con[CON_NUM];for(int i 0; i PRO_NUM; i){// 线程名char name[128];sprintf(name, thread-productor-%d, i 1);// 线程所需数据ThreadDate* pd new ThreadDate(name, bq);pthread_create(tids_pro i, nullptr, productorRoutine, pd);}for(int i 0; i CON_NUM; i){// 线程名char name[128];sprintf(name, thread-consumer-%d, i 1);// 线程所需数据ThreadDate* pd new ThreadDate(name, bq);pthread_create(tids_con, nullptr, consumerRoutine, pd);}// 等待线程for(int i 0; i PRO_NUM; i){pthread_join(tids_pro[i], nullptr);}for(int i 0; i CON_NUM; i){pthread_join(tids_con[i], nullptr);}// 释放堆上申请内存delete pd;return 0;
}
// BlockQueue.hpp文件
#pragma once
#include queue
#include pthread.hnamespace BQ
{// 阻塞队列默认长度const int default_num 5;// 阻塞队列templateclass Tclass BlockQueue{public:BlockQueue(int capacity default_num):_capacity(capacity){// 初始化互斥锁与条件变量pthread_mutex_init(_mtx, nullptr);pthread_cond_init(_con, nullptr);pthread_cond_init(_pro, nullptr);}~BlockQueue(){// 销毁锁与条件变量pthread_mutex_destroy(_mtx);pthread_cond_destroy(_con);pthread_cond_destroy(_pro);}// 判断阻塞队列是否为空bool isEmpty(){return _bq.size() 0;}// 判断阻塞队列是否为空bool isFull(){return _bq.size() _capacity;}// 往阻塞队列放数据void push(const T data, const std::string name){// 加锁pthread_mutex_lock(_mtx);// 如果阻塞队列为满则无法放数据while(isFull()){// 线程等待等待唤醒pthread_cond_wait(_pro, _mtx);}// 运行到这里阻塞队列一定有存数据的位置_bq.push(data);std::cout name pruduct data - data std::endl;// 唤醒在消费者线程条件变量的一个线程来消费pthread_cond_signal(_con);// 解锁pthread_mutex_unlock(_mtx);}// 从阻塞队列拿数据void pop(T* data, const std::string name){// 加锁pthread_mutex_lock(_mtx);while(isEmpty()){// 线程等待等待唤醒pthread_cond_wait(_con, _mtx);}// 运行到这里阻塞队列里一定有数据可以消费*data _bq.front();_bq.pop();std::cout name consume data - *data std::endl;// 唤醒一个线程来生产数据pthread_cond_signal(_pro);// 解锁pthread_mutex_unlock(_mtx);}private:std::queueT _bq; // 阻塞队列int _capacity; // 队列长度pthread_mutex_t _mtx; // 维护互斥关系的锁pthread_cond_t _con; // 消费者条件变量pthread_cond_t _pro; // 生产者条件变量};
} 上述代码中我们的阻塞队列里放的是整型我们设计的是模板因此我们还可以替换成函数指针表示任务可以执行一项又一项的任务代码都有注释不懂还可以评论区留言 六、信号量
1、基本概念 信号量信号量本质就是一个计数器是实现进程同步与互斥的一种手段 我们通过初始化给信号量赋予初始值这个值通常为一个大于等于0的整型我们每次申请临界资源前我们首先申请信号量也就是执行P操作若信号量大于0此时不会阻塞并将信号量的值减一这个减一是原子的若我们使用完临界资源要出临界区则我们就执行V操作返还信号量的值也就是对信号量的值加一这个操作也是原子的 2、接口介绍 这里介绍POSIX版本信号量以及给我们提供的接口 1类型与初始化 同样我们使用信号量之前需要引入头文件 semaphore.h 信号量的类型为sem_t我们使用前需要对其初始化初始化函数声明如下 int sem_init(sem_t *sem, int pshared, unsigned int value); 注意同样我们需要编译时需要指定库名 pthread 参数一上面我们定义信号量的地址与上面互斥锁、条件变量类似这是一个输出型参数 参数二0表示线程间共享1表示进程间共享 参数三信号量的初始值就是我们上面说的数字后续会对这个数字加加或减减 返回值调用成功返回0失败返回-1错误码被设置 2销毁信号量 int sem_destroy(sem_t *sem); 参数信号量的地址 返回值调用成功返回0失败返回-1错误码被设置 3信号量的等待 该操作为申请信号量也就是判断信号量的值是否大于0若大于另则减减否则阻塞等待 int sem_wait(sem_t *sem); 参数信号量的地址 返回值调用成功返回0失败返回-1错误码被设置 4信号量的发布 所谓信号量的发布就是信号量的归还使用完临界资源后返还信号量对信号量进行加加操作 int sem_post(sem_t *sem); 参数信号量的地址 返回值调用成功返回0失败返回-1错误码被设置 3、基于环形队列的生产者消费者模型 环形队列想必大家都不陌生可以通过顺序表实现我们仅需维护一个数组数据起始下标和结束下表即可若对下标进行加加时我们注意模上数组大小即可 // Main.cc 文件
#include iostream
#include string
#include time.h
#include pthread.h
#include unistd.h
#include Ringqueue.hpp// 定义消费者线程数量
#define CON_NUM 3
// 定义生产者线程数量
#define PRO_NUM 3// 线程参数
class ThreadDate
{
public:ThreadDate(const std::string name, RQ::RingQueueint* rq):_name(name),_rq(rq){}
public:std::string _name;RQ::RingQueueint* _rq;};// 生产者执行线程函数
void* productorRoutine(void* args)
{ThreadDate* td (ThreadDate*)args;while(true){int num rand() % 10;td-_rq-push(num, td-_name);//sleep(1);}delete td;return nullptr;
}
// 消费者执行线程函数
void* consumerRoutine(void* args)
{ThreadDate* td (ThreadDate*)args;while(true){int num;td-_rq-pop(num, td-_name);sleep(1);}delete td;return nullptr;
}int main()
{// 种下随机数种子srand((unsigned int)time(nullptr));// 创建循环队列RQ::RingQueueint* rq new RQ::RingQueueint(5);// 创建线程pthread_t tids_pro[PRO_NUM];pthread_t tids_con[CON_NUM];for(int i 0; i PRO_NUM; i){char name[128];sprintf(name, thread-productor-%d, i 1);ThreadDate* td new ThreadDate(name, rq);pthread_create(tids_pro i, nullptr, productorRoutine, td);}for(int i 0; i CON_NUM; i){char name[128];sprintf(name, thread-consumer-%d, i 1);ThreadDate* td new ThreadDate(name, rq);pthread_create(tids_con i, nullptr, consumerRoutine, td);}// 线程等待for(int i 0; i PRO_NUM; i){pthread_join(tids_pro[i], nullptr);}for(int i 0; i CON_NUM; i){pthread_join(tids_con[i], nullptr);}// 销毁堆上申请资源delete rq;return 0;
}
// Ringqueue.hpp 文件
#pragma once
#include vector
#include string
#include semaphore.hnamespace RQ
{const int default_num 5;templateclass Tclass RingQueue{public:RingQueue(int capacity default_num):_capacity(capacity){// 初始化环形队列_rq.resize(capacity);// 初始化信号量sem_init(_data_sem, 0, 0); // 刚开始队列里没有数据故初始化为0sem_init(_space_sem, 0, _capacity); // 刚开始队列为空故初始化为capacity// 初始化下标值_start _end 0;// 初始化锁pthread_mutex_init(_c_mtx, nullptr);pthread_mutex_init(_p_mtx, nullptr);}~RingQueue(){// 销毁信号量sem_destroy(_data_sem);sem_destroy(_space_sem);// 销毁锁pthread_mutex_destroy(_c_mtx);pthread_mutex_destroy(_p_mtx);}void push(const T data, const std::string name){// 申请空间信号量P(_space_sem);// 加锁pthread_mutex_lock(_p_mtx);_rq[_end] data;_end % _capacity;std::cout name pruduct data - data std::endl;// 解锁pthread_mutex_unlock(_p_mtx);// 增加数据信号量V(_data_sem);}void pop(T* data, const std::string name){// 申请数据信号量P(_data_sem);// 加锁pthread_mutex_lock(_c_mtx);// 取走数据*data _rq[_start];_start % _capacity;std::cout name pruduct data - *data std::endl;// 解锁pthread_mutex_unlock(_c_mtx);// 增加空间信号量V(_space_sem);}private:void P(sem_t* sem){sem_wait(sem);}void V(sem_t* sem){sem_post(sem);}private:std::vectorT _rq; // 环形队列int _capacity;int _start; // 起始下标int _end; // 末尾下标sem_t _data_sem; // 数据信号量sem_t _space_sem; // 空间信号量pthread_mutex_t _c_mtx; // 消费者之间的互斥锁pthread_mutex_t _p_mtx; // 生产者之间的互斥锁};
} 上面所有代码均有注释若还是不大理解可私信小编看到会及时回复 七、线程池 可能大家多少都听说过线程池不知道也没有关系所谓线程池就是利用池化技术来减少我们进程向OS申请空间的次数以达到提高性能的效果关于池化技术我们可以这么理解比如我们以前在大学问父母讨要生活费时我们父母可能是一个月给你一次生活费但也有一种父母让你在每次需要花钱时给父母发消息然后父母在进行转账而这种转账效率太低了因此诞生了池化技术我们一次性让父母给一个月的生活费也可以理解为我们提前向父母透支这个月的生活费同样我们可以把这样的技术用到编码中我们提前向操作系统申请资源等我们需要该资源时我们直接拿着用就行而无需向操作系统申请 接下来的线程池也是这种技术我们有如下需求我们想要提前创建N个线程并用一种数据结构组织起来这里选择顺序结构也就是C中的vector然后我们用主线程模拟派发任务然后往任务队列放任务而我们每放一个任务就唤醒一个线程来处理这个任务如下图所示 具体代码如下所示可结合注释观看代码 // main.cc文件 主线程用于不断派发任务
#include iostream
#include unistd.h
#include Task.hpp
#include ThreadPool.hppint main()
{// 生成随机数种子srand((unsigned int)time(nullptr));// 初始化线程池ThreadPoolTask* tp new ThreadPoolTask(3);tp-run();// 生产任务while(true){int x rand() % 10;int y rand() % 10;Task t(x, y, [](int x, int y){return x y;});tp-push(t);std::cout x y ? std::endl;sleep(1);}return 0;
}
// lockGuard.hpp 文件采用RAII技术实现自动解锁出作用域
#pragma once
#include pthread.hclass Mutex
{
public:Mutex(pthread_mutex_t* mtx):_mtx(mtx){}void lock(){pthread_mutex_lock(_mtx);}void unlock(){pthread_mutex_unlock(_mtx);}
private:pthread_mutex_t* _mtx;
};class LockGuard
{
public:LockGuard(pthread_mutex_t* mtx):_mtx(mtx){_mtx.lock();}~LockGuard(){_mtx.unlock();}
private:Mutex _mtx;
};
// Task.hpp 文件任务类模拟任务
#pragma once
#include iostream
#include functional
using call_back std::functionint(int, int);
class Task
{
public:Task(){}Task(int x, int y, call_back cb):_x(x),_y(y),_func(cb){}void operator()(){int ret _func(_x, _y);std::cout _x _y ret std::endl;}
private:int _x;int _y;call_back _func;
};// Thread.hpp文件模拟封装我们自己的线程
#pragma once
#include cstdio
#include string
#include functional
#include pthread.htypedef void*(*func_t)(void*);class ThreadDate
{
public:ThreadDate(){}ThreadDate(const std::string name, void* args):_name(name),_args(args){}
public:std::string _name;void* _args;
};class Thread
{
public:Thread(int num, func_t func, void* args):_func(func){char buf[128];sprintf(buf, thread-%d, num);_td new ThreadDate(buf, args);}~Thread(){delete _td;}void start(){pthread_create(_tid, nullptr, _func, _td);}void join(){pthread_join(_tid, nullptr);}
private:std::string _name;pthread_t _tid;func_t _func;ThreadDate* _td;
};
// ThreadPool.hpp文件线程池核心代码
#pragma once
#include vector
#include queue
#include pthread.h
#include functional
#include Thread.hpp
#include LockGuard.hppconst int g_thread_num 3;
templateclass T
class ThreadPool
{
public:// 构造ThreadPool(int num g_thread_num):_num(num){// 初始化线程对象数组for(int i 0; i num; i){_threads.push_back(new Thread(i 1, threadRoutine, this));}// 初始化锁pthread_mutex_init(_mtx, nullptr);// 初始化条件变量pthread_cond_init(_cond, nullptr);}// 析构~ThreadPool(){// 线程等待for(auto it : _threads){it-join();delete it; // 注意我们线程是new出来的记得释放}// 释放锁资源pthread_mutex_destroy(_mtx);// 释放条件变量pthread_cond_destroy(_cond);}// 启动线程池里线程void run(){// 让线程启动for(auto it : _threads){it-start();}}// 往任务队列放数据相当于生产者void push(const T data){LockGuard lock(_mtx);// 往任务队列塞数据_tasks.push(data);// 唤醒线程来消费数据pthread_cond_signal(_cond);}
private:// 线程执行的线程函数这里必须设置成static如果不设置成static会默认传this指针// 这个相当于cp模型中的消费者static void* threadRoutine(void* args){// 获取传参ThreadDate* td (ThreadDate*)args;// 获取当前线程池ThreadPool* tp (ThreadPool*)td-_args;while(true){Task t;{LockGuard lock(tp-_mtx);// 判断任务队列是否有任务while(tp-_tasks.empty()) pthread_cond_wait(tp-_cond, tp-_mtx);// 执行到这里肯定有任务了t tp-_tasks.front();tp-_tasks.pop();}// 执行任务t();}}
private:int _num; // 线程数量std::vectorThread* _threads; // 保存thread的数组std::queueT _tasks; // 任务队列pthread_mutex_t _mtx; // 保护任务队列的互斥锁pthread_cond_t _cond; // 唤醒新线程来消费数据的条件变量
};
