线程的互斥与同步
线程的互斥
相关概念：
临界资源：多线程执行流共享的资源就叫做临界资源
临界区：每个线程内部，访问临界资源的代码，就叫做临界区
互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用
原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么未完成
简单的抢票程序
//未保护临界区#include<stdio.h>#include<unistd.h>#include<pthread.h>#include<sys/stat.h>#include<sys/types.h>int tickets=1000;void*TicketGrabbing(void* arg){constchar* str=(char*) arg;while(1){if(tickets>0){usleep(100);printf("%s:get a ticket:%d\n",str,tickets--);}else{break;}}printf("%s quit!\n",str);pthread_exit((void*)0);}intmain(){
pthread_t t1,t2,t3,t4;pthread_create(&t1,NULL,TicketGrabbing,"thread 1");pthread_create(&t2,NULL,TicketGrabbing,"thread 2");pthread_create(&t3,NULL,TicketGrabbing,"thread 3");pthread_create(&t4,NULL,TicketGrabbing,"thread 4");pthread_join(t1,NULL);pthread_join(t2,NULL);pthread_join(t3,NULL);pthread_join(t4,NULL);return0;}
可以看到，票数竟然被抢到了负数，那这是为什么呢？
a++为什么不是原子的：
这就是对临界资源没有进行保护而造成了问题
互斥量
为解决上面的问题，本质上就是需要一把锁。Linux上提供的这把锁叫互斥量
初始化互斥量
方法1，静态分配:
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER
方法2，动态分配:
int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);
参数：
mutex：要初始化的互斥量
attr：初始属性，NULL默认
销毁互斥量
int pthread_mutex_destroy(pthread_mutex_t *mutex)；
注意：
1.使用PTHREAD_ MUTEX_ INITIALIZER初始化的互斥量不需要销毁
2.不要销毁一个已经加锁的互斥量
3.已经销毁的互斥量，要确保后面不会有线程再尝试加锁
互斥量加锁和解锁
加锁是有损于性能的
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
返回值:成功返回0,失败返回错误号
对于lock函数：其他线程已经锁定互斥量，或者存在其他线程同时申请互斥量，但没有竞争到互斥量，那么pthread_ lock调用会陷入阻塞(执行流被挂起)，等待互斥量解锁，这样保证临界区只有一个线程访问，保护了临界区
通过互斥量实现线程安全版本的抢票：
#include<stdio.h>#include<unistd.h>#include<pthread.h>#include<sys/stat.h>#include<sys/types.h>int tickets=1000;
pthread_mutex_t mutex;void*TicketGrabbing(void* arg){constchar* str=(char*) arg;while(1){pthread_mutex_lock(&mutex);if(tickets>0){usleep(100);printf("%s:get a ticket:%d\n",str,tickets--);pthread_mutex_unlock(&mutex);}else{pthread_mutex_unlock(&mutex);break;}}printf("%s quit!\n",str);pthread_exit((void*)0);}intmain(){pthread_mutex_init(&mutex,NULL);
pthread_t t1,t2,t3,t4;pthread_create(&t1,NULL,TicketGrabbing,"thread 1");pthread_create(&t2,NULL,TicketGrabbing,"thread 2");pthread_create(&t3,NULL,TicketGrabbing,"thread 3");pthread_create(&t4,NULL,TicketGrabbing,"thread 4");pthread_join(t1,NULL);pthread_join(t2,NULL);pthread_join(t3,NULL);pthread_join(t4,NULL);pthread_mutex_destroy(&mutex);return0;}
互斥量实现原理
为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换，由于只有一条指令,保证了原子性
即使是多处理器平台,访问内存的 总线周期也有先后,一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期
加锁解锁是如何完成的：
所以本质上是1只有一份，只有拿到那个1的线程才能return执行临界区的代码
可重入VS线程安全
概念
线程安全：多个线程并发同一段代码时，不会出现不同的结果。常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题
重入：同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数
线程安全与重入的关系：
常见的线程不安全的情况
1.不保护共享变量的函数
2.函数状态随着被调用，状态发生变化的函数
3.返回指向静态变量指针的函数
4.调用线程不安全函数的函数
常见的线程安全的情况
1.每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的
2.类或者接口对于线程来说都是原子操作
3.多个线程之间的切换不会导致该接口的执行结果存在二义性
常见不可重入的情况
1.调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
2.调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
3.可重入函数体内使用了静态的数据结构
常见可重入的情况
1.不使用全局变量或静态变量
2.不使用用malloc或者new开辟出的空间
3.不调用不可重入函数
4.不返回静态或全局数据，所有数据都有函数的调用者提供
5.使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据
可重入与线程安全联系
1.函数是可重入的，那就是线程安全的
2.函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
3.如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的
可重入与线程安全区别
1.可重入函数是线程安全函数的一种
2.线程安全不一定是可重入的，而可重入函数则一定是线程安全的
3.如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生死锁，因此是不可重入的
死锁
死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态
死锁四个必要条件
①互斥条件：一个资源每次只能被一个执行流使用
②请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
③不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
④循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系
避免死锁
1.破坏死锁的四个必要条件
2.加锁顺序一致
3.避免锁未释放的场景
4.资源一次性分配
进程同步
同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步
为什么要有进程同步？
单纯的互斥会让竞争力强的线程获得锁，竞争力弱的一直等待，效率十分低
条件变量
进程同步通过条件变量来实现，类似于进程互斥的互斥量
条件变量：用来描述某种临界资源是否就绪的一种数据化描述
条件变量通常要与互斥搭配使用
条件变量初始化
int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
参数：
cond：要初始化的条件变量
attr：NULL
销毁
int pthread_cond_destroy(pthread_cond_t *cond)
等待条件满足
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
参数：
cond：要在这个条件变量上等待
mutex：互斥量
线程挂起等待时还拿着互斥锁，别的线程无法进入临界区，这时需要释放锁mutex
唤醒等待
int pthread_cond_broadcast(pthread_cond_t *cond);//唤醒所有
int pthread_cond_signal(pthread_cond_t *cond);//唤醒一个
示例：
#include<iostream>#include<pthread.h>#include<cstdio>usingnamespace std;
pthread_mutex_t mutex;
pthread_cond_t cond;void*run(void* arg){pthread_detach(pthread_self());
cout<<(char*)arg<<"run....."<<endl;while(true){pthread_cond_wait(&cond,&mutex);//初始条件不满足，等待条件满足
cout<<"thread"<<pthread_self()<<"runing.."<<endl;}}intmain(){pthread_mutex_init(&mutex,NULL);pthread_cond_init(&cond,NULL);
pthread_t t1,t2,t3;pthread_create(&t1,NULL,run,(void*)"thread t1");pthread_create(&t2,NULL,run,(void*)"thread t2");pthread_create(&t3,NULL,run,(void*)"thread t3");while(true){getchar();pthread_cond_signal(&cond);}pthread_mutex_destroy(&mutex);pthread_cond_destroy(&cond);return0;}
可以看到使用同步后，打印结果出现了相当强的顺序性
因为每个线程等待时都在资源的等待队列中，唤醒队头后再等待插入到队尾，这样就出现每个线程顺序打印
资源等待队列
生产者消费者模型
本质上是一个线程安全的队列，和两种角色的线程（生产者和消费者）
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的
三种关系：
生产者与生产者（竞争关系，互斥关系）
消费者和消费者（竞争关系，互斥关系）
生产者和消费者（互斥关系（保证数据的正确性），同步关系（多线程协同））
两种角色：
生产者和消费者
一个交易场所：
通常是内存中的缓冲区，自己通过某种方式实现的
生产者消费者模型优点：
1.解耦
2.支持并发
3.支持忙闲不均
为什么要有生产者与消费者模型？
通过缓冲区，实现了生产者与消费者的解耦
单生产者，单消费者，一个blockingqueue
代码实现：
blockingqueue.c
#pragmaonce#include<iostream>usingnamespace std;#include<pthread.h>#include<queue>#include<cstdlib>#include<ctime>#defineNUM10template<classT>classblockingqueue{boolisFull(){return q.size()==cap;}boolisEmpty(){return q.size()==0;}public:blockingqueue(int _cap=NUM):cap(_cap){pthread_mutex_init(&lock,NULL);pthread_cond_init(&full,NULL);pthread_cond_init(&empty,NULL);}voidPush(T& in){pthread_mutex_lock(&lock);while(isFull()){//wait是函数可能失败，造成伪唤醒，//所以通过while循环再次进行判定//队列为空不能push等待消费者Poppthread_cond_wait(&full,&lock);}
q.push(in);pthread_mutex_unlock(&lock);if(q.size()>=cap/2){
cout<<"数据太多了，消费者来消费吧"<<endl;pthread_cond_signal(&empty);}}voidPop(T& out){pthread_mutex_lock(&lock);while(isEmpty()){pthread_cond_wait(&empty,&lock);}
out=q.front();
q.pop();pthread_mutex_unlock(&lock);if(q.size()<=cap/2){
cout<<"数据已经快不够了，生产者来生产吧"<<endl;pthread_cond_signal(&full);}}~blockingqueue(){pthread_mutex_destroy(&lock);pthread_cond_destroy(&full);pthread_cond_destroy(&empty);}public:
queue<T> q;private:int cap;
pthread_mutex_t lock;
pthread_cond_t full;
pthread_cond_t empty;};
main函数
#include"blockingqueue.c"#include<unistd.h>void*Consumer(void* arg){auto bq=(blockingqueue<int>*) arg;while(true){int data=0;//输出型参数
bq->Pop(data);
cout<<"consumer:"<<data<<"当前个数:"<<bq->blockingqueue::q.size()<<endl;sleep(3);}}void*Producer(void* arg){auto bq=(blockingqueue<int>*) arg;while(true){int data=rand()%100+1;
bq->Push(data);
cout<<"Producer:"<<data<<"当前个数:"<<bq->blockingqueue::q.size()<<endl;sleep(1);}}intmain(){srand((unsignedlong)time(NULL));
blockingqueue<int>* bq=new blockingqueue<int>();
pthread_t c,p;pthread_create(&c,NULL,Consumer,bq);pthread_create(&p,NULL,Producer,bq);pthread_join(c,NULL);pthread_join(p,NULL);return0;}
POSIX信号量
信号量本身是一个计数器，描述临界资源的计数器
拥有更细粒度的临界资源的管理
POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步和互斥
P操作：
①信号量S减1；
②若S减1后仍大于或等于0，则进程继续执行；
③若S减1后小于0，则该进程被阻塞后放入等待该信号量的等待队列中
V操作：
①S加1；
②若相加后结果大于0，则进程继续执行；
③若相加后结果小于或等于0，则从该信号的等待队列中释放一个等待进程，然后再返回原进程继续执行或转进程调度
初始化信号量
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值，大小表示可以同时访问临界资源的线程数
销毁信号量
int sem_destroy(sem_t *sem);
等待信号量
功能：等待信号量，会将信号量的值减1
int sem_wait(sem_t *sem); //P()
发布信号量
功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1
int sem_post(sem_t *sem);//V()
信号量版本实现线程安全的抢票程序
#include<iostream>#include<string>#include<pthread.h>#include<unistd.h>#include<semaphore.h>usingnamespace std;classSem{private:
sem_t sem;public:Sem(int num){sem_init(&sem,0,num);}voidP(){sem_wait(&sem);}voidV(){sem_post(&sem);}~Sem(){sem_destroy(&sem);}};
Sem sem(1);//只能有一个人在抢int tickets=100;void*Gettickets(void* arg){
string s=(char*)arg;while(true){sleep(1);
sem.P();if(tickets>0){usleep(10000);
cout<<s<<"get tickets:"<<tickets--<<endl;
sem.V();}else{
sem.V();break;}}pthread_exit((void*)0);}intmain(){
pthread_t t1,t2,t3;pthread_create(&t1,NULL,Gettickets,(void*)"thread t1");pthread_create(&t2,NULL,Gettickets,(void*)"thread t2");pthread_create(&t3,NULL,Gettickets,(void*)"thread t3");pthread_join(t1,NULL);pthread_join(t2,NULL);pthread_join(t3,NULL);return0;}
基于环形队列的生产消费模型
通过信号量实现：
#include<iostream>#include<unistd.h>#include<pthread.h>#include<stdlib.h>#include<semaphore.h>#include<vector>usingnamespace std;#defineNUM5template<classT>classring{private:
sem_t sem_blank;
sem_t sem_data;
vector<T> v;int _cap;int c_pos;int p_pos;voidP(sem_t& t){sem_wait(&t);}voidV(sem_t& t){sem_post(&t);}public:ring(int num=NUM):_cap(num),c_pos(0),p_pos(0){
v.resize(_cap);sem_init(&sem_data,0,0);sem_init(&sem_blank,0,_cap);}voidPush(T& in){P(sem_blank);
v[p_pos]=in;V(sem_data);
p_pos++;
p_pos%=_cap;}voidPop(T& out){P(sem_data);
out=v[c_pos];V(sem_blank);
c_pos++;
c_pos%=_cap;}~ring(){sem_destroy(&sem_data);sem_destroy(&sem_blank);}};void*produce(void* arg){
ring<int>* rq=(ring<int>*) arg;while(true){sleep(1);int a=rand()%100+1;
rq->Push(a);
cout<<"produce done:"<<a<<endl;}}void*consume(void* arg){
ring<int>* rq=(ring<int>*) arg;while(true){sleep(2);int a=0;
rq->Pop(a);
cout<<"consume done:"<<a<<endl;}}intmain(){
pthread_t p,c;
ring<int>* rp=new ring<int>();srand((unsignedlong)time(NULL));pthread_create(&p,NULL,produce,rp);pthread_create(&c,NULL,consume,rp);pthread_join(p,NULL);pthread_join(c,NULL);return0;}
该队列不可能出现数据不一致的情况
只有两种情况生产者和消费者指向同一空间，访问同一临界资源
1.队列为空时，消费者不能消费，只有生产者生成，不存在竞争
2.队列为满时，生产者不能生产，等待消费者消费，也不存在竞争
线程池
一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
- 线程池的应用场景：
- 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。 *
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.
Task.hpp
#pragmaonce #include<iostream>usingnamespace std;#include<pthread.h>classTask{private:int _x;int _y;char _op;public:Task(){}Task(int x,int y,char op):_x(x),_y(y),_op(op){}voidrun(){int z=0;switch(_op){case'+':
z=_x+_y;break;case'-':
z=_x-_y;break;case'*':
z=_x*_y;case'/':if(_y==0){
cerr<<"div zero"<<endl;}else
z=_x/_y;break;case'%':if(_y==0){
cerr<<"mod zero"<<endl;}else
z=_x%_y;break;default:
cerr<<"default"<<endl;}
cout<<_x<<_op<<_y<<"="<<z<<endl;}};
Threadpool.hpp
#include<iostream>usingnamespace std;#include<pthread.h>#include<queue>#defineNUM5template<classT>classThreadPool{private:
queue<T> task_queue;int thread_num;
pthread_mutex_t lock;
pthread_cond_t cond;public:ThreadPool(int num=NUM):thread_num(num){pthread_mutex_init(&lock,NULL);pthread_cond_init(&cond,NULL);}voidLockQueue(){pthread_mutex_lock(&lock);}voidUnlockQueue(){pthread_mutex_unlock(&lock);}voidwaitcond(){pthread_cond_wait(&cond,&lock);}voidwakeup(){pthread_cond_signal(&cond);}boolisEmptyThreadPool(){return task_queue.size()==0;}staticvoid*Routine(void* arg){pthread_detach(pthread_self());
ThreadPool* This=(ThreadPool*) arg;while(true){
This->LockQueue();if(This->isEmptyThreadPool()){
This->waitcond();}
T t;
This->Pop(t);
This->UnlockQueue();
cout<<"thread:"<<pthread_self()<<"run:";
t.run();}}voidPush(T& in){LockQueue();
task_queue.push(in);UnlockQueue();wakeup();}voidPop(T& out){
out=task_queue.front();
task_queue.pop();}voidInitThreadPool(){
pthread_t tid;for(int i=0;i<thread_num;i++){pthread_create(&tid,NULL,Routine,this);}}~ThreadPool(){pthread_cond_destroy(&cond);pthread_mutex_destroy(&lock);}};
对于Routine函数
main函数
#include"Task.hpp"#include"Threadpool.hpp"#include<unistd.h>intmain(){
ThreadPool<Task>* tp=new ThreadPool<Task>();
tp->InitThreadPool();srand((unsignedint)time(NULL));constchar* op="+-*/%";while(true){sleep(1);int x=rand()%100+1;int y=rand()%100+1;
Task t(x,y,op[x%5]);
tp->Push(t);}return0;}
线程安全的单例模式
某些类, 只应该具有一个对象(实例), 就称之为单例
单例模式通常分为饿汉模式和懒汉模式
吃完饭, 立刻洗碗, 这种就是饿汉方式. 因为下一顿吃的时候可以立刻拿着碗就能吃饭.
吃完饭, 先把碗放下, 然后下一顿饭用到这个碗了再洗碗, 就是懒汉方式.
懒汉方式最核心的思想是 “延时加载”，“需要时再加载” 从而能够优化服务器的启动速度以及有效利用内存
饿汉方式实现单例模式
template<typenameT>classSingleton{static T data;public:static T*GetInstance(){return&data;}};
懒汉方式实现单例模式(线程安全版本)
template<typenameT>classSingleton{volatilestatic T* inst;// 需要设置 volatile 关键字, 否则可能被编译器优化. static std::mutex lock;public:static T*GetInstance(){if(inst ==NULL){// 双重判定空指针, 降低锁冲突的概率, 提高性能.
lock.lock();// 使用互斥锁, 保证多线程情况下也只调用一次 new. if(inst ==NULL){
inst =newT();}
lock.unlock();}return inst;}};
注意事项:
- 加锁解锁的位置
- 双重 if 判定, 避免不必要的锁竞争
- volatile关键字防止过度优化
STL,智能指针线程安全问题
STL中的容器是否是线程安全的
不是
原因是, STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响.
而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶).
因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全.
智能指针是否是线程安全的?
是
对于 unique_ptr:
由于只是在当前代码块范围内生效, 因此不涉及线程安全问题
对于 shared_ptr,:
多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数.
其他常见的各种锁
悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
自旋锁:
读写者问题
初始化
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,const pthread_rwlockattr_t *restrict attr);
销毁
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
加锁和解锁
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
多线程下：
如果读写写者同时到来，如果有人还在读，则不让读者进入临界区，让写者进入并写完后再读，称之为写者优先
反之，为读者优先
示例：
版权归原作者 TZC⑥ 所有， 如有侵权，请联系我们删除。