发布时间:2023-03-09 08:30
死锁是指在一组进程中的各个进程均占有不会释放的资源`,但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态
互斥条件:
一个资源每次只能被一个执行流使用
请求与保持条件:
一个执行流因请求资源而阻塞时,对已获得的资源保持不放
不剥夺条件:
一个执行流已获得的资源,在末使用完之前,不能强行剥夺
循环等待条件:
若干执行流之间形成一种头尾相接的循环等待资源的关系
问题:
一个线程一把锁有没有可能形成死锁呢?
答案是当然有可能,例如对一把锁进行多次加锁,而不解锁,就会形成死锁
代码演示:
同步:
在保证数据安全的前提下,让线程能够按照某种特定的顺序访问临界资源,从而有效避免饥饿问题,叫做同步(Linux中的管道就自带同步机制)。
竞态条件:
因为时序问题,而导致程序异常,我们称之为竞态条件。在线程场景下,这种问题也不难理解。
同步存在是为了多线程协同高效的进行合作
pthread原生线程库给用户提供了能够操作条件变量的相关接口,例如初始化条件变量、销毁条件变量等。
返回值:
如果定义一个全局或者静态的条件变量,也可以直接使用宏PTHREAD_COND_INITIALIZER
;来进行初始化,例如:
pthread_cond_t cond = PTHREAD_COND_INITIALIZER
返回值:
返回值:
pthread_cond_brpadcast和pthread_cond_signal的参数和返回值都是一样的
参数:
返回值:
两者的区别:
#include
#include
#include
#include
using namespace std;
//全局的条件变量和互斥锁能被所有线程看到
pthread_mutex_t mtx;
pthread_cond_t cond;
// ctrl thread 控制work线程,让它定期运行
void *ctrl(void *args)
{
string name = (char *)args;
while (true)
{
//唤醒在条件变量下的一个线程
pthread_cond_signal(&cond);
sleep(1);
}
}
void *work(void *args)
{
int number = *(int *)args;
delete (int *)args;
while (true)
{
//在cond条件变量下进行等待
pthread_cond_wait(&cond, &mtx);
cout << \"worker:\" << number << \" is working...\" << endl;
}
}
int main()
{
#define NUM 5
//初始化互斥锁和条件变量
pthread_mutex_init(&mtx, nullptr);
pthread_cond_init(&cond, nullptr);
//创建6个线程,master线程用来控制5个worker线程
pthread_t master;
pthread_t worker[5];
pthread_create(&master, nullptr, ctrl, (void *)\"boss\");
for (int i = 0; i < NUM; ++i)
{
int *number = new int(i);
pthread_create(worker + i, nullptr, work, (void *)number);
}
for (int i = 0; i < NUM; ++i)
{
pthread_join(worker[i], nullptr);
}
pthread_join(master, nullptr);
//释放互斥锁和条件变量
pthread_mutex_destroy(&mtx);
pthread_cond_destroy(&cond);
return 0;
}
运行代码:
如果将 pthread_cond_signal(&cond)改为pthread_cond_broadcast(&cond),再运行代码:
通过上面的测试,我们发现好像worker的执行顺序好像是有序的,这是为什么呢?
其实可以把worker的等待过程想象成队列,当一个worker执行完毕后,又要进行等待时,此时就需要在队尾进行等待,所以在用户看来就是一个排队的过程
为什么pthread_cond_wait()需要互斥量呢?
原因:
为何要使用生产者消费者模型 ?
什么时候该生产,什么时候该消费?
生产者不知道什么时候该生产,但是它知道什么时候该消费。消费者不知道什么时候该消费,但它知道什么时候该生产
。为什么这么说呢?
因为生产者生产数据,放进阻塞队列时,说明已经有数据可以被消费了,此时就应该告诉消费者你该消费了。当阻塞队列被放满时,生产者就无法生产数据,但是什么时候又要开始生产呢?生产者并不知道,因为它不知道消费者什么时候消费。
当消费者消费数据时,阻塞队列的数据就会减少,此时就能告诉生产者,数据被我消费了,你可以开始生产了。当阻塞队列没有数据时,消费者就无法消费数据,但是什么时候又可以开始消费呢?消费者并不知道,因为它不知道生产者什么时候生产。
生产者消费者模型优点:
生产者消费者\"321\"原则----方便理解
3种关系
:生产者&消费者(互斥/同步),生产者&生产者(竞争&互斥),消费者&消费者(竞争&互斥)2种角色
:生产者(n个),消费者(n个),两种执行流一个交易场所(一段缓冲区)
//BlockQueue.hpp
#pragma once
#include
#include
#include
#include
#include
#include
using namespace std;
namespace ns_blockqueue
{
const int default_cap = 5;
template <class T>
class BlockQueue
{
private:
queue<T> _bq; //阻塞队列
int _cap; //队列元素上限
pthread_mutex_t _mtx; //保护临界资源的锁
// 1. 当队列满了,就不应该生产了(不要竞争锁了),而应该让消费者进行消费
// 2. 当队列空了,就不应该消费了(不要竞争锁了),而应该让生产者进行生产
pthread_cond_t _is_full; // bq_满了,消费者应该在该条件变量下等待
pthread_cond_t _is_empty; // bq_空了,生产者应该在该条件变量下等待
bool IsFull()
{
return _bq.size() == _cap;
}
bool IsEmpty()
{
return _bq.size() == 0;
}
void LockQueue()
{
pthread_mutex_lock(&_mtx);
}
void UnLockQueue()
{
pthread_mutex_unlock(&_mtx);
}
void ProducterWait()
{
// pthread_cond_wait
// 1. 调用的时候,会首先自动释放_mtx,然后在挂起自己
// 2. 返回的时候,首先会自动竞争锁,获取到锁之后,才能返回
pthread_cond_wait(&_is_empty, &_mtx);
}
void ConsumerWait()
{
pthread_cond_wait(&_is_full, &_mtx);
}
void WakeupConsumer()
{
pthread_cond_signal(&_is_full);
}
void WakeupProducter()
{
pthread_cond_signal(&_is_empty);
}
public:
BlockQueue(int cap = default_cap)
: _cap(cap)
{
pthread_mutex_init(&_mtx, nullptr);
pthread_cond_init(&_is_full, nullptr);
pthread_cond_init(&_is_empty, nullptr);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mtx);
pthread_cond_destroy(&_is_full);
pthread_cond_destroy(&_is_empty);
}
//生产函数,向队列里放数据
void Push(const T &in) // in 为输入型参数
{
LockQueue();
//临界区
if (IsFull())
{
//队列满了,则无法生产,需要等待消费者消费
//等待时,把线程挂起,但是现在该线程是持有锁的
ProducterWait();
}
_bq.push(in);
WakeupConsumer();
UnLockQueue();
}
//消费函数,向队列里拿数据
void Pop(T *out) // out 为输出型参数
{
//临界区
LockQueue();
if (IsEmpty())
{
//队列空了,则无法消费,需要等待生产者生产
ConsumerWait();
}
*out = _bq.front();
_bq.pop();
WakeupProducter();
UnLockQueue();
}
};
}
//CpTest.cpp
#include \"BlockQueue.hpp\"
using namespace ns_blockqueue;
void *consumer(void *args)
{
BlockQueue<int> *_bq = (BlockQueue<int> *)args;
while (true)
{
int data = 0;
_bq->Pop(&data);
cout << \"消费者消费了一个数据: \" << data << endl;
sleep(1);
}
}
void *producter(void *args)
{
BlockQueue<int> *_bq = (BlockQueue<int> *)args;
while (true)
{
//产生0到20的随机数
int data = rand() % 20 + 1;
cout << \"生产者生产数据: \" << data << endl;
_bq->Push(data);
//sleep(2);
}
}
int main()
{
srand((long long)time(nullptr));//种下随机数种子
BlockQueue<int> *bq = new BlockQueue<int>();
pthread_t c, p; // c为消费者, 跑为生产者
pthread_create(&c, nullptr, consumer, (void *)bq);
pthread_create(&p, nullptr, producter, (void *)bq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}
使用sleep函数让消费者每隔1秒消费一个数据:
使用sleep函数让生产者每隔1秒生产一个数据,消费者每隔2秒消费一个数据:
#include
int sem_init(sem_t *sem, int pshared, unsigned int value)
参数:
返回值:
-成功返回0,失败返回-1,同时错误码将被设置
int sem_destroy(sem_t *sem)
参数:
返回值:
-成功返回0,失败返回-1,同时错误码将被设置
int sem_wait(sem_t *sem);//P()操作
sem_wait表示需要申请信号量,如果申请不成功则会被挂起进行等待
参数:
返回值:
-成功返回0,失败返回-1,同时错误码将被设置
int sem_post(sem_t *sem)//V()操作
参数:
返回值:
-成功返回0,失败返回-1,同时错误码将被设置
//ring_queue.hpp
#pragma once
#include
#include
#include
using namespace std;
namespace ns_ring_queue
{
const int g_cap_default = 10;
template <class T>
class RingQueue
{
private:
vector<T> _ring_queue;
int _cap;
//生产者关心空位置资源
sem_t blank_sem;
//消费者关心数据资源
sem_t data_sem;
//生产者生产数据的位置
int p_step;
//消费者消费数据的位置
int c_step;
public:
RingQueue(int cap = g_cap_default)
: _cap(cap), _ring_queue(cap), c_step(0), p_step(0)
{
sem_init(&blank_sem, 0, cap);
sem_init(&data_sem, 0, 0);
}
~RingQueue()
{
sem_destroy(&blank_sem);
sem_destroy(&data_sem);
}
//生产
void Push(const T &in)
{
//申请信号量
sem_wait(&blank_sem); // P(空位置)
//可以生产了,可是往哪个位置生产呢?
_ring_queue[p_step] = in;
sem_post(&data_sem); // V(数据)
p_step++;
p_step %= _cap;
}
//消费
void Pop(T *out)
{
sem_wait(&data_sem); // P(数据)
*out = _ring_queue[c_step];
sem_post(&blank_sem); // V(空位置)
c_step++;
c_step %= _cap;
}
};
}
//ring_cp.cpp
#include \"ring_queue.hpp\"
#include
#include
#include
using namespace ns_ring_queue;
void *consumer(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
int data = 0;
rq->Pop(&data);
cout << \"消费数据是:\" << data << endl;
sleep(1);
}
}
void *producter(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
int data = rand() % 20 + 1;
cout << \"生产数据是:\" << data << endl;
rq->Push(data);
sleep(1);
}
}
int main()
{
srand((long long)time(nullptr));
RingQueue<int> *rq = new RingQueue<int>();
pthread_t c, p;
pthread_create(&c, nullptr, consumer, (void *)rq);
pthread_create(&c, nullptr, producter, (void *)rq);
pthread_join(c, nullptr);
pthread_join(p, nullptr);
return 0;
}
不过上诉代码是单生产者单消费者。我们只需要加两把互斥锁,分别用于生产者和消费者,就可以将其改为多生产者多消费者:
//ring_queue.hpp
#pragma once
#include
#include
#include
#include
//多生产者多消费者
using namespace std;
namespace ns_ring_queue
{
const int g_cap_default = 10;
template <class T>
class RingQueue
{
private:
vector<T> _ring_queue;
int _cap;
//生产者关心空位置资源
sem_t blank_sem;
//消费者关心数据资源
sem_t data_sem;
//生产者生产数据的位置
int p_step;
//消费者消费数据的位置
int c_step;
//加两把锁,实现多生产者多消费者
pthread_mutex_t c_mtx;
pthread_mutex_t p_mtx;
public:
RingQueue(int cap = g_cap_default)
: _cap(cap), _ring_queue(cap), c_step(0), p_step(0)
{
sem_init(&blank_sem, 0, cap);
sem_init(&data_sem, 0, 0);
pthread_mutex_init(&c_mtx, nullptr);
pthread_mutex_init(&p_mtx, nullptr);
}
~RingQueue()
{
sem_destroy(&blank_sem);
sem_destroy(&data_sem);
pthread_mutex_destroy(&c_mtx);
pthread_mutex_destroy(&p_mtx);
}
//生产
void Push(const T &in)
{
//申请信号量
sem_wait(&blank_sem); // P(空位置)
//注意:lock放在wait之后
pthread_mutex_lock(&p_mtx);
//可以生产了,可是往哪个位置生产呢?
_ring_queue[p_step] = in;
sem_post(&data_sem); // V(数据)
p_step++;
p_step %= _cap;
pthread_mutex_unlock(&p_mtx);
}
//消费
void Pop(T *out)
{
sem_wait(&data_sem); // P(数据)
//注意:lock放在wait之后
pthread_mutex_lock(&c_mtx);
*out = _ring_queue[c_step];
sem_post(&blank_sem); // V(空位置)
c_step++;
c_step %= _cap;
pthread_mutex_unlock(&c_mtx);
}
};
}
//ring_cp.cpp
#include \"ring_queue1.hpp\"
#include
#include
#include
using namespace ns_ring_queue;
void *consumer(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
int data = 0;
rq->Pop(&data);
cout << \"线程\"
<< \"[\" << pthread_self() << \"]\"
<< \"消费数据:\" << data << endl;
sleep(1);
}
}
void *producter(void *args)
{
RingQueue<int> *rq = (RingQueue<int> *)args;
while (true)
{
int data = rand() % 20 + 1;
cout << \"线程\"
<< \"[\" << pthread_self() << \"]\"
<< \"生产数据:\" << data << endl;
rq->Push(data);
sleep(1);
}
}
int main()
{
srand((long long)time(nullptr));
RingQueue<int> *rq = new RingQueue<int>();
pthread_t c1, c2, c3, p1, p2, p3;
pthread_create(&c1, nullptr, consumer, (void *)rq);
pthread_create(&c2, nullptr, consumer, (void *)rq);
pthread_create(&c3, nullptr, consumer, (void *)rq);
pthread_create(&p1, nullptr, producter, (void *)rq);
pthread_create(&p2, nullptr, producter, (void *)rq);
pthread_create(&p3, nullptr, producter, (void *)rq);
pthread_join(c1, nullptr);
pthread_join(c2, nullptr);
pthread_join(c3, nullptr);
pthread_join(p1, nullptr);
pthread_join(p2, nullptr);
pthread_join(p3, nullptr);
return 0;
}