【Linux】线程同步和互斥
目录
一、线程互斥
1.相关概念
1.临界资源：多线程执行流共享的资源，且一次只能允许一个执行流访问的资源就叫做临界资源。(多线程、多进程打印数据)
2.临界区：每个线程内部，访问临界资源的代码，就叫做临界区。
3.互斥：任何时刻，互斥保证有且只有一个执行流进入临界区，访问临界资源，通常对临界资源起保护作用。
4.原子性：不会被任何调度机制打断的操作，该操作只有两态，要么完成，要么不执行 。
实现一个小实例
#include<iostream> #include<unistd.h> #include<stdio.h> using namespace std; int ticket=10000; void* threadRoutinue(void* args) { const char* name=static_cast<const char*>(args); while(true) { if(ticket>0) { usleep(1000);//模拟抢票花费时间 cout<<name<<"get a ticket: "<<ticket--<<endl; } else{ break; } } return nullptr; } int main() { //创建线程模拟抢票 pthread_t tid[4]; int n=sizeof(tid)/sizeof(tid[0]); for(int i=0;i<n;i++) { char buffer[64]; snprintf(buffer,sizeof(buffer),"thread_%d",i); pthread_create(tid+i,nullptr,threadRoutinue,buffer); } for(int i=0;i<n;i++) { pthread_join(tid[i],nullptr); } return 0; }
从程序中可以看到，票数到0的时候就没有票了，线程就应该退出了。
但是结果中，票数甚至被抢到了负数，这是怎么回事。
这里提一个问题，这里对票(临界资源)的访问是原子的吗？(是安全的吗？) 答案肯定不是！！
可能在一个线程A中，刚刚将tickets加载到内存上，线程A就被切走了，这时线程A的数据和上下文被保存，线程A从CPU上被剥离。
线程B开始抢票，如果他的竞争力非常强，一次运行后抢到了1000张票。
线程B执行完后线程A又来了，他会从上次执行的地方继续执行，但是他上次保存的tickets的数据是10000，所以抢到了一张票后，将剩余的9999张票写回内存，本来线程B执行完后还剩9000张票，但是线程A执行完后剩余的票数反而增多了。
2.互斥锁（mutex）
对于上面的抢票程序，要想使每个线程正确的抢票就要保证：当一个线程在进入到抢票环节时，其他线程不能进行抢票。
所以就可以对抢票环节加互斥锁。
pthread_mutex_init、pthread_mutex_destroy：对线程锁进行初始化和销毁
#include <pthread.h> // pthread_mutex_t mutex: 锁变量，所有线程都可看到 int pthread_mutex_destroy(pthread_mutex_t *mutex);// 销毁锁 int pthread_mutex_init(pthread_mutex_t *restrict mutex,constpthread_mutexattr_t *restrict attr);// 初始化锁 // attr: 锁属性，我们传入空指针就可 // 如果将锁定义为静态或者全局的，可以使用宏直接初始化，且不用销毁 pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;pthread_mutex_lock、int pthread_mutex_unlock：对线程进行加锁和解锁
#include <pthread.h> int pthread_mutex_lock(pthread_mutex_t *mutex); int pthread_mutex_unlock(pthread_mutex_t *mutex);
对抢票小demo进行加锁
#include<iostream> #include<unistd.h> #include<stdio.h> using namespace std; int ticket=10000; //临界资源 pthread_mutex_t mutex; void* threadRoutinue(void* args) { const char* name=static_cast<const char*>(args); while(true) { pthread_mutex_lock(&mutex); if(ticket>0) { usleep(1000);//模拟抢票花费时间 cout<<name<<" get a ticket: "<<ticket--<<endl; pthread_mutex_unlock(&mutex); } else{ cout<<name<<"票抢完了"<<endl; pthread_mutex_unlock(&mutex); break; } usleep(1000); } return nullptr; } int main() { pthread_mutex_init(&mutex,nullptr); //创建线程模拟抢票 pthread_t tid[4]; int n=sizeof(tid)/sizeof(tid[0]); for(int i=0;i<n;i++) { char buffer[64]; snprintf(buffer,sizeof(buffer),"thread_%d",i); pthread_create(tid+i,nullptr,threadRoutinue,buffer); } for(int i=0;i<n;i++) { pthread_join(tid[i],nullptr); } pthread_mutex_destroy(&mutex); return 0; }多线程临界资源原子
细节：
1.凡是访问同一个临界资源的线程，都要进行加锁保护，而且必须加同一把锁，这是一个规则，不能有例外
2.每一个线程访问临界资源之前，得加锁，加锁本质是给 临界区加锁，加锁的粒度尽量细一些。
3.线程访问临界区的时候，需要先加锁 -> 所以线程都必须看到同一把锁 -> 锁本身就是公共资源 -> 锁如何保证自己的安全？ -> 加锁和解锁本身就是原子的。
4.临界区可以是一行代码，可以是一批代码，a.线程可能被切换？ 当然可能 b.切换会有影响嘛？ 没有，因为一个线程申请一个锁以后，该线程被临时切换，其他任何线程没有办法进入临界区，无法申请到锁，所以无法访问到临界资源。
5.这也正是体现互斥带来的串行化的表现，站在其他线程的角度，对其他线程有意义的状态是：锁被申请（持有锁），锁被释放（不持有锁），原子性。
3.互斥锁的原理
以抢票程序为例，当线程需要访问临界资源时，需要先访问mtx，为了所有的线程都能看到它，所以锁肯定是全局的。
且锁本身也是临界资源。那么如何保证锁本身是安全的，即获取锁的过程是安全的。
其原理是：加锁(lock)、解锁(unlock)的过程是原子的！
那怎样才算是原子的呢：一行代码被翻译成汇编后只有一条汇编，就是原子的。
为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令。
该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性。
即使是多处理器平台,访问内存的 总线周期也有先后,一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。
当线程申请到锁之后，进入到临界区访问临界资源，这时线程也可能被切走，被切走后会保护上下文，而锁数据也在上下文中。
所以锁也被带走了，所以即便是该线程被挂起了，其他线程也不能申请到锁，也不能进入临界区。
必须等待拥有锁的线程释放锁之后才能申请到锁。
4.自定义封装一个锁
#pragma once
#include<iostream>
#include<pthread.h>
//封装锁
class _Mutex
{
public:
_Mutex(pthread_mutex_t* mutex):_mutex(mutex)
{}
void lock()
{
pthread_mutex_lock(_mutex);
}
void unlock()
{
pthread_mutex_unlock(_mutex);
}
private:
pthread_mutex_t* _mutex;
};
class lockGuard
{
public:
lockGuard(pthread_mutex_t* mutex):_mutex(mutex)
{
_mutex.lock();
}
~lockGuard()
{
_mutex.unlock();
}
private:
_Mutex _mutex;
};
我们可以使用我们自己封装的锁解决抢票问题
二、可重入和线程安全
线程安全： 线程安全指的是在多线程编程中，多个线程对临界资源进行争抢访问而不会造成数据二义或程序逻辑混乱的情况。常见对全局变量或者静态变量进行操作，并且没有锁保护的情况下，会出现该问题。
重入： 同一个函数被不同的执行流调用，当前一个流程还没有执行完，就有其他的执行流再次进入，我们称之为重入。一个函数在重入的情况下，运行结果不会出现任何不同或者任何问题，则该函数被称为可重入函数，否则，是不可重入函数
线程安全的实现，通过同步与互斥实现
具体互斥的实现可以通过互斥锁和信号量实现、而同步可以通过条件变量与信号量实现。
常见的线程不安全的情况：
不保护共享变量的函数
函数状态随着被调用，状态发生变化的函数
返回指向静态变量指针的函数
调用线程不安全函数的函数
常见不可重入的情况：
调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
可重入函数体内使用了静态的数据结构
可重入与线程安全联系:
函数是可重入的，那就是线程安全的
函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。
可重入与线程安全区别:
可重入函数是线程安全函数的一种
线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数锁还未释放则会产生死锁
三、死锁
死锁概念
死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所占用不会释放的资 源而处于的一种永久等待状态。
死锁四个必要条件
互斥条件：一个资源每次只能被一个执行流使用
请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系
如何避免死锁
核心思想：破坏死锁的4个必要条件中任意一个！
不加锁
主动释放锁
按顺序申请锁
资源一次性分配
破坏死锁的一个小demo（主动释放锁）
#include <iostream> #include<unistd.h> #include <pthread.h> using namespace std; pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; void *pthreadRoutinue(void *args) { pthread_mutex_lock(&mutex); //加锁 cout<<"I get a mutex"<<endl; pthread_mutex_lock(&mutex); //产生死锁 cout<<"i alive again"<<endl; return nullptr; } int main() { pthread_t pid; pthread_create(&pid, nullptr, pthreadRoutinue, nullptr); sleep(3); cout<<"main thread run"<<endl; pthread_mutex_unlock(&mutex);//主线程区解锁 cout<<"main thread unlock"<<endl; sleep(3); return 0; }
四、线程同步
同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问 题，叫做同步 。
1.条件变量
概念
与互斥锁不同，条件变量是用来等待而不是用来上锁的。条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。通常条件变量和互斥锁同时使用。
条件变量使我们可以睡眠等待某种条件出现。条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待"条件变量的条件成立"而挂起；另一个线程使"条件成立"（给出条件成立信号）。
条件变量接口
pthread_cond_init、pthread_cond_destroy：初始化、销毁条件变量
#include <pthread.h> int pthread_cond_destroy(pthread_cond_t *cond); // pthread_cond_t:条件变量类型，类似pthread_mutex_t int pthread_cond_init(pthread_cond_t *restrict cond,constpthread_condattr_t *restrict attr); // 如果是静态或全局的条件变量可使用宏初始化： pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_cond_wait、pthread_cond_signal：等待条件、唤醒线程
#include <pthread.h> // 等待条件满足 int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex); // 唤醒一个线程，在cond等待队列里的第一个线程 int pthread_cond_signal(pthread_cond_t *cond); // 一次唤醒所有线程 int pthread_cond_broadcast(pthread_cond_t *cond); ```
demo
#include <iostream> #include<unistd.h> #include <pthread.h> using namespace std; #define num 5 int ticket =1000; pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER; pthread_cond_t cond=PTHREAD_COND_INITIALIZER; void *active(void *args) { string name=static_cast<const char*>(args); while(true) { pthread_mutex_lock(&mutex); pthread_cond_wait(&cond,&mutex); //调用该函数，会自己释放锁 cout<<name<<" 活动"<<endl; pthread_mutex_unlock(&mutex); } return nullptr; } int main() { pthread_t tids[num]; for(int i=0;i<num;i++) { char * name=new char[64]; snprintf(name,64,"thread-%d",i); //线程创 pthread_create(tids+i,nullptr,active,name); } sleep(3); while(true) { cout<<"main thread wakeup thread..."<<endl; //pthread_cond_signal(&cond); //唤醒cond队列中的一个线程 pthread_cond_broadcast(&cond); //将cond队列中所以线程唤醒 sleep(1); } for(int i=0;i<num;i++) { pthread_join(tids[i],nullptr); //线程等待 } sleep(3); return 0; }
基于阻塞队列实现生产者消费者模型
生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。
生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
生产者消费者模型的优点：解耦 支持并发 支持忙闲不均
实则之前所讲的进程间通信中的管道通信就是一种生产者消费者模型，管道就是让不同的进程能够看到同一份资源，且管道自带同步和互斥的机制。进程间通信的本质其实就是生产者消费者模型。
代码：
blockQueue.hpp
#pragma once #include <iostream> #include <queue> #include <pthread.h> const int gcap = 5; template <class T> class BlockQueue { public: BlockQueue(const int cap = gcap) : _cap(cap) { pthread_mutex_init(&_mutex, nullptr); pthread_cond_init(&_consumerCond, nullptr); pthread_cond_init(&_productorCond, nullptr); } ~BlockQueue() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_consumerCond); pthread_cond_destroy(&_productorCond); } bool isFull() { return _cap == _q.size(); } void push(const T &in) { // 生产 pthread_mutex_lock(&_mutex); while (isFull()) //细节1：使用while ，防止多线程被唤醒生产过多 { // 我们只能在临界区内部，判断临界资源是否就绪 注定了我们在当前一定是持有锁的 pthread_cond_wait(&_productorCond, &_mutex); // 如果队列为满，生产者线程休眠 ,此时持有锁，wait会将锁unlock // 当线程醒来的时候，注定了继续从临界区内部继续运行，因为是在临界区被切走的 // 注定了当线程被唤醒的时候，继续在pthread_cond_wait()函数继续向后运行，又要重新申请锁，申请成功才会彻底返回 } // 没有满，让他继续生产 _q.push(in); //策略，唤醒消费者线程 pthread_cond_signal(&_consumerCond); pthread_mutex_unlock(&_mutex); } void pop(T *out) { pthread_mutex_lock(&_mutex); while (_q.empty()) //队列为空 { pthread_cond_wait(&_consumerCond, &_mutex); } *out = _q.front(); _q.pop(); //策略，唤醒生产者 pthread_cond_signal(&_productorCond); pthread_mutex_unlock(&_mutex); } private: std::queue<T> _q; int _cap; pthread_mutex_t _mutex; pthread_cond_t _consumerCond; // 消费者对应的条件变量 空 wait pthread_cond_t _productorCond; // 生产者对应的条件变量 满 wait }; ```
task.hpp
#pragma once #include <iostream> #include <string> class Task { public: Task() {} Task(int x, int y, char op) : _x(x), _y(y), _op(op), _result(0), _exitCode(0) { } void operator()() { switch (_op) { case '+': _result = _x + _y; break; case '-': _result = _x - _y; break; case '*': _result = _x * _y; break; case '/': if(_y==0) _exitCode=-1; else _result = _x / _y; break; case '%': if(_y==0) _exitCode=-1; else _result = _x % _y; break; default: break; } } std::string formatArg() { return std::to_string(_x)+' '+ _op+ ' '+std::to_string(_y)+" = "; } std::string formatRes() { return std::to_string(_result) + "(" +std::to_string(_exitCode)+")"; } ~Task(){} private: int _x; int _y; char _op; int _result; int _exitCode; }; ```
main.cc
#include "blockQueue.hpp" #include"task.hpp" #include<ctime> #include<unistd.h> void *consumer(void *args) { BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args); while(true) { sleep(1); Task t; //1.将数据从blockqueue中获取 -- 获取到数据 bq->pop(&t); t(); //2.结合某种业务逻辑，处理数据！ std::cout<<"consumer data: "<<t.formatArg()<<t.formatRes()<<std::endl; } } void *productor(void *args) { srand((uint64_t)time(nullptr)^getpid()); BlockQueue<Task> *bq = static_cast<BlockQueue<Task> *>(args); std::string opers="+-*/%"; while(true) { //1.先通过某种渠道获取数据 int x=rand()%20+1; int y=rand()%10+1; //2.将数据推送到blockqueue -- 完成生产过程 char op=opers[rand()%opers.size()]; Task t(x,y,op); bq->push(t); std::cout<<"productor Task: "<<t.formatArg()<<"?"<<std::endl; } } int main() { //BlockQueue<int> *bq = new BlockQueue<int>(); BlockQueue<Task> *bq = new BlockQueue<Task>(); // 单生产，单消费 支持多生产，多消费，因为看到同一快资源，使用同一把锁 pthread_t c, p; pthread_create(&c, nullptr, consumer, bq); pthread_create(&p, nullptr, productor, bq); pthread_join(c, nullptr); pthread_join(p, nullptr); delete bq; return 0; } ```
运行结果：
2.信号量
概念
信号量本质就是一个计数器，用来描述临界区中临界资源的数目大小。
临界资源如果可以被划分为更小的资源，如果处理得当，我们也有可能让多个线程同时访问临界资源，从而实现并发。
但是每个线程想访问临界资源，都得先申请信号量资源。
信号量操作接口
申请信号量成功时，临界资源的数目会减一；释放信号量时，临界资源的数目会加一。
由于信号量是用来维护临界资源的，首先必须得保证自身是安全的，所以常规的对全局变量的++或–操作肯定是不行的。
P操作(申请信号量)
V操作(释放信号量)
sem_init、sem_destroy：初始化销毁信号量(具体用法与mutex和cond十分类似)
#include <semaphore.h> int sem_init(sem_t *sem, int pshared, unsigned int value); // pshared: 默认为0， value：信号量的初始值(count) int sem_destroy(sem_t *sem); // sem_t ：信号量类型 // Link with -pthread. ```
sem_wait、sem_signal: 申请、释放信号量
int sem_wait(sem_t *sem); // P操作 int sem_post(sem_t *sem); // V操作 // Link with -pthread. ```
基于环形队列的生产者消费者模型
但是现在的环形队列的判空判满不再使用中的两种方式判断，因为有了信号量可以判定。
队列为空的时候，消费者和生产者指向同一个位置。(生产和消费线程不能同时进行)(生产者执行)
队列为满的时候，消费者和生产者也指向同一个位置。(生产和消费线程不能同时进行)(消费者执行)
当队列不为空不为满的时候，消费者和生产者不指向同一个位置。(生产和消费线程可以并发执行)
根据上面三种情况，基于环形队列的生产者消费者模型应该遵守以下规则：
生产者不能把消费者套一个圈
消费者不能超过生产者
当指向同一个位置的时候，要根据空、满状态，判断让谁先执行
其他情况，消费者和生产者可以并发执行
实现：
ringQueue.hpp
#pragma once #include <iostream> #include <vector> #include <pthread.h> #include <semaphore.h> static const int N = 5; template <class T> class RingQueue { private: void P(sem_t &s) { sem_wait(&s); } void V(sem_t &s) { sem_post(&s); } void Lock(pthread_mutex_t &m) { pthread_mutex_lock(&m); } void Unlock(pthread_mutex_t &m) { pthread_mutex_unlock(&m); } public: RingQueue(int num = N) : _ring(num), _cap(num) { sem_init(&_data_sem, 0, 0); sem_init(&_space_sem, 0, num); _c_step = _p_step = 0; pthread_mutex_init(&_c_mutex, nullptr); pthread_mutex_init(&_p_mutex, nullptr); } // 生产 void push(const T &in) { // 1. 可以不用在临界区内部做判断，就可以知道临界资源的使用情况 // 2. 什么时候用锁，对应的临界资源，是否被整体使用 P(_space_sem); // P() Lock(_p_mutex); _ring[_p_step++] = in; _p_step %= _cap; Unlock(_p_mutex); V(_data_sem); } // 消费 void pop(T *out) { P(_data_sem); Lock(_c_mutex); *out = _ring[_c_step++]; _c_step %= _cap; Unlock(_c_mutex); V(_space_sem); } ~RingQueue() { sem_destroy(&_data_sem); sem_destroy(&_space_sem); pthread_mutex_destroy(&_c_mutex); pthread_mutex_destroy(&_p_mutex); } private: std::vector<T> _ring; int _cap; // 环形队列容器大小 sem_t _data_sem; // 只有消费者关心 sem_t _space_sem; // 只有生产者关心 int _c_step; // 消费位置 int _p_step; // 生产位置 pthread_mutex_t _c_mutex; pthread_mutex_t _p_mutex; };
task.hpp
#pragma once #include <iostream> #include <string> class Task { public: Task() {} Task(int x, int y, char op) : _x(x), _y(y), _op(op), _result(0), _exitCode(0) { } void operator()() { switch (_op) { case '+': _result = _x + _y; break; case '-': _result = _x - _y; break; case '*': _result = _x * _y; break; case '/': if(_y==0) _exitCode=-1; else _result = _x / _y; break; case '%': if(_y==0) _exitCode=-1; else _result = _x % _y; break; default: break; } } std::string formatArg() { return std::to_string(_x)+' '+ _op+ ' '+std::to_string(_y)+" = "; } std::string formatRes() { return std::to_string(_result) + "(" +std::to_string(_exitCode)+")"; } ~Task(){} private: int _x; int _y; char _op; int _result; int _exitCode; }; ```
main.cc
#include "ringQueue.hpp" #include "task.hpp" #include <ctime> #include <pthread.h> #include <memory> #include <sys/types.h> #include <unistd.h> #include <cstring> using namespace std; const char *ops = "+-*/%"; void *consumerRoutine(void *args) { RingQueue<Task> *rq = static_cast<RingQueue<Task> *>(args); while (true) { Task t; rq->pop(&t); t(); cout << "consumer done, 处理完成的任务是： " << t.formatRes() << endl; } } void *productorRoutine(void *args) { RingQueue<Task> *rq = static_cast<RingQueue<Task> *>(args); while (true) { // sleep(1); int x = rand() % 100; int y = rand() % 100; char op = ops[(x + y) % strlen(ops)]; Task t(x, y, op); rq->push(t); cout << "productor done, 生产的任务是: " << t.formatArg() << endl; } } int main() { srand(time(nullptr) ^ getpid()); RingQueue<Task> *rq = new RingQueue<Task>(); // 单生产单消费 // pthread_t c, p; // pthread_create(&c, nullptr, consumerRoutine, rq); // pthread_create(&p, nullptr, productorRoutine, rq); // pthread_join(c, nullptr); // pthread_join(p, nullptr); //多生产，多消费 pthread_t c[3], p[2]; for (int i = 0; i < 3; i++) pthread_create(c + i, nullptr, consumerRoutine, rq); for (int i = 0; i < 2; i++) pthread_create(p + i, nullptr, productorRoutine, rq); for (int i = 0; i < 3; i++) pthread_join(c[i], nullptr); for (int i = 0; i < 2; i++) pthread_join(p[i], nullptr); delete rq; return 0; } ```
运行结果
五、总结
互斥锁与信号量的异同
互斥锁由同一线程加放锁，信号量可以由不同线程进行PV操作。
计数信号量允许多个线程，且值为剩余可用资源数量。互斥锁保证多个线程对一个共享资源的互斥访问，信号量用于协调多个线程对一系列资源的访问条。
条件变量与信号量的异同
使用条件变量可以一次唤醒所有等待者，而这个信号量没有的功能。
信号量是有一个值，而条件变量是没有的。从实现上来说一个信号量可以是用mutex + count + cond实现的。因为信号量有一个状态，可以精准的同步，信号量可以解决条件变量中存在的唤醒丢失问题。
条件变量一般需要配合互斥锁使用，而信号量可根据情况而定。
有了互斥锁和条件变量还提供信号量的原因是：尽管信号量的意图在于进程间同步，互斥锁和条件变量的意图在于线程间同步，但是信号量也可用于线程间，互斥锁和条件变量也可用于进程间。信号量最有用的场景是用以指明可用资源的数量。