-
Linux线程
从线程是什么，为什么要用和怎么用出发。下面探讨的都是linux下的线程🌲线程是什么
线程是进程的一个执行分支，是在进程内部运行的一个执行流，是操作系统进行运算调度的最小单位。
在linux里我们也把线程成为轻量级进程（LWP，LightWeightProcess）,因为linux里其实没有真正的线程，线程是通过进程模拟出来的（在内核里都是一个个的task_struct)。
没学线程前我们说进程是操作系统最小的调度单位，因为那时我们写的代码都是单线程的，一个进程只有一个执行流，所以那么说也没错，准确一点就是线程是操作系统调度的最小单位。
单执行流
多执行流
通过上面两张图得出的一些结论，同时补充一些概念：
- 站在CPU的角度，进程与线程没有任何区别，它看到的都是task_struct,这也是为什么线程在linux里也叫轻量级进程，也说明linux下没有真正意义上的线程。简单来说，CPU对线程0感知。
- 线程之间是共用一个地址空间的，这说明比起进程之间的切换，线程的切换更加轻量级。因为进程切换可能要保存页表、地址空间的数据甚至缓存的数据等等，而线程之间切换时这些都不用动，自然切换的代价也比较小。
- 进程：线程=1：n，说明系统内有大量的线程，所以操作系统必定要把线程管理起来，说明如果一个系统支持真正的线程，比如windows，那必然是有一个结构（TCB）来描述这个线程的属性，并且将其组织起来，但是往往比较复杂，linux下虽然没有真正的线程，但是优点就是简单。
- linux下没有真正意义的线程，这就说明OS不可能在系统层面提供操作线程的接口，而是一些封装好给用户的轻量级接口。
- CPU调度的都是线程，即线程是CPU调度的最小单位
- 站在系统的角度，进程是承担系统资源的基本单位。因为第二个线程用的资源都是第一个线程申请好的。
- 线程在进程内部运行，这句话的意思是线程在进程地址空间内运行。
- 通过页表可以看到物理内存，也即真正的资源，同时说明只要划分页表我们就可以让线程看到进程的部分资源。
- 关于页表，页表不仅仅记录了虚拟地址与物理地址的映射，还有一些别的属性，如权限，是否命中等等，这说明页表的大小不是一个字节就能记录的。一般32位的机器物理内存是4G，说明有232这么多个地址要映射，即页表要记录232个映射关系，表示每一个映射关系需要的字节都大于1，说明页表整体大小大于4G，放不进内存，此时就引出了二级页表。负责这块的硬件就是MMU，具体的了解可以查询二级页表的相关资料。
- 线程数越多越好吗？并不是，建议与计算机的核数相当。线程过多会导致大部分时间花在调度上，而没有花在线程的执行上，有点买椟还珠的意思。
- 没有线程替换，线程替换就是整个进程被替换。
- CPU不需要线程的概念，linux下线程这个概念是给用户的，因为用户需要多线程编程
- 用户层通过TCB来知道线程的id、状态、优先级和其他属性，用来进行用户级的线程管理。TCB不由内核维护，而是由用户空间维护。
随笔记录:一个程序被加载到内存，系统就创建了一个进程。请问这句话什么意思？为什么要先加载到内存？ (sogou.com)
🌲线程常见接口使用
线程的优缺点放在最后，直接贴在这里有点太抽象了。
相关接口的使用细节可以用man命令查询，下面直接贴基本用法。
🌵相关概念补充
线程之间有共享进程数据，但也有独有的资源。
共享的资源：进程代码段、进程的公有数据，进程所拥有的的资源。
**独有的资源：**寄存器内的数据，线程的独立栈，自己的状态（如自己的线程id、调度优先级、信号屏蔽字、errno等）。可以概括为上下文数据和独立的栈结构，上下文数据说明线程线程是可以被切换的，独立的栈结构说明线程是独立运行的。
进程地址空间里不是有一个栈了，那这个独立栈在哪？在共享区。
tid1和tid2两个值本质是一个地址，进程地址空间内的栈只能被主线程用，别的线程有自己的独立栈，在共享区内。
删除线程就是删掉内核级别的LWP，再释放共享区里的资源（数据结构）。
一个线程异常终止，会导致整个进程终止。比如线程中出现除零错误、野指针越界等，进程就会触发信号机制，系统就发送一个信号终止进程（因为信号是针对进程而不是针对线程的）。
了解：vfork()的使用，简单来说，创建子进程并让父进程阻塞，并且子进程共享父进程的地址空间，一般vfork出来的子进程都是为了替换，因为这种替换后不需要父进程的数据也就不需要像fork()一样去拷贝父进程的进程地址空间。具体使用可以查询资料和文档。
🌵pthread_create创建线程
创建一个新线程，成功返回0，失败返回一个错误码。
#includeint pthread_create(pthread_t *thread, const pthread_attr_t *attr,void *(*start_routine) (void *), void *arg); Compile and link with -pthread. - 1
- 2
- 3
- 4
thread:输出型参数，返回创建的线程的id。
attr:设置线程的属性，attr为NULL表示默认属性
start_routine：函数指针，表示线程启动后要执行的函数。函数返回值和参数都为void*。
参数是void*,"void * "被称为万能指针，可以帮助接收各种参数。
arg:传给线程启动函数的参数。
编译时要加lpthread选项，表示链接到pthread库
- 关于pthread_t类型，typedef uintptr_t pthread_t;uintptr_t是unsigned long int。但是我们不应该这么做，因为pthread_t是一个不透明的类型，根据平台不同实现也不同，所以不应该用无符号长整形去定义它。
这里我们只需知道线程ID的类型是pthread_t。判断两个线程ID相等也应该用相应的接口，而不是“==”，尽管线程ID打印出来是一个数字。
- 关于“Compile and link with -pthread.”
🍃例子
创建一个新线程，主线程每隔1s打印一次，新线程每隔2s打印一次。
#include#include #include using namespace std; void* routine(void* args)//新线程执行的函数 { while(true) { cout<<(char*)args<<endl; sleep(2); } return nullptr; } int main() { pthread_t tid; pthread_create(&tid,nullptr,routine,(void*)"thread 1");//创建新线程 while(true) { cout<<"main thread is running"<<endl; sleep(1); } 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
🌵pthread_self()获取线程自身id
获取当前线程的id，函数调用总是成功，返回值为线程id。
#includepthread_t pthread_self(void); Compile and link with -pthread. - 1
- 2
- 3
- 4
🍃例子1
每隔1s打印主线程和新线程的线程id
#include#include #include using namespace std; void* routine(void* args) { while(true) { cout<<"thread1 id: "<<pthread_self()<<endl; sleep(1); } return nullptr; } int main() { pthread_t tid; pthread_create(&tid,nullptr,routine,(void*)"thread 1"); while(true) { //cout<<"main thread is running"< - 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
🥝ps-aL命令查看系统轻量级进程
ps -aL //这里查看的进程是上面的例子- 1
可以看到进程id相同，因为两个线程属于同一个进程，主线程id和新线程id不同，主线程的PID和LWP相同。
可以看到例子里打印的线程pthread_t表示的线程ID和LWP表示的线程ID不同，直接给结论，LWP是系统唯一标识线程的ID，pthread_t变量的值本质是进程地址空间上共享区的一个起始地址，是用户层的线程id。
拿到用户级别的线程id，就可以在库里找到线程相关属性，包括线程的独立栈。
🍃例子2
循环创建多个进程，利用数组+指针让线程调用的函数知道自己是第几个线程。
#include#include #include using namespace std; void* routine(void* args)//参数为void*可以帮助接收各种参数 { while(true) { cout<<"thread"<<*(int*)args<<" id: "<<pthread_self()<<endl; sleep(2); } return nullptr; } int main() { pthread_t tid[5]; for(int i=0;i<5;i++) { int* p=new int(i); pthread_create(tid+i,nullptr,routine,(void*)p); } while(true) { //cout<<"main thread is running"< - 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
🌵pthread_join()等待线程
🥝为什么要等待线程呢？
- 防止内存泄漏（一直占着资源没释放）
- 让main线程最后退出（main没了别的线程还在跑不合理）
- 获的新线程的退出码（任务做的怎么样）
和等待进程类似，我们创建线程是给了他对应的任务，我们是需要知道这个线程把这个任务做的怎么样的，此外，已经退出的线程空间没有被释放，仍然会在进程的地址空间内占用着资源，可以保证创建新的线程不会复用旧线程的地址空间。
不等待会怎么样？资源一直没有被回收，造成类似僵尸进程的问题。
等待的时候要不要考虑线程崩溃的问题？不需要，线程崩了表示进程崩了。
🍬接口使用
pthread_join()等待线程，成功返回0，失败返回错误码。
#includeint pthread_join(pthread_t thread, void **retval); Compile and link with -pthread. - 1
- 2
- 3
- 4
- 5
第一个thread传进程标识符,这里不是传指针了.
第二个传参数retval传二级指针，接收线程执行函数的返回值.这个指针是一个输出型参数，可以获取进程的退出码，也可以获取更加详细的信息，比如用结构体描述退出信息。
线程执行的函数routine的返回值是void *，这说明返回值如果是数字（退出码）的话那就是void *,更改void *的值应该用void ** ，所以第二个参数为二级指针。
-
注意pthread_join是阻塞等待的，我们可以选择所有线程运行完了等待所有线程，也可以选择运行一个线程就等待一个线程。具体情况具体分析。
-
如果不关心线程退出的状态，第二个参数传空指针即可。
🍃例子
创建五个线程，用结构体来描述退出信息，主线程等待并获取退出信息。
#include#include #include #include using namespace std; struct exit_code { int code; string info; }; void* routine(void* args) { int cnt=3; while(cnt--) { cout<<"thread"<<*(int*)args<<" id: "<<pthread_self()<<" cnt: "<<cnt<<endl; sleep(1); } exit_code* p=new exit_code(); p->code=10; p->info="thread quit normal!"; return p; } int main() { pthread_t tid[5]; for(int i=0;i<5;i++) { int* p=new int(i); pthread_create(tid+i,nullptr,routine,(void*)p); // void* st=nullptr;//初始化一下,这里是一个线程跑完就等一个线程 // 创建一个线程后，主线程一直阻塞在这，不会再创建下一个进程，所以一个线程跑完等待完了才会创建下一个进程 // if(pthread_join(tid[i],&st)==0) // { // cout<<"thread exit code: "<<((exit_code*)(st))->code<<" "<<((exit_code*)(st))->info< - 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
如果想知道具体等待的是哪个进程可以在结构体传入线程ID。
🌵pthread_exit()终止线程
🥝终止线程的方法
- return
- pthread_exit
- pthread_cancel
不能用exit退出，exit是用来退出进程的
🍬接口使用
pthread_exit - terminate calling thread #includevoid pthread_exit(void *retval); Compile and link with -pthread. - 1
- 2
- 3
- 4
一个参数:指针retval。这个指针不能指向局部变量，不然线程退出局部变量那块空间也被“销毁”了。
🍃例子
#include#include #include #include using namespace std; struct exit_code { int code; string info; }; void* routine(void* args) { int cnt=3; while(cnt--) { cout<<"thread"<<*(int*)args<<" id: "<<pthread_self()<<" cnt: "<<cnt<<endl; sleep(1); } pthread_exit((void*)10);//设置退出码 } int main() { pthread_t tid[5]; for(int i=0;i<5;i++) { int* p=new int(i); pthread_create(tid+i,nullptr,routine,(void*)p); } for(int i=0;i<5;i++) { void* ret=nullptr;//初始化一下 if(pthread_join(tid[i],&ret)==0) { cout<<"thread exit code: "<<(int)(ret)<<endl; } } 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
🌵pthread_cancel()取消执行中的线程
pthread_cancel - send a cancellation request to a thread #includeint pthread_cancel(pthread_t thread); Compile and link with -pthread. - 1
- 2
- 3
- 4
- 5
- 6
参数thread表示要取消的线程ID
调用成功返回0，否则返回非0、
-
cancel本身具有一定的延时性，并不是立即处理的，所以主线程取消其他线程时得先保证其他线程跑起来了
-
如果一个线程被其他线程pthread_cancel异常终止掉，线程函数的返回值会被设定为PTHREAD_CANCELED，PTHREAD_CANCELED本质是一个宏
#define PTHREAD_CANCELED ((void *) -1)- 1
🍃例子
创建五个进程，2s后取消后三个线程。
#include#include #include #include using namespace std; void* routine(void* args) { int cnt=3; while(cnt--) { cout<<"thread"<<*(int*)args<<" id: "<<pthread_self()<<" cnt: "<<cnt<<endl; sleep(1); } pthread_exit((void*)10); } int main() { pthread_t tid[5]; for(int i=0;i<5;i++) { int* p=new int(i); pthread_create(tid+i,nullptr,routine,(void*)p); } sleep(2);//让线程都跑起来，不然新线程可能被创建了，但是还没有被调度，导致取消出问题 for(int i=2;i<5;i++) { pthread_cancel(tid[i]); } for(int i=0;i<5;i++) { void* ret=nullptr;//初始化一下 if(pthread_join(tid[i],&ret)==0) { cout<<"thread join success "<<endl; } } 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
🍃例子2
子线程干掉主线程导致主线程僵尸。此时如果子线程一直在跑就会导致内存泄漏（要模拟这个情况加上一个死循环即可）
#include#include #include #include using namespace std; pthread_t main_thread; void* routine(void* args) { int cnt=3; while(cnt--) { cout<<"thread"<<*(int*)args<<" id: "<<pthread_self()<<" cnt: "<<cnt<<endl; sleep(1); pthread_cancel(main_thread);//干掉主线程 } pthread_exit((void*)10); } int main() { main_thread=pthread_self(); pthread_t tid[5]; for(int i=0;i<5;i++) { int* p=new int(i); pthread_create(tid+i,nullptr,routine,(void*)p); } sleep(2); for(int i=2;i<5;i++) { pthread_cancel(tid[i]); } for(int i=0;i<5;i++) { void* ret=nullptr;//初始化一下 if(pthread_join(tid[i],&ret)==0) { cout<<"thread join success "<<endl; } } 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
左边是运行结果，右边是监控脚本
while :; do ps -aL |head -1 &&ps -aL |grep myproc; sleep 1; echo “############”; done- 1
随笔记录：进程为什么独立？因为有自己独立的地址空间 页表等。
🌵pthread_detach()分离线程
默认情况下，新创建的线程是joinable的，线程退出后，需要用pthread_join等待线程释放资源，如果不关心线程的返回值，join就是一种负担，此时我们可以告诉系统线程退出时自动释放资源。告诉系统线程退出时的操作就是分离。
分离的本质是让主线程不用再join新线程，让新线程退出的时候自动回收资源
接口使用
pthread_detach - detach a thread #includeint pthread_detach(pthread_t thread); Compile and link with -pthread. - 1
- 2
- 3
- 4
- 5
- 6
- 7
一个参数：分离的线程ID
- 一个分离的线程不应该被等待，被等待的结果是未定义的，可能成功有可能不成功。
- 线程设置为分离状态，如果该线程崩了进程一样会崩。
🍃例子
#include#include #include #include using namespace std; void* routine(void* args) { pthread_t(pthread_self()); int cnt=3; while(cnt--) { cout<<"new thread is running "<<endl; sleep(1); } return (void*)10; } int main() { pthread_t tid; pthread_create(&tid,nullptr,routine,nullptr); while(true);//不让主线程return 0，不然进程直接结束了 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
linux里主线程return 0结束的话，会自动结束其进程。
[linux下主线程return 0和pthread_exit(NULL)的区别](https://www.cnblogs.com/Stephen-Qin/p/12730670.html#:~:text=1.当linux和Windows中%2C主线程以return 0结束时%2C程序会在主线程运行完毕后结束.,2.当linux中%2C主线程以pthread_exit (NULL)作为返回值%2C则主线程会等待子线程.)
🌲线程同步和互斥
🌴相关概念
-
临界资源：多线程执行流共享的资源。（即多个线程都能看到的同一份资源）
-
临界区：访问临界资源的代码
-
互斥：任何时候保证只有一个执行流进入临界区访问临界资源，可以保护临界资源
-
原子性：不能被打断的操作，对于一个具有原子性的操作来说，要么做了要么没做
-
我们借助锁来实现互斥进而达到保护临界资源的效果。
🌴售票例子
定义一个全局变量，所有线程都可以访问，那全局变量就是一个临界资源，对全局变量做的修改，如++，–，就有风险，因为++和–不是原子性的。（从汇编上可以知道++和–都对应三条汇编代码，说明是可以打断的）
//全局变量a++对应的三句汇编指令 mov eax, DWORD PTR a[rip] add eax, 1 mov DWORD PTR a[rip], eax- 1
- 2
- 3
- 4
把这里的全局变量a换成售票时的票会发生什么？
下面创建五个线程模拟抢票时的场景。
票数++需要经过以下几个步骤：tickets从内存写入寄存器，CPU对寄存器里的tickets运算，算好的结果再写回内存，这三步都可能被中断，所以访问tickets这个临界资源是不安全的。
下面寄存器和CPU之间的箭头表示的是运算，不代表数据的拷贝。
如果tickets是1，CPU算完后在写回内存这一步被打断了，然后寄存器2相关的线程进来，把tickets改成了999，别的线程再看tickets本来应该是0的却被写成了999，就出现了问题。
#include#include #include #include using namespace std; int tickets=100 ;//临界资源 void* routine(void* args) { while(1) { if(tickets>0) { usleep(30000);//原因在这 printf("thread 0x%x: get a ticket %d\n",pthread_self(),tickets); tickets--; } else { break; } } printf("thread 0x%xquit! ticket %d\n",pthread_self(),tickets); return (void*)10; } #define NUM 5 int main() { pthread_t tid[NUM]; for(int i=0;i - 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
出现了负数，显然不合理。这就是典型的多线程切换的时候因为数据交叉操作导致的数据不一致问题。OS在内核态返回用户态的时候进行线程切换。这种现象的本质是ticket–不是原子性的。
本质就是多个执行流进入了if语句。
具体原因是因为抢票前睡了一下，系统就切到别的线程了，然后别的线程抢完票后票数已经小于0了，此时刚才睡的线程醒了过来又抢，自然就是负数了。举个例子，线程1进入了if语句开始了睡眠，切到了线程2,2去抢票，抢完后票数小于0，线程1苏醒，此时线程1已经在if语句里面，再去抢票票数就是负数了。
🌴互斥量pthread_mutex_t
为了解决这种数据不一致的问题，我们引入了锁。下面借助锁来解决这个问题，其实就是保证一次只有一个执行流访问临界区。
在linux里这把锁就叫互斥量。下面主要是锁的使用。
可以把临界区比作一个房间，一个人（执行流）进了临界区就上锁，那别的人都不能进来了，此时只有这一个人可以访问临界资源，其余人（执行流）全在门外等着解锁。用锁的代价很大，所以上锁和解锁的方式就挺有讲究的。
在内核里找互斥量的时候发现pthread_mutex_t被封装过了，不太好找🌵pthread_mutex_init和pthread_mutex_destroy
锁使用前先要初始化，用完也要销毁。
pthread_mutex_destroy, pthread_mutex_init - destroy and initialize a mutex #includeint pthread_mutex_destroy(pthread_mutex_t *mutex); int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr); pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; - 1
- 2
- 3
- 4
- 5
- 6
- 7
pthread_mutex_init的第一个参数是一个互斥量的指针，第二个参数attr表示属性，传NULL表示默认的属性。成功返回0，失败返回错误码。
pthread_mutex_init的参数是一个互斥量的指针，成功返回0，失败返回错误码。
此外，除了用pthread_mutex_init进行动态分配的初始化，还有一种静态分配的方法：、
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//PTHREAD_MUTEX_INITIALIZER本质是一个宏- 1
随笔记录：线程A的时间片到了，CPU的寄存器里还有数据，这部分数据就称作上下文数据，也就是临时数据
寄存器是线程共享的，数据是私有的。
🌵pthread_mutex_lock和pthread_mutex_unlock
上锁和解锁
pthread_mutex_lock, pthread_mutex_trylock, pthread_mutex_unlock - lock and unlock a mutex #includeint pthread_mutex_lock(pthread_mutex_t *mutex); int pthread_mutex_trylock(pthread_mutex_t *mutex); int pthread_mutex_unlock(pthread_mutex_t *mutex) - 1
- 2
- 3
- 4
- 5
- 6
- 7
关于 pthread_mutex_trylock会尝试对互斥量加锁，如果互斥量已经上锁了。函数调用失败，否则调用成功返回0
🥝解决票数为负数的问题
#include#include #include #include using namespace std; pthread_mutex_t lock; int tickets=100 ; void* routine(void* args) { while(1) { pthread_mutex_lock(&lock); if(tickets>0) { usleep(30000); printf("thread 0x%x: get a ticket %d\n",pthread_self(),tickets); tickets--; pthread_mutex_unlock(&lock); } else { pthread_mutex_unlock(&lock);//这必须解锁，不然的话可能上了锁没解锁，导致自己一直拿着这把锁导致阻塞 break; } } printf("thread 0x%x quit! ticket %d\n",pthread_self(),tickets); return (void*)10; } #define NUM 5 int main() { pthread_mutex_init(&lock,nullptr); pthread_t tid[NUM]; for(int i=0;i - 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
这种做法又衍生出了一个问题，一直都是一个人在买票。要解决这个问题需要在抢完票后让这个线程休息一会，比如睡一会做点别的工作之类的，别的进程才有可能进来抢票。
🥝相关概念补充
-
锁的存在是为了保护临界资源
-
锁的本身就是一种临界资源（大家都能申请锁），那谁来保护锁？只需保证lock和unlock是原子的即可。
-
一个线程上了锁，别的线程也能继续申请这把锁，只不过无法申请成功。
-
上了锁也会发生线程切换，但是切换后的锁资源会在pthread_mutex_lock这阻塞住（挂起），直到锁被解开。
🌴互斥量（锁）的原理
锁的使用
锁的原理：简单概括为01两种状态
写段伪代码理解一下
lock: movb $0,%al xchgb %al,mutex//xchgb是一条交换指令，等同于swap if(al寄存器内容>0) { return 0; } else { 挂起等待; } goto lock; unlock: movb $1,mutex 唤醒等待mutex的线程 return 0;- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
mutex为1时表示可以申请锁成功，当mutex为1时:
当mutex为0时:
-
解锁也好理解，把mutex置为1，下次就能申请成功了，再唤醒其他在等待的线程就行了。
-
可以看到只有一个1，这个1被拿走后只有解锁的时候才会还回来，别的线程在这期间都是拿不到1的
🌴死锁
一组进程中的各个进程均占有不会释放的资源，又因为相互申请被其他进程占用所不会释放的资源而处于的一种永久等待状态。
简单来说，自己手里的不释放，又要别人手里的，别人手里的也不释放，两个人就在这僵着。也就是多个进程再运行过程中争夺资源造成的一种僵局。
套入某种具体情况就是：请求你的锁，又不释放自己的锁。
我可以和你耗一整天。–美队
🥝死锁产生的条件
互斥条件：一个资源每次只能被一个执行流使用
请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
不剥夺条件：一个执行流已获得的资源，在未使用之前，不能强行剥夺
循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系。
一个资源只能一次被一个执行流使用，已获得的资源不能强行剥夺，同时申请别的资源阻塞时又不放弃自己的资源，若干执行流之间有循环等待的关系
🥝避免死锁
-
破坏死锁产生的条件
-
加锁顺序一致（比如锁1和锁2，线程1先上锁1再上锁2，线程2先上锁2再上锁1，就可能导致线程1拿着锁1去申请锁2，此时线程2拿着锁2去申请锁1，两个人都在那等，结果就死锁了）
-
避免锁未释放（避免有人一直拿着锁不放）
-
资源一次性分配（一次性把资源都分配好了，就不会去请求别人的资源了）
避免死锁的算法：死锁检测算法和银行家算法。（待了解
模拟死锁的思路:两个线程,线程a申请锁a,线程b拿到锁b,拿到后让两个线程睡一下,保证不会出现时序问题,融合让线程a申请锁b线程b申请锁a,这样满足循环等待就死锁了
🌴同步
上面讲了加锁保持互斥，下面提的就是同步。
保证数据安全的前提下让执行流以某种特定的顺序去访问资源。
从执行流的执行顺序的角度看，同步是一种更为复杂的互斥，互斥是多个执行流直接不可以同时访问一块临界资源，他们互相排斥，比如一个访问完了另一个再访问。同步则更为特殊，同步是以一种特定的顺序安排多个执行流访问一块资源。
🥝同步的必要性
以特定的顺序访问资源，有效避免了"饥饿问题"。
以抢票为例，如果一个线程锁的竞争能力很强，导致别的线程都抢不到票，这有错吗？没有错，但是不合理。
再以去自习室为例，一个自习室每次只能容纳一个人，一个人进去学完了出来后发现外面排了很长的队等着进来，结果他刚解锁出门一想要不我再学一下吧，又开门进去了，外面的人继续在外面等，导致自习室一直就是这一个人自习，没有错但不合理。同步解决的就是这种问题，抢票那就你抢完然后下一个人去抢，自习室你出来了马上到排队队伍后面去，要想再进来就要等前面人都自习完了。
竞争产生的原因是因为我们的操作不是原子的。
排队的本质：在安全的前提下获取锁，按照某种顺序进行申请和释放，这就是同步的过程
同步解决了资源分配不合理的问题，让线程有序的申请锁。
🌴条件变量
条件变量是线程库提供的一个描述临界资源的对象，是一个变量。
举个例子，有两个人A和B蒙着眼，一个人往盒子里放苹果，一个人去盒子里拿苹果，他们都不知道对方对方做到哪一步了，比如A不知道B放没放进去，只能一直伸手去试，也就是去检测盒子里有没有苹果，这就是轮询检测，苹果就是临界资源，A和B就是两个线程。
A在拿苹果时，B有三种状态，放前，放中，放后。放中的状态是不确定的所以需要拿苹果的人轮询检测。检测这个操作本身就需要资源。所以我们可以在B放完苹果后摇一下铃铛告诉A我放好苹果了，你来拿吧，这个铃铛就是条件变量。检测的本质就是我们不知道临界资源的情况，但需要我们通过某种手段知道。
条件变量使得不用频繁通过申请和释放锁的方式，也能达到检测临界资源的目的。
🌵pthread_cond_init和pthread_cond_destroy
和锁的初始化很像
pthread_cond_destroy, pthread_cond_init - destroy and initialize condition variables #includeint pthread_cond_destroy(pthread_cond_t *cond); int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr); pthread_cond_t cond = PTHREAD_COND_INITIALIZER; - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
pthread_cond_t类型，本质是一个结构体。可简单看为有两种取值0和1，即满足条件和不满足条件
pthread_cond_init初始化一个条件变量，pthread_cond_destroy销毁一个条件变量。
pthread_cond_init的第一个参数cond是条件变量类型的指针，第二个参数attr表示属性，传入NULL表示使用默认属性。成功返回0，失败返回错误码。
pthread_cond_destroy的参数cond是条件变量类型的指针。成功返回0，失败返回错误码。
调用pthread_cond_init是动态初始化，还有一种静态初始化的方式:
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;- 1
🌵pthread_cond_wait
让线程在cond条件变量下等待
pthread_cond_timedwait, pthread_cond_wait - wait on a condition #includeint pthread_cond_timedwait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime); int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex); - 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
pthread_cond_wait阻塞等待一个条件变量，pthread_cond_timedwait限时等待一个条件变量。
pthread_cond_wait第一个参数是要等的条件变量的指针，mutex表示当前线程所处临界区对应的互斥锁。
pthread_cond_timedwait前两个参数和pthread_cond_wait相同，第三个参数表示时间。
- 等待的线程不止一个，所以肯定是有等待队列的，所以条件变量的实现和队列应该相关，猜的。
- 为什么pthread_cond_wait要传入锁参数？
- 条件变量不会无缘无故的满足，所以肯定要一个参数通过某些操作来改变共享变量满足条件，会牵扯到共享数据（临界资源）的变化，所以要加锁。（如果大家都看到同一个条件变量，那条件变量本身就是一个临界资源）
- 在条件变量下等待的线程可能在临界区，所以阻塞自己（等待）的同时释放锁，被唤醒后重新获取锁，这样才能保护临界资源。
所以wait和signal操作是会涉及到锁的,这里也导致了生产者消费者模型里面一个很经典的问题:为什么判断的条件要用while而不能用if?
因为wait会释放锁,此时有多个线程在同一个地方等待,如果使用pthread_cond_broadcast唤醒多个线程,唤醒他们后他们都去竞争锁,只有一个线程拿到锁,这个线程做完了自己的事情后,别的线程都是醒着的,是可以往下执行的,本来不应该醒的此时醒了还处理了数据所以就会出问题,这个问题涉及到的两个概念,一个是惊群现象,一个是伪唤醒
🌵pthread_cond_signal和pthread_cond_broadcast
pthread_cond_broadcast, pthread_cond_signal - broadcast or signal a condition #includeint pthread_cond_broadcast(pthread_cond_t *cond); int pthread_cond_signal(pthread_cond_t *cond); - 1
- 2
- 3
- 4
- 5
- 6
唤醒在cond条件变量下等待的线程。
pthread_cond_signal唤醒一个条件变量为cond的线程，pthread_cond_broadcast唤醒所有条件变量为cond的线程。成功返回0，失败返回错误码。
🔥例子
线程0控制其余几个线程，让他们有序进行。
#include#include #include #include using namespace std; pthread_mutex_t lock; pthread_cond_t cond; void* routine2(void* args)//被控制的线程 { printf("thread %d start\n",*((int*)args)); while(1) { pthread_cond_wait(&cond,&lock); printf("thread %d is running\n",*((int*)args));//要处理的工作数据等 sleep(1); } } void* routine1(void* args)//控制线程 { while(1) { pthread_cond_signal(&cond);//唤醒一个线程，即摇铃铛 printf("i am ctrl thread %d ",*((int*)args)); sleep(1); } } #define NUM 5 int main() { pthread_cond_init(&cond,nullptr); pthread_mutex_init(&lock,nullptr); pthread_t tid[NUM]; for(int i=0;i<NUM;i++) { int* p=new int(i); if(i==0) { pthread_create(tid+i,nullptr,routine1,(void*)p);usleep(30000);//创建完睡一下确保一定跑起来了 } else { pthread_create(tid+i,nullptr,routine2,(void*)p);usleep(30000);//创建完睡一下让线程跑起来 } } for(int i=0;i<NUM;i++) { pthread_join(tid[i],nullptr); } pthread_mutex_destroy(&lock); pthread_cond_destroy(&cond); 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
可以看到四个线程有序的跑起来了。
🥝条件变量使用规范
等待条件代码
pthread_mutex_lock(&mutex); while(条件为假) pthread_cond_wait(cond,&mutex); 修改条件 pthread_mutex_unlock(&mutex);- 1
- 2
- 3
- 4
- 5
给条件发送信号代码
pthread_mutex_lock(&mutex); 设置条件为真 pthread_cond_signal(cond); pthread_mutex_unlock(&mutex);- 1
- 2
- 3
- 4
简单来说就是，条件不满足就去等，另一边设置条件为真，发送一个信号唤醒等待的线程，线程唤醒后执行自己的工作，执行完之后再去修改条件。
🥝小结
条件变量可以看做0和1，只有满足和不满足两种状态，满足就开始跑，不满足就不跑，其中经常需要一个控制者。
唤醒线程就是在摇铃铛，告诉等待队列现在可以来拿”苹果“了，同时排队本身就是有序的，所以可以保证多个线程有序的来拿“苹果”,即有序的访问临界资源。
说了这么多，其实不就是等和唤醒，条件满足就唤醒，不满足就等到条件满足。等待是有队列的，保证了线程有序执行，进而就实现了同步。
🌲信号量
🌴概念
- 信号量是什么?
本质上是一个计数器,用来描述临界资源的数目.
临界资源其实并不是一次只能一个现场访问,其实可以把临界资源分成多份,这样每个线程访问不同份的资源,自然不会发生冲突.此时我们最担心的是当我们访问线程时有别的线程进来,这样造成线程安全问题,由此引出了信号量,信号量则是一种对临界资源的预定机制.
任何一个线程,要去访问临界资源里的某一个,都要先申请信号量,申请到了再去访问临界资源,访问完了再释放信号量.
- 信号量的两种操作
P操作和V操作.
P操作是申请信号量,V操作是释放信号量.
多个线程要申请信号量,说明信号量本身就是一种临界资源,所以信号量在保护临界资源的同时得先保证自己是线程安全的,所以PV操作是原子的.
伪代码:
设sem为信号量
P操作:
lock(); if(sem>0) { sem--; } else { 释放锁并挂起; } unlock()- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
V操作
lock(); sem-- unlock();- 1
- 2
- 3
这只是一段伪代码,信号量在内核里的实现肯定比这个复杂,比如信号量里包含一个等待队列
🌴接口
🌵sem_init
初始化信号量
#includeint sem_init(sem_t *sem, int pshared, unsigned int value); Link with -pthread. - 1
- 2
- 3
- 4
- 5
-
sem:表示信号量
-
pshared:传入0表示线程间共享,传入非0表示进程间共享
-
value是信号量的初始值
-
返回值:成功返回0,失败返回-1
🌵sem_destroy
销毁信号量
#includeint sem_destroy(sem_t *sem); Link with -pthread. - 1
- 2
- 3
- 4
- 5
- sem:信号量
- 返回值:成功返回0,失败返回-1
🌵sem_wait
P操作
#includeint sem_wait(sem_t *sem); Link with -pthread. - 1
- 2
- 3
- 4
- 5
- sem:信号量
- 返回值:申请成功返回0,信号量的值-1,申请失败返回-1,信号量的值不变
🌵sem_post
V操作
#includeint sem_post(sem_t *sem); Link with -pthread. - 1
- 2
- 3
- 4
- 5
- sem信号量
- 返回值:成功返回0,信号量的值+1,失败返回-1,信号量的值不变
🌲生产者消费者模型
🌴概念
从图中可以看出三种关系,两种角色,一个交易场所.
-
三种关系
消费者和消费者:竞争关系,线程上体现为互斥
生产者和生产者:竞争关系,线程上体现为互斥
消费者和生产者:互斥与同步关系,生产者和消费者需要在同一个交易场所交易,当有多个执行流就得竞争锁,所以生产者和消费者之间也存在竞争关系,关于两者的同步关系:一个生产的快,一个消费的慢,就可能造成一个角色的"饥饿",所以两者之间需要同步.
-
两种角色
消费者和生产者,一般是线程或者进程
-
一个交易场所
常用的可以是数组,队列等数据结构
举个例子,学校里有超市,我们要一些日用品是去超市买还是直接去生产地工厂买呢,肯定是超市,此时的超市就是生产场所,而生产者就是工厂.
-
那为什么需要超市呢?
解耦和支持忙闲不均.
解耦:生产环节和消费环节解耦
支持忙闲不均,工厂一次生产一堆,消费者一个一个的消费,生产速度和消费速度不一样时超市就起到了协调的作用
🌴基于阻塞队列的生产者消费者模型
- 队列为空时消费者阻塞,队列为满时生产者阻塞
- 生产者就是往阻塞队列里写数据,消费者就是取数据.借助条件变量控制读写的策略,比如可以是写入一个数据就通知消费者来取,还是说写满了整个阻塞队列等消费者来取.
- 数据满了还往里写就会造成覆盖问题,导致数据的二义性，队列为空还去读 读到的就是旧数据了，也是有问题的,所以数据为满时生产者阻塞,数据空了消费者阻塞
🔥代码
BlockQueue.hpp
#pragma once #include#include #include #include #include using namespace std; template<class T> class BlockQueue { public: BlockQueue(int cap) :_cap(cap) { pthread_mutex_init(&_lock,nullptr); pthread_cond_init(&_have_space,nullptr); pthread_cond_init(&_have_data,nullptr); } bool IsFull() { return _bq.size()==_cap; } bool IsEmpty() { return _bq.size()==0; } void Put(const T&in)//生产者 { //p_lock 多生产时在这加锁 pthread_mutex_lock(&_lock); //满了就不能生产了 while(IsFull())//放在while里防止伪唤醒 //多个线程在这等待,如果用if判断条件和pthread_cond_broadcast唤醒多个线程,多个线程竞争锁,其中一个线程竞争到了锁执行了对应的任务,此时条件变量变了,即别的线程不应该被唤醒,但此时别的线程已经醒了会去往下执行处理数据,所以会出问题.如果用while的话,别的这些已经醒了的线程再走while判断就回到等待状态.说了这么多,记住结论:wait和while连用,特别是线程池的时候注意这块 { pthread_cond_wait(&_have_space,&_lock); } //走到这肯定有空间了 _bq.push(in); //有的话就直接喊人来消费 pthread_cond_signal(&_have_data); pthread_mutex_unlock(&_lock); //p_unlock 多生产时在这解锁 } void Get(T* out) { //c_lock 多消费时在这上锁 pthread_mutex_lock(&_lock); //当队列为空时就不能消费了 while(IsEmpty()) { pthread_cond_wait(&_have_data,&_lock); } //走到这肯定不为空 *out=_bq.front(); _bq.pop(); pthread_cond_signal(&_have_space); pthread_mutex_unlock(&_lock); //c_unlock 多消费时在这解锁 } ~BlockQueue() { pthread_mutex_destroy(&_lock); pthread_cond_destroy(&_have_data); pthread_cond_destroy(&_have_space); } private: int _cap; queue<T>_bq; pthread_mutex_t _lock; pthread_cond_t _have_space; pthread_cond_t _have_data; }; - 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
BlockQueue.cc
#include"BlockQueue.hpp" #includeclass Task { public: int operator()(int num)//仿函数,模拟在对数据进行处理,这里模拟的是1-num的和 { return (1+num)*num/2; } }; void* Producer(void* p) { srand(time(0)); BlockQueue<int>* bq=(BlockQueue<int>*)p; while(1) { int num=rand()%10; bq->Put(num); cout<<"producer: "<<num<<endl; sleep(1);//生产的慢,消费的快,所以现象是生产一个立马会被消费掉 } } void* Consumer(void* c) { BlockQueue<int>* bq=(BlockQueue<int>*)c; int out=0; Task task; while(1) { bq->Get(&out); cout<<"consumer: "<<task(out)<<endl; sleep(1);//消费的慢,生产者直接生产满了等消费者来拿 } } int main() { pthread_t producer; pthread_t consumer; BlockQueue<int>* bq=new BlockQueue<int>(10); pthread_create(&producer,nullptr,Producer,bq); pthread_create(&consumer,nullptr,Consumer,bq); pthread_join( producer,nullptr); pthread_join( consumer,nullptr); delete bq; 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
🥝wait与while连用的原因
原因:多个线程被唤醒抢一把锁,导致其他线程在这把锁外面等,其他线程醒了又去争这把锁,但是数据已经被上一个线程读走了.
void Put(const T&in)//生产者 { pthread_mutex_lock(&_lock); //满了就不能生产了 while(IsFull())//放在while里防止伪唤醒 //多个线程在这等待,如果用if判断条件和pthread_cond_broadcast唤醒多个线程,多个线程竞争锁,其中一个线程竞争到了锁执行了对应的任务,此时条件变量变了,即别的线程不应该被唤醒,但此时别的线程已经醒了会去往下执行处理数据,所以会出问题.如果用while的话,别的这些已经醒了的线程再走while判断就回到等待状态.说了这么多,记住结论:wait和while连用,特别是线程池的时候注意这块 { pthread_cond_wait(&_have_space,&_lock); } //走到这肯定有空间了 _bq.push(in); //有的话就直接喊人来消费 pthread_cond_signal(&_have_data); pthread_mutex_unlock(&_lock); }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
🌿运行效果
生产者每隔一秒生产一个,消费者就消费一个
生产者一直生产,消费者每隔1s消费一个
消费者一直消费,生产者每隔1s生产一个
🥝补充
signal放在unlock前,防止刚释放锁就又被别的线程拿走了去判断条件了,如果事务的处理不够及时就会造成线程安全问题
当Transactional碰到锁，有个大坑，要小心。 - 掘金 (juejin.cn)
这张截图是上面的博客里截下来的,下次自己复习的时候好理解.
🌴基于环形队列的生产者消费者模型
铺垫
采用环形队列+信号量的方式来实现一个生产者消费者模型.
每个格子都相当于一份临界资源,只要访问的的格子不同就不会有线程安全问题,所以这里我们采用信号量的预定机制保证多个线程不会操作同一个格子.
🔥代码
RingQueue.hpp
#pragma once #include#include #include #include #include #include #include using namespace std; template<class T> class RingQueue { public: RingQueue(int cap) :_cap(cap) ,_ring(_cap) ,c_index(0) ,p_index(0) { sem_init(&sem_space,0,_cap); sem_init(&sem_data,0,0); } void Put(const T& in) { //生产者 //放里面放资源时先申请信号量 sem_wait(&sem_space); //lock 多生产时在这加锁,此时别的线程进来也可以去申请信号量,而不用进到锁里面再申请信号量 _ring[p_index]=in; sem_post(&sem_data);//减少临界区的代码 p_index++; p_index%=_cap; //unlock } void Get(T* out) { //消费者 sem_wait(&sem_data); //lock *out=_ring[c_index]; sem_post(&sem_space); c_index++; c_index%=_cap; //unlock } ~RingQueue() { sem_destroy(&sem_space); sem_destroy(&sem_data); } private: int _cap; vector<T> _ring; sem_t sem_space; sem_t sem_data; size_t c_index; size_t p_index; }; - 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
main.cc
#include "RingQueue.hpp" class Task { public: int operator()(int num) { int ans=0; for(int i=1;i<=num;i++) { ans+=i; } return ans; } }; void* Producer(void* p) { RingQueue<int>* rq=(RingQueue<int>*)p; while(1) { int num=rand()%10; rq->Put(num); cout<<"producer: "<<num<<endl; //sleep(1);//生产的慢,消费的快,现象应该是有一个就会被直接消费掉 } } void* Consumer(void* c) { RingQueue<int>* rq=(RingQueue<int>*)c; int out=0; Task t; while(1) { rq->Get(&out); int ans=t(out); cout<<"consumer: "<<ans<<endl; sleep(1);//生产的快,消费的慢,应该是一开始生产满了,后面消费一个就补充一个,队列一直是满的 } } int main() { pthread_t producer; pthread_t consumer; RingQueue<int> rq(10); pthread_create(&producer,nullptr,Producer,&rq); pthread_create(&consumer,nullptr,Consumer,&rq); pthread_join(producer,nullptr); pthread_join(consumer,nullptr); 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
🌿运行效果
消费者每隔1s消费一个,生产者一直在生产,现象应该是生产者直接把队列塞满了,然后消费者拿完一个生产者就补充一个.
生产者每隔1s生产一个,消费者一直在消费
生产者每隔1s生产,消费者每隔1s消费.
🥝补充
这里只是单生产单消费,可以注意到这里与阻塞队列不同的点在于这里没手动加锁.
当然多生产多消费的时候肯定要加锁,此外要注意加锁的位置,比如锁是加在申请信号量的前面还是后面,结论是加在后面,那多个线程可以在进入锁之前就申请到信号量,相比于进入锁才能拿到信号量肯定是前者效率更高.
🌲线程池
🌴概念
- 池化技术
【项目】实现一个mini的tcmalloc（高并发内存池）,里面讲了池化技术
池化技术可以避免频繁申请带来的开销,可以提高性能,套到线程池这里,那就是创建一个线程池,池子里有很多线程,一旦有任务要执行,不必申请去创建线程,而是直接从池子里拿出线程去完成对应的任务,这个过程就节省了创建线程的开销.
- 线程池
一次预先创建一大批线程,让这些线程处于待机状态,一旦有数据或者任务就可以直接交给线程处理
🔥代码实现
实现一个简单的线程池
这个线程池里有num个线程,一旦有任务就立马从池子里拿出线程去处理.我这里模拟的任务为数字的加减乘除.
ThreadPool.hpp
#pragma once #include#include #include #include #include #include #include "Task.hpp" using namespace std; template<class T> class ThreadPool { public: ThreadPool() { pthread_cond_init(&_cond,nullptr); pthread_mutex_init(&_lock,nullptr); } void Lock() { pthread_mutex_lock(&_lock); } void UnLock() { pthread_mutex_unlock(&_lock); } void Wait() { pthread_cond_wait(&_cond,&_lock); } void WakeUp() { pthread_cond_signal(&_cond); } bool IsEmpty() { return _tq.size()==0; } void PopTask(T* out) { *out=_tq.front(); _tq.pop(); } void PushTask(const T& t) { Lock(); _tq.push(t); WakeUp(); UnLock(); cout<<"加入任务队列,当前任务数:"<<_tq.size()<<","; } size_t Tqsize() { return _tq.size(); } static void* Routine(void* args)//非static函数会有隐含的参数this指针,所以用static { pthread_detach(pthread_self()); //因为static函数只能调用static方法或数据,但是我们需要访问lock这些所以需要this指针,此外lock相关操作也要进行封装 ThreadPool* threadPool=(ThreadPool*)args; //看队列里有没有待处理的任务,没有的话就进行等待,有的话直接给处理了 //_tq表示任务队列,由于不是线程安全的,得加锁 while(true) { //这里必须一直在跑,不然创建的线程只会处理一次任务 threadPool->Lock(); while(threadPool->IsEmpty())//这里不能用if进行判断 { cout<<"队列为空开始等待"<<endl; threadPool->Wait(); } //走到这说明任务队列不为空 T t; cout<<"取出任务"<<threadPool->Tqsize(); threadPool->PopTask(&t);//把任务取出来 threadPool->UnLock(); cout<<",开始处理任务..."<<endl; t();//不用在锁里面处理了,相当于线程的局部变量,被一个线程独有 } } void ThreadPoolInit(int num)//这里不是构造函数,别忘了手动启动,有人忘了调试了半天是谁我不说,是我啊那没事了 { for(int i=0;i<num;i++) { pthread_t tid; pthread_create(&tid,nullptr,Routine,(void*)this); } } ~ThreadPool() { pthread_cond_destroy(&_cond); pthread_mutex_destroy(&_lock); } private: pthread_cond_t _cond; pthread_mutex_t _lock; queue<T> _tq; }; - 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
Task.hpp
#pragma once #includeusing namespace std; class Task { public: Task() {} Task(int _a,int _b,char _op) :a(_a) ,b(_b) ,op(_op) {} ~Task() {} void operator()()//处理对象的数据,不要传参 { int ans=0; switch(op) { case '+': ans=a+b; break; case '-': ans=a-b; break; case '*': ans=a*b; break; case '/': if(b==0) { cout<<"div 0"<<endl; } else { ans=a/b; } break; case '%': if(b==0) { cout<<"mod 0"<<endl; } else { ans=a%b; } break; default: cout<<"unknown op!"<<endl; break; } cout<<"thread["<<pthread_self()<<"]"<<": "<<a<<" "<<op<<" "<<b<<"="<<ans<<endl; } private: int a; int b; char op; }; - 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
main.cc
#include"ThreadPool.hpp" #include"Task.hpp" int main() { srand(time(nullptr)); ThreadPool<Task>* threadPool=new ThreadPool<Task>(); threadPool->ThreadPoolInit(10); sleep(1); string ops="+-*/%"; while(true) { int x=rand()%50+1; int y=rand()%50+1; char op=ops[rand()%5]; Task t(x,y,op); cout<<"成功创建任务,"; threadPool->PushTask(t); sleep(1); } 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
运行效果
🌲多线程优缺点和线程的优缺点
这都是些概念了
- 多线程优点
线程间通信,成本特别低
- 多线程缺点
存在大量的临界资源,所以肯定需要各种互斥和同步机制来保护临界资源.
-
线程优点
- 创建线程相比创建进程的代价小
- 线程切换代价小,占用资源小,都是相当于进程来说
- 充分利用多处理器
- 等待慢速IO结束时持续可以处理其他任务,此外通过一些操作使得IO操作重叠让线程可以同时等待不同的IO操作
-
线程缺点
一个线程崩了,进程就崩了,所以我们也说线程降低了程序的健壮性
🌲补充
🌴自旋锁(spinlock)
当我们使用互斥锁时,比如线程A拿到了锁,线程B去尝试加锁就会失败,内核把这个线程有就绪转为睡眠,直到合适的时候再把锁给他,即发生了上下文切换,发生上下文切换就会有开销,如果此时代码执行1ns,线程切换4ns,那显然得不偿失.
所以为什么我们不让这个线程一直去等呢,这就是自旋锁了,自旋锁加锁失败会一直等,直到成功.所以我们如果直到锁里面的代码运行时间很短,可以采用自旋锁来提高性能.
在用户看来线程没有申请到互斥锁和自旋锁时,都像"卡住了"一样
自旋锁的实现思路,用while循环一直去检测上锁的状态,只要能加锁立即就加锁,上面那篇博客里也说了可以用CPU指令的,指令方面我就不太清楚了.
🌴读写者问题
- 读写者问题也有三个关系,两个角色,一个交易场所,有点像黑板报.
场景常见于读者很多,写者很少.
三个关系
角色 关系 理由 读者与读者 没有关系 多个人可以同时去读,不会竞争,因为不会取走数据 写者与写者 互斥 写数据时有竞争关系,比如竞争锁之类的 读者与写者 互斥与同步 竞争关系,如果一边写一边读就有同步关系 两个角色:写者和读者
一个交易场所:一段缓冲区
相关的接口有pthread_rwlock_rdlock,pthread_rwlock_wrlock,pthread_rwlock_unlock
- 伪代码
//简化版,利于理解,不涉及等待队列等 int reader_cnt=0; pthread_mutex_t lock; //---------读者的锁--------- lock(); ++reader_cnt; unlock(); 访问临界资源;//读者与读者之间没有关系,所以访问临界资源不必加锁 lock(); --reader_cnt; unlock(); //---------写者的锁--------- start: lock(); if(reader_cnt>0)//有读者时保证写者不在访问临界资源 { unlock(); goto start; } 访问临界资源;//写时必定没有读者,且写的时候肯定在锁的保护下 unlock();- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
为了避免读者饥饿或者写者饥饿问题,有对应的策略解决,如读者优先和写者优先等.这里不深入探究
🌲总结
- 线程,轻量级进程,linux系统实现线程的方式和实现进程差不多(加几个task_struct)
- 线程的创建,销毁,等待,分离,取消,同步等,没有线程替换
- 条件变量,主要就是两个函数的使用,一个是pthread_cond_wait,一个是pthread_cond_signal,使用时记得pthread_cond_wait和while连用,条件满足开始pthread_cond_wait使得线程陷入等待,条件改变后用pthread_cond_signal函数唤醒线程,本质上是为了线程同步,可以做到线程有序的运行,实例就是线程池和生产者消费者模型
- 信号量本质就是一个线程安全的计数器,描述了临界资源的数目,是一种预定机制,信号量的操作主要是PV操作,P操作申请信号量,V操作释放信号量
- 生产者消费者模型,理清楚三种关系,两个角色和一个交易场所,具体实现可以阻塞队列+条件变量或者环形队列+信号量
- 自旋锁和互斥锁的区别
- 线程池可以及时的处理任务,且省去了频繁创建线程的开销,如果不想创建多个线程池,可以结合单例模式,结合单例模式里的懒汉模式记得双检查.
Maven的基础知识
CLIP与DINOv2的图像相似度对比
中南林业科技大学Java实验报告十二：数据库系统设计 - 从0到1搭建java可视化学生管理系统源代码
vue2文件下载和合计表格
pytorch的searchsorted解释
2023第十二届中国智能产业高峰论坛之文档大模型的探索与思考
一个讲座监控软件
整站抓取的神器
机器学习（3）