【Linux学习笔记】一篇文章彻底搞定“Linux生产者与消费者“！

本章重点

1.生产者消费者模型
2.posix信号量，以及读写锁。
3. 理解基于读写锁的读者写者问题。

一. 生产者消费者模型

为何要使用生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

生产者消费者模型优点

解耦
支持并发
支持忙闲不均

基于BlockingQueue的生产者消费者模型

在多线程编程中阻塞队列(Blocking Queue)是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出(以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞

C++ queue模拟阻塞队列的生产消费模型

makefile

blockqueue:main.ccg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -f blockqueue

BlockQueue.hpp

#include <iostream>
#include <queue>
#include <pthread.h>
#include <unistd.h>using namespace std;template<class T>
class BlockQueue
{static const int defaultcap = 5;
public:BlockQueue(int capacity = defaultcap): _capacity(capacity){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond_productor, nullptr);pthread_cond_init(&_cond_consumer, nullptr);}T pop(){pthread_mutex_lock(&_mutex);if(_q.size() == 0){// 在调用的时候,自动释放锁pthread_cond_wait(&_cond_consumer, &_mutex); }T x = _q.front();_q.pop(); // 当前线程想消费就能消费吗?不一定,你首先得确保消费条件满足pthread_cond_signal(&_cond_productor);pthread_mutex_unlock(&_mutex);return x;}void push(const T& x){pthread_mutex_lock(&_mutex);// 判断本身也是临界资源,如果放外面,就会资源并发访问if(_q.size() == _capacity){// 在调用的时候,自动释放锁pthread_cond_wait(&_cond_productor, &_mutex); }// 1.队列没满 2.被唤醒_q.push(x);// 当前线程想生产就能生产吗?不一定,你首先得确保生产条件满足pthread_cond_signal(&_cond_consumer);// 当前线程push的时候，也没有其他线程pop呢?所以我们要加锁pthread_mutex_unlock(&_mutex);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond_consumer);pthread_cond_destroy(&_cond_productor);}
private:queue<T> _q; // 共享资源int _capacity; //极值// 实现多线程之间的同步问题pthread_mutex_t _mutex;pthread_cond_t _cond_consumer;pthread_cond_t _cond_productor;
};

main.cc

#include "BlockQueue.hpp"void* consumer(void* args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);while(true){// 消费int data = bq->pop();cout << "消费了一个数据: " << data << endl;}
}void* productor(void* args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);int data = 0;while(true){// 生产data++;bq->push(data);cout << "生产了一个数据: " << data << endl;}
}int main()
{BlockQueue<int> *bq = new BlockQueue<int>;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;
}

ok啊，代码已经手撕好了，现在我们就来执行一下，首先我们让生产者先来生产，让消费者不急着消费。

然后我们再让生产者不急着生产，看看运行结果：

但是我们当前都是生产一个数据，然后再消费一个数据，我们能不能生产一批数据然后消费一批数据呢？我们可以设置两个水平线，当消费者消耗到一定程度low_water，让生产者去生产，生产者生产到一定程度high_water，就去让消费者去消费。

#include <iostream>
#include <queue>
#include <pthread.h>
#include <unistd.h>using namespace std;template<class T>
class BlockQueue
{static const int defaultcap = 8;
public:BlockQueue(int capacity = defaultcap): _capacity(capacity){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond_productor, nullptr);pthread_cond_init(&_cond_consumer, nullptr);_low_water = 3;_high_water = 5;}// 消费者消耗数据T pop(){pthread_mutex_lock(&_mutex);if(_q.size() == 0){// 在调用的时候,自动释放锁pthread_cond_wait(&_cond_consumer, &_mutex); }T x = _q.front();_q.pop(); // 当前线程想消费就能消费吗?不一定,你首先得确保消费条件满足if(_q.size() < _low_water)  // 通知生产者来生产pthread_cond_signal(&_cond_productor);pthread_mutex_unlock(&_mutex);return x;}void push(const T& x){pthread_mutex_lock(&_mutex);// 判断本身也是临界资源,如果放外面,就会资源并发访问if(_q.size() == _capacity){// 在调用的时候,自动释放锁pthread_cond_wait(&_cond_productor, &_mutex); }// 1.队列没满 2.被唤醒_q.push(x);// 当前线程想生产就能生产吗?不一定,你首先得确保生产条件满足if(_q.size() > _high_water) // 通知消费者来消费pthread_cond_signal(&_cond_consumer);// 当前线程push的时候，也没有其他线程pop呢?所以我们要加锁pthread_mutex_unlock(&_mutex);}~BlockQueue(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond_consumer);pthread_cond_destroy(&_cond_productor);}
private:queue<T> _q; // 共享资源int _capacity; //极值// 实现多线程之间的同步问题pthread_mutex_t _mutex;pthread_cond_t _cond_consumer;pthread_cond_t _cond_productor;int _low_water;int _high_water;
};

运行结果：

这里很奇怪为什么这里生产了5个数据，但是却消费了6个数据？因为Linux一切皆文件，所有多线程往显示器打印的时候，显示器就是一个文件，此时它就是一个共享资源，所以此时那个线程的竞争力更强它就会优先输出，所以是上面的结果，我们可以对显示器文件进行加锁，保证能输出正确的结果，当生产者打印了5次生产，就通知消费者去打印消费5次，紧接着生产者打印了5次生产...

#include "BlockQueue.hpp"pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond1 = PTHREAD_COND_INITIALIZER;
pthread_cond_t cond2 = PTHREAD_COND_INITIALIZER;
int cnt = 0;void* consumer(void* args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);while(true){// 消费int data = bq->pop();pthread_mutex_lock(&mutex);if(cnt == 0)pthread_cond_wait(&cond2, &mutex);cout << "消费了一个数据: " << data << endl;cnt--;if(cnt <= 0)pthread_cond_signal(&cond1);pthread_mutex_unlock(&mutex);//sleep(2);}
}void* productor(void* args)
{BlockQueue<int> *bq = static_cast<BlockQueue<int>*>(args);int data = 0;while(true){// 生产data++;bq->push(data);pthread_mutex_lock(&mutex);if(cnt == 5)pthread_cond_wait(&cond1, &mutex);cout << "生产了一个数据: " << data << endl;cnt++;if(cnt >= 5){pthread_cond_signal(&cond2);}pthread_mutex_unlock(&mutex);//usleep(1);}
}int main()
{BlockQueue<int> *bq = new BlockQueue<int>;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;
}

运行结果：

此时输出结果就对了，我们经常听到书上说生产者消费者模型是高效的，那么它的高效体现在哪里呢？

我们这里的高效性生产者生产一个数据，消费者消费一个数据，按照这样那么他们其实是串行的，那么效率不高，效率高体现在一个访问临界区代码，一个访问非临界区代码，两个并发执行。我们来模拟一个场景：给我们的生产者获取数据并产生一个计算任务，让消费者去消费任务并执行这个任务。

Task.hpp

#pragma once#include <iostream>
#include <string>using namespace std;enum
{div_zero = 1,mod_zero = 2,unknown = 3
};const string opsum = "+-*/%";class Task
{
public:Task(int x, int y, char op): _x(x), _y(y), _op(op), _result(0), _exitcode(0){}void run(){switch(_op){case '+':_result = _x + _y;break;case '-':_result = _x - _y;break;case '*':_result = _x * _y;break;case '/':if(_y == 0)_exitcode = div_zero;else_result = _x / _y;break;case '%':if(_y == 0)_exitcode = mod_zero;else_result = _x % _y;break;default:_exitcode = unknown;break;}}string  GetResult(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=";ret += to_string(_result);ret += ",[exitcode: ";ret += to_string(_exitcode);ret += "]";return ret;}string GetTask(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=?";return ret;}~Task(){}
private:int _x;int _y;char _op;int _result;int _exitcode;
};

main.cc

#include "BlockQueue.hpp"
#include "Task.hpp"
#include <ctime>void* consumer(void* args)
{BlockQueue<Task> *bq = static_cast<BlockQueue<Task>*>(args);while(true){// 消费者消费数据Task task = bq->pop();// 加工数据task.run();cout << "处理了一个任务: " << task.GetResult() << endl;//sleep(2);}
}void* productor(void* args)
{int len = opsum.size();BlockQueue<Task> *bq = static_cast<BlockQueue<Task>*>(args);while(true){// 生产者获取数据int x = rand() % 10 + 1;  // [1,10]int y = rand() % 10 + 1;  // [1,10]char op = opsum[rand() % len];Task task(x, y, op);// 生产任务bq->push(task);cout << "生产了一个任务: " << task.GetTask() << endl; sleep(1);}
}int main()
{srand(time(nullptr));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;
}

我们来看看运行结果：

此时一个单线程生产任务，一个单线程处理任务的场景就是这样啦！下面我们在来看看伪唤醒的情况。

所以在加速之后的判断资源有没有就绪的时候，我们就要防止出现伪唤醒的情况发送。

好，这一点我们理解之后，我们就要来看看多生产者，多消费者之间的场景了，我们只需要在main函数添加即可。

int main()
{srand(time(nullptr));BlockQueue<Task> *bq = new BlockQueue<Task>;pthread_t c[3], p[5];for(int i = 0; i < 3; i++){pthread_create(c + i, nullptr, consumer, bq);}for(int i = 0; i < 5; i++){pthread_create(p + i, nullptr, productor, bq);}for(int i = 0; i < 3; i++){pthread_join(c[i], nullptr);}for(int i = 0; i < 5; i++){pthread_join(p[i], nullptr);}delete bq;return 0;
}

此时我们就能观察结果了。

为什么改main函数就能是多生产者，多消费者之间的场景了？因为此时我们用了同意一把锁，生产者和生产者之间是互斥关系，消费者和消费者之间是互斥关系，生产者和消费者满足互斥和同步关系。有人会说此时多线程同样也只有一个人能访问锁呀！那有什么意义呢？高效性，还记得吗？此时可以获取数据和处理数据上减少时间！！！

二、POSIX信号量

POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。但POSIX可以用于线程间同步。

总结一下就是未来我们将一个临界资源分为很多份，每个线程想去访问这个临界资源中的某一个，先不急着访问，先全部去竞争式的去抢占我们的信号量，申请到信号量的线程，将来一定有一份临界资源分给你，没申请到进不能进入到临界资源中去访问某一份资源。

但是此时信号量也式临界资源，谁来保证信号量的安全呢？直接说结论，信号量的设计必须是原子的！！！

⭐申请信号量，本质是对计数器--，P操作

⭐释放信号量，本质是对计数器++，V操作

初始化信号量

#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()

上一节生产者----消费者的例子是基于queue的,其空间可以动态分配,现在基于固定大小的环形队列重写这个程序（POSIX信号量）:

基于环形队列的生产消费模型

环形队列采用数组模拟，用模运算来模拟环状特性

环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来判断满或者空。另外也可以预留一个空的位置，作为满的状态

但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程

makefile

RingQueue:main.cppg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -f RingQueue

RingQueue.hpp

#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <vector>
#include <semaphore.h>
#include <ctime>
using namespace std;template<class T>
class RingQueue
{
private:void P(sem_t &sem){sem_wait(&sem);}void V(sem_t &sem){sem_post(&sem);}
public:RingQueue(int capacity = 5): _ringqueue(capacity)  , _capacity(capacity), _cstep(0), _pstep(0){sem_init(&_cdatasem, 0, 0); // 消费者sem_init(&_pspacesem, 0, capacity); // 生产者}void push(const T& in) //生产{P(_pspacesem); //申请空间资源,申请成功空间少一个// 生产_ringqueue[_pstep] = in;V(_cdatasem); // 得到数据资源,得到成功数据多一个// 更新生产者下标,维持环形特征_pstep++;_pstep %= _capacity;}void pop(T *out) // 消费{P(_cdatasem); // 申请数据资源,申请成功数据少一个// 消费*out = _ringqueue[_cstep];V(_pspacesem);// 释放空间资源,释放成功空间多一个// 更新消费者下标,维持环形特征_cstep++;_cstep %= _capacity;}~RingQueue(){sem_destroy(&_cdatasem);sem_destroy(&_pspacesem);}
private:vector<T> _ringqueue;int _capacity;int _cstep;//消费者下标int _pstep;//生产者下标//信号量sem_t _cdatasem; // 消费者信号 -> 关注数据资源sem_t _pspacesem; // 生产者信号 -> 关注空间资源
};

main.cpp

#include "RingQueue.hpp"//1.下标为空和为满的时候,只能有一个线程进入环形队列,从而保证互斥性
//2.为空的时候只让生产者去运行,为满的时候只让消费者去运行
//  此时在同一个位置时,此时让生产者和消费者具有一定的顺序性,出现了同步性
//  如过不为空,那么生产者和消费者的下标都是不同的,运行互不影响,具有并发性
//  但是此时对于生产者和生产者之间的互斥性不能保证,因为下标只有一个
//  但是此时对于消费者和消费者之间的互斥性不能保证,因为下标只有一个
//  但是此时是单生产者,单消费者,所以没有影响
void* Productor(void* args)
{RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);while(true){// 1.获取数据int data = rand() % 10 + 1;// 2.生产数据rq->push(data);cout << "Productor data done, data is: " << data << endl;sleep(1);}return nullptr;
}
void* Consumer(void* args)
{RingQueue<int>* rq = static_cast<RingQueue<int>*>(args);while(true){// 1.消费数据int data = 0;rq->pop(&data);cout << "Consumer get data, data is: " << data << endl;// 2.处理数据        sleep(1);}return nullptr;
}int main()
{srand(time(nullptr));RingQueue<int>* rq = new RingQueue<int>;pthread_t c, p;pthread_create(&c, nullptr, Productor, rq);pthread_create(&p, nullptr, Consumer, rq);pthread_join(c, nullptr);pthread_join(p, nullptr);return 0;}

好了，代码已经手撕完成了，直接来测试，首先为满测试让生产者先等待3秒，然后再让生产者生产数据，看看结果如何。

此时我们会发现执行sleep(3)生产者不生产，消费者就被阻塞了，3秒过后，生产者生产了一个数据，消费者立马消费数据，随后执行sleep(3)，消费者不生产了，消费者就又被阻塞了，等待下一个3秒是才会消费数据。现在我们再来测试一下让消费者先等待3秒，然后再去让消费者消费数据。

此时我们会发现刚开始由于没有消费者消费，生产者会一直生产数据，直到队列为满就阻塞住了，等待3秒之后，消费者就消费数据，刚消费一个生产者立马就生产一个，并且我们还能发现消费者消费的顺序和生产者当初生产数据的顺序是一样的。

那如果我们是多生产者，多消费者，此时我们就需要维护生产者和生产者之间的互斥性和消费者和消费者之间的互斥性，所以此时我们就需要访问下标的让其访问具有安全性。我们该如何来修改代码呢？

加锁的位置在哪呢？

直接上代码：

main.cc

#include "RingQueue.hpp"//  1.下标为空和为满的时候,只能有一个线程进入环形队列,从而保证互斥性
//  2.为空的时候只让生产者去运行,为满的时候只让消费者去运行
//  此时在同一个位置时,此时让生产者和消费者具有一定的顺序性,出现了同步性
//  如过不为空,那么生产者和消费者的下标都是不同的,运行互不影响,具有并发性
//  但是此时对于生产者和生产者之间的互斥性不能保证,因为下标只有一个
//  但是此时对于消费者和消费者之间的互斥性不能保证,因为下标只有一个
//  但是此时是单生产者,单消费者,所以没有影响
void *Productor(void *args)
{RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);while (true){// sleep(3);//  1.获取数据int data = rand() % 10 + 1;// 2.生产数据rq->push(data);cout << "Productor data done, data is: " << data << endl;}return nullptr;
}
void *Consumer(void *args)
{RingQueue<int> *rq = static_cast<RingQueue<int> *>(args);while (true){sleep(3);// 1.消费数据int data = 0;rq->pop(&data);cout << "Consumer get data, data is: " << data << endl;// 2.处理数据}return nullptr;
}int main()
{srand(time(nullptr));RingQueue<int> *rq = new RingQueue<int>;pthread_t c[5], p[3];for (int i = 0; i < 5; i++){pthread_create(c + i, nullptr, Productor, rq);}for (int i = 0; i < 3; i++){pthread_create(p + i, nullptr, Consumer, rq);}for (int i = 0; i < 5; i++){pthread_join(c[i], nullptr);}for (int i = 0; i < 3; i++){pthread_join(p[i], nullptr);}return 0;
}

RingQueue.hpp

#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <vector>
#include <semaphore.h>
#include <ctime>using namespace std;template <class T>
class RingQueue
{
private:void P(sem_t &sem){sem_wait(&sem);}void V(sem_t &sem){sem_post(&sem);}void Lock(pthread_mutex_t &mutex){pthread_mutex_lock(&mutex);}void Unlock(pthread_mutex_t &mutex){pthread_mutex_unlock(&mutex);}public:RingQueue(int capacity = 5): _ringqueue(capacity), _capacity(capacity), _cstep(0), _pstep(0){sem_init(&_cdatasem, 0, 0);         // 消费者sem_init(&_pspacesem, 0, capacity); // 生产者pthread_mutex_init(&_cmutex, nullptr);pthread_mutex_init(&_pmutex, nullptr);}void push(const T &in) // 生产{// 多个线程来申请信号量,一定会出现都申请信号量成功的现象// 此时对这个环形队列的下标展开了竞争// 从而就会出现值覆盖的情况// 多生产者和多消费的场景在任一时刻,依然只允许一个生产者一个消费者进入队列// 消费者和消费者之间对下标访问需要有一把锁：_cmutex;// 生产者和生产者之间对下标访问需要有一把锁：_pmutex;// 加锁位置应该放在那里呢？// 1.首先我们要知道信号量资源是原子的,它不需要加锁的,不需要被保护// 之前我们也曾提到临界区的代码量一定要少// 所以一般而言我们不需要加锁信号量// 2.如果先申请锁,再去申请信号量,这样他们之间一定是串行的// 但是把申请信号量放在加锁之前,一个线程释放锁的时候,其他线程早已经申请到了信号量// 此时只用去申请锁,一定程度上做到了申请信号量和申请锁的并行P(_pspacesem); // 申请空间资源,申请成功空间少一个Lock(_pmutex);// 生产_ringqueue[_pstep] = in;// 更新生产者下标,维持环形特征_pstep++;_pstep %= _capacity;Unlock(_pmutex);V(_cdatasem); // 得到数据资源,得到成功数据多一个}void pop(T *out) // 消费{P(_cdatasem); // 申请数据资源,申请成功数据少一个Lock(_cmutex);// 消费*out = _ringqueue[_cstep];// 更新消费者下标,维持环形特征_cstep++;_cstep %= _capacity;Unlock(_cmutex);V(_pspacesem); // 释放空间资源,释放成功空间多一个}~RingQueue(){sem_destroy(&_cdatasem);sem_destroy(&_pspacesem);pthread_mutex_destroy(&_cmutex);pthread_mutex_destroy(&_pmutex);}private:vector<T> _ringqueue;int _capacity;int _cstep; // 消费者下标int _pstep; // 生产者下标// 信号量sem_t _cdatasem;  // 消费者信号 -> 关注数据资源sem_t _pspacesem; // 生产者信号 -> 关注空间资源pthread_mutex_t _cmutex;pthread_mutex_t _pmutex;
};

此时我们来运行一下看看结果：

但是上面观察的还是不够明显，我们不知道是不是有多个线程执行的，是哪个线程执行的。

#include "RingQueue.hpp"struct ThreadData
{RingQueue<int> *rq;string threadname;
};
void *Productor(void *args)
{ThreadData* td = static_cast<ThreadData*>(args);RingQueue<int> *rq = td->rq;string name = td->threadname;while (true){// sleep(3);//  1.获取数据int data = rand() % 10 + 1;// 2.生产数据rq->push(data);cout << "Productor data done, data is: " << data << ", who: " << name << endl;}return nullptr;
}
void *Consumer(void *args)
{ThreadData* td = static_cast<ThreadData*>(args);RingQueue<int> *rq = td->rq;string name = td->threadname;while (true){sleep(1);// 1.消费数据int data = 0;rq->pop(&data);cout << "Consumer get data, data is: " << data << ", who: " << name  << endl;// 2.处理数据}return nullptr;
}int main()
{srand(time(nullptr));RingQueue<int> *rq = new RingQueue<int>;pthread_t c[5], p[3];for (int i = 0; i < 5; i++){ThreadData* td = new ThreadData;td->rq = rq;td->threadname = "Productor-" + to_string(i);pthread_create(c + i, nullptr, Productor, td);}for (int i = 0; i < 3; i++){ThreadData* td = new ThreadData;td->rq = rq;td->threadname = "Consumer-" + to_string(i);pthread_create(p + i, nullptr, Consumer, td);}for (int i = 0; i < 5; i++){pthread_join(c[i], nullptr);}for (int i = 0; i < 3; i++){pthread_join(p[i], nullptr);}return 0;
}

运行结果：

我们上面生产者让它获取了数据，然后再生产数据，但是消费者仅仅只做了消费数据，并没有处理数据，所以我们就让消费者来处理一下任务，为了方便观察，我们就退回到单生产单消费。

Tas.hpp

#pragma once#include <iostream>
#include <string>using namespace std;enum
{div_zero = 1,mod_zero = 2,unknown = 3
};const string opsum = "+-*/%";class Task
{
public:Task(){}Task(int x, int y, char op): _x(x), _y(y), _op(op), _result(0), _exitcode(0){}void run(){switch (_op){case '+':_result = _x + _y;break;case '-':_result = _x - _y;break;case '*':_result = _x * _y;break;case '/':if (_y == 0)_exitcode = div_zero;else_result = _x / _y;break;case '%':if (_y == 0)_exitcode = mod_zero;else_result = _x % _y;break;default:_exitcode = unknown;break;}}string GetResult(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=";ret += to_string(_result);ret += ",[exitcode: ";ret += to_string(_exitcode);ret += "]";return ret;}string GetTask(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=?";return ret;}~Task(){}private:int _x;int _y;char _op;int _result;int _exitcode;
};

main.cpp

#include "RingQueue.hpp"
#include "Task.hpp"struct ThreadData
{RingQueue<Task> *rq;string threadname;
};
void *Productor(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->rq;string name = td->threadname;int len = opsum.size();while (true){sleep(1);//  1.获取数据int data1 = rand() % 10 + 1;usleep(10);int data2 = rand() % 10;usleep(10);char op = opsum[rand() % len];Task t(data1, data2, op);// 2.生产数据rq->push(t);cout << "Productor task done, task is: " << t.GetTask() << ", who: " << name << endl;}return nullptr;
}
void *Consumer(void *args)
{ThreadData *td = static_cast<ThreadData *>(args);RingQueue<Task> *rq = td->rq;string name = td->threadname;while (true){// 1.消费数据Task t;rq->pop(&t);// 2.处理数据t.run();cout << "Consumer deal task, result is: " << t.GetResult() << ", who: " << name << endl;}return nullptr;
}int main()
{srand(time(nullptr));RingQueue<Task> *rq = new RingQueue<Task>;pthread_t c[5], p[3];for (int i = 0; i < 1; i++){ThreadData *td = new ThreadData;td->rq = rq;td->threadname = "Productor-" + to_string(i);pthread_create(c + i, nullptr, Productor, td);}for (int i = 0; i < 1; i++){ThreadData *td = new ThreadData;td->rq = rq;td->threadname = "Consumer-" + to_string(i);pthread_create(p + i, nullptr, Consumer, td);}for (int i = 0; i < 1; i++){pthread_join(c[i], nullptr);}for (int i = 0; i < 1; i++){pthread_join(p[i], nullptr);}return 0;
}

运行结果：

三、线程池

注意：

线程池中的任务队列是会被多个执行流同时访问的临界资源，因此我们需要引入互斥锁对任务队列进行保护。线程池当中的线程要从任务队列里拿任务，前提条件是任务队列中必须要有任务，因此线程池当中的线程在拿任务之前，需要先判断任务队列当中是否有任务，若此时任务队列为空，那么该线程应该进行等待，直到任务队列中有任务时再将其唤醒，因此我们需要引入条件变量，接下来直接上手代码：

makefile

threadpool:main.cppg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -f threadpool

Task.hpp

#pragma once#include <iostream>
#include <string>using namespace std;enum
{div_zero = 1,mod_zero = 2,unknown = 3
};const string opsum = "+-*/%";class Task
{
public:Task(){}Task(int x, int y, char op): _x(x), _y(y), _op(op), _result(0), _exitcode(0){}void run(){switch (_op){case '+':_result = _x + _y;break;case '-':_result = _x - _y;break;case '*':_result = _x * _y;break;case '/':if (_y == 0)_exitcode = div_zero;else_result = _x / _y;break;case '%':if (_y == 0)_exitcode = mod_zero;else_result = _x % _y;break;default:_exitcode = unknown;break;}}string GetResult(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=";ret += to_string(_result);ret += ",[exitcode: ";ret += to_string(_exitcode);ret += "]";return ret;}string GetTask(){string ret = to_string(_x);ret += _op;ret += to_string(_y);ret += "=?";return ret;}~Task(){}private:int _x;int _y;char _op;int _result;int _exitcode;
};

ThreadPool.hpp

#include <iostream>
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>using namespace std;struct ThreadInfo
{pthread_t tid;string name;
};template <class T>
class ThreadPool
{static const int defaultnum = 5;public:ThreadPool(int num = defaultnum): _threads(num){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond, nullptr);}void* HandlerTask(void* args){while(true){sleep(1);cout << "new thread wait task..." << endl;}}void push(const T& t){   }void start(){int num = defaultnum;for (int i = 0; i < num; i++){_threads[i].name = "thread-" + to_string(i + 1);pthread_create(&(_threads[i].tid), nullptr, HandlerTask, nullptr);}}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}private:vector<ThreadInfo> _threads;queue<T> _tasks;pthread_mutex_t _mutex;pthread_cond_t _cond;
};

main.cpp

#include "threadpool.hpp"
#include "Task.hpp"int main()
{ThreadPool<Task>* tp = new ThreadPool<Task>(5); tp->start(); // 创建五个线程while(true){// 1.构建任务sleep(1);// 2.交给线程池处理cout << "main thread is runing..." << endl;}
}

然后我们来编译运行一下，此时我们只是创建了五个线程，然后让他们输出线程在等待任务，主线程正在运行。

此时我们发现程序崩溃了，为什么呢？这就要涉及到我们C++的知识了，我们上面提示到HandlerTask出现问题了，我们把它截出来。

此时HandlerTask作为类的成员函数，该函数的第一个参数是隐藏的this指针，因此这里的HandlerTask函数，虽然看起来只有一个参数，而实际上它有两个参数，参数个数不对，函数匹配不上，此时直接将该HandlerTask函数作为创建线程时的执行例程是不行的，无法通过编译。

怎么解决呢？静态成员函数属于类，而不属于某个对象，也就是说静态成员函数是没有隐藏的this指针的，因此我们需要将HandlerTask设置为静态方法，此时HandlerTask函数只有一个参数类型为void*的参数。

此时编译就没有问题了，我们直接来运行一下

此时这些线程就已经创建好了，都在嗷嗷待哺，等待执行队列里面的任务，那我们接下来继续写。

#include <iostream>
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>using namespace std;struct ThreadInfo
{pthread_t tid;string name;
};template <class T>
class ThreadPool
{static const int defaultnum = 5;void Lock(){pthread_mutex_lock(&_mutex);}void Unlock(){pthread_mutex_unlock(&_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}void Sleep(){pthread_cond_wait(&_cond,&_mutex);}
public:ThreadPool(int num = defaultnum): _threads(num){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond, nullptr);}static void* HandlerTask(void* args){while(true){Lock();while(_tasks.size() == 0){Sleep();}T t = _tasks.front();_tasks.pop();Unlock();t.run(); // 每个任务线程私有,不用加锁}}void push(const T& t){   Lock();_tasks.push(t);// 唤醒线程Wakeup();Unlock();}void start(){int num = defaultnum;for (int i = 0; i < num; i++){_threads[i].name = "thread-" + to_string(i + 1);pthread_create(&(_threads[i].tid), nullptr, HandlerTask, nullptr);}}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}private:vector<ThreadInfo> _threads;queue<T> _tasks;pthread_mutex_t _mutex;pthread_cond_t _cond;
};

此时我们再编译一下，发现又出错了。

为什么呢？这是因为我们的HandlerTask已经是一个静态成员函数，它是无法访问非静态成员函数的，因为静态成员函数没有对象的地址，无法调用，我们需要通过对象去调用静态成员函。

运行一下：

此时我们的用户没有给任务队列里面发送任何任务，所以此时线程都是被阻塞住的。

main.cpp

#include "threadpool.hpp"
#include "Task.hpp"
#include <ctime>int main()
{ThreadPool<Task>* tp = new ThreadPool<Task>(5); tp->start(); // 创建五个线程srand(time(nullptr));int len = opsum.size();while(true){// 1.构建任务int x = rand() % 10 + 1;usleep(10);int y = rand() % 5;char op = opsum[rand() % len];Task t(x, y, op);// 2.交给线程池处理tp->push(t);cout << "main thread make task: " << t.GetTask() << endl;sleep(1);}
}

PthreadPool.hpp

#include <iostream>
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>using namespace std;struct ThreadInfo
{pthread_t tid;string name;
};template <class T>
class ThreadPool
{static const int defaultnum = 5;
public:void Lock(){pthread_mutex_lock(&_mutex);}void Unlock(){pthread_mutex_unlock(&_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}void Sleep(){pthread_cond_wait(&_cond,&_mutex);}bool IsQueueEmpty(){return _tasks.empty();}string GetThreadName(pthread_t tid){for(auto e : _threads){if(tid == e.tid)return e.name;}return "";}public:ThreadPool(int num = defaultnum): _threads(num){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond, nullptr);}static void* HandlerTask(void* args){ThreadPool<T>* tp = static_cast<ThreadPool<T>*>(args);string name = tp->GetThreadName(pthread_self());while(true){tp->Lock();while(tp->IsQueueEmpty()){tp->Sleep();}T t = tp->Pop();tp->Unlock();t.run(); // 每个任务线程私有,不用加锁cout << name << " run, result: " << t.GetResult() << endl;}}T Pop(){T t = _tasks.front();_tasks.pop();return t;}void push(const T& t){   Lock();_tasks.push(t);// 唤醒线程Wakeup();Unlock();}void start(){int num = defaultnum;for (int i = 0; i < num; i++){_threads[i].name = "thread-" + to_string(i + 1);pthread_create(&(_threads[i].tid), nullptr, HandlerTask, this);}}~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}private:vector<ThreadInfo> _threads;queue<T> _tasks;pthread_mutex_t _mutex;pthread_cond_t _cond;
};

运行结果：

此时我们就能看到结果，由于线程是排队的，我们唤醒的时候也是按照顺序的，所以线程会依次被唤醒去执行任务，然后我们再看看C++的线程。

#include <iostream>
#include <thread>
#include <unistd.h>using namespace std;void run()
{while(true){sleep(1);cout << "hello thread!" << endl;}
}int main()
{thread t(run);t.join();return 0;
}

上面是c++的线程库的代码，然后我们来编译一下

此时编译器失败了，编译器提示没有定义pthread_cteate，这里我们会惊奇的发现这不就是我们Linux上刚刚使用的创建线程的接口，所以C++底层也是用了pthread库，我们编译的时候带上指定的库看看行不行。

我们会发现上面的C++使用线程库非常的简单，只需要定义一个类，然后调用成员方法即可，但是我们自己写的线程库使用起来比较繁琐，我们现在基于C++的方式对我们的线程池进行封装。

makefile

thread:main.cppg++ -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:rm -f thread

thread.hpp

#pragma onec#include <iostream>
#include <pthread.h>
#include <ctime>
#include <string>using namespace std;typedef void (*callback_t)(); // 函数指针
static int num = 1;class Thread
{ 
public:static void* Rontine(void* args){Thread* thread = static_cast<Thread*>(args);thread->Entery();return nullptr;}Thread(callback_t cb): _tid(0), _name(""), _start_timestamp(0), _isrunning(false), _cb(cb){}void Run(){_name = "thread-" + to_string(num++);_start_timestamp = time(nullptr);_isrunning = true;pthread_create(&_tid, nullptr, Rontine, this);}void Join(){pthread_join(_tid, nullptr);_isrunning = false;}string Name(){return _name;}uint64_t StartTimestamp(){return _start_timestamp;}bool IsRunning(){return _isrunning;}void Entery(){_cb();}~Thread(){}
private:pthread_t _tid;string _name;uint64_t _start_timestamp;bool _isrunning;callback_t _cb;
};

main.cpp

#include "thread.hpp"
#include <unistd.h>void Print()
{while(true){cout << "哈哈,我是一个封装的线程..." << endl;sleep(1);}
}int main()
{Thread t(Print);t.Run();t.Join();return 0;
}

然后我们来运行一下：

同时我们还可以根据我们上面设置的线程属性来看看线程是否符合预期

int main()
{Thread t(Print);t.Run();cout << "是否启动成功: " << t.IsRunning() << endl;cout << "启动的时间戳: " << t.StartTimestamp() << endl;cout << "线程的名字: " << t.Name() << endl;t.Join();return 0;
}

上面我们对线程进行了封装，那么此时我们就可以创建线程的对象，那么也就可以创建多个线程啦！

int main()
{vector<Thread> threads;for(int i = 0; i < 10; i++){threads.push_back(Thread(Print));}for(auto& e : threads){e.Run();}for(auto& e : threads){e.Join();}return 0;
}

此时就创建了一批线程。

四、线程安全的单例模式

什么是单例模式

单例模式是一种 "经典的, 常用的, 常考的" 设计模式.

什么是设计模式

IT行业这么火, 涌入的人很多. 俗话说林子大了啥鸟都有. 大佬和菜鸡们两极分化的越来越严重. 为了让菜鸡们不太拖大佬的后腿, 于是大佬们针对一些经典的常见的场景, 给定了一些对应的解决方案, 这个就是设计模式

单例模式的特点

某些类, 只应该具有一个对象(实例), 就称之为单例. 例如一个男人只能有一个女朋友. 在很多服务器开发场景中, 经常需要让服务器加载很多的数据 (上百G) 到内存中. 此时往往要用一个单例的类来管理这些数据.

饿汉实现方式和懒汉实现方式

吃完饭, 立刻洗碗, 这种就是饿汉方式. 因为下一顿吃的时候可以立刻拿着碗就能吃饭.

吃完饭, 先把碗放下, 然后下一顿饭用到这个碗了再洗碗, 就是懒汉方式.

懒汉方式最核心的思想是 "延时加载". 从而能够优化服务器的启动速度，理解下面的知识我们首先需要知道全局对象和局部对象加载的时间。

全局对象：
- 全局对象的构造发生在main()函数执行之前。这意味着在程序的任何部分开始执行用户定义的代码之前，所有全局对象都会被构造。这个过程按照对象定义的顺序进行。静态全局对象（即使用static关键字定义在全局作用域的对象）也遵循同样的规则。
局部对象：
- 局部对象的构造发生在定义它们的作用域内，当程序执行流首次到达该对象的定义处时。这意味着，如果一个局部对象定义在一个函数内，那么它会在每次函数被调用并且执行到该定义点时构造。
静态局部对象：
- 静态局部对象（在函数内部使用static关键字定义的对象）的构造时机比较特别。它们的构造在首次函数被调用时进行，且在整个程序运行期间只构造一次。这发生在该函数被调用之前，但仍然是在main()函数执行期间。静态局部对象的析构发生在程序结束之后，但早于全局对象的析构。

饿汉方式实现单例模式

template <typename T>
class Singleton
{static T data;public:static T *GetInstance(){return &data;}
};

只要通过 Singleton 这个包装类来使用 T 对象, 则一个进程中只有一个 T 对象的实例.

懒汉方式实现单例模式

template <typename T>
class Singleton
{static T *inst;public:static T *GetInstance(){if (inst == NULL){inst = new T();}return inst;}
};

存在一个严重的问题, 线程不安全. 第一次调用 GetInstance 的时候, 如果两个线程同时调用, 可能会创建出两份 T 对象的实例. 但是后续再次调用, 就没有问题了.

懒汉方式实现单例模式(线程池安全版本)

threadpool.hpp

#include <iostream>
#include <pthread.h>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>using namespace std;struct ThreadInfo
{pthread_t tid;string name;
};template <class T>
class ThreadPool
{static const int defaultnum = 5;
public:void Lock(){pthread_mutex_lock(&_mutex);}void Unlock(){pthread_mutex_unlock(&_mutex);}void Wakeup(){pthread_cond_signal(&_cond);}void Sleep(){pthread_cond_wait(&_cond,&_mutex);}bool IsQueueEmpty(){return _tasks.empty();}string GetThreadName(pthread_t tid){for(auto e : _threads){if(tid == e.tid)return e.name;}return "";}public:static void* HandlerTask(void* args){ThreadPool<T>* tp = static_cast<ThreadPool<T>*>(args);string name = tp->GetThreadName(pthread_self());while(true){tp->Lock();while(tp->IsQueueEmpty()){tp->Sleep();}T t = tp->Pop();tp->Unlock();t.run(); // 每个任务线程私有,不用加锁cout << name << " run, result: " << t.GetResult() << endl;}}T Pop(){T t = _tasks.front();_tasks.pop();return t;}void push(const T& t){   Lock();_tasks.push(t);// 唤醒线程Wakeup();Unlock();}void start(){int num = defaultnum;for (int i = 0; i < num; i++){_threads[i].name = "thread-" + to_string(i + 1);pthread_create(&(_threads[i].tid), nullptr, HandlerTask, this);}} static ThreadPool<T>* GetInstance(){if(_tp == nullptr){cout << "log singleton create done first!" << endl;_tp = new ThreadPool<T>;}return _tp;}
private:ThreadPool(int num = defaultnum): _threads(num){pthread_mutex_init(&_mutex, nullptr);pthread_cond_init(&_cond, nullptr);}ThreadPool(const ThreadPool<T>& t) = delete;const ThreadPool<T>& operator=(const ThreadPool<T>& t) = delete;~ThreadPool(){pthread_mutex_destroy(&_mutex);pthread_cond_destroy(&_cond);}
private:vector<ThreadInfo> _threads;queue<T> _tasks;pthread_mutex_t _mutex;pthread_cond_t _cond;static  ThreadPool<T>* _tp; //懒汉
};template <class T>
ThreadPool<T>* ThreadPool<T>::_tp = nullptr; //类外面定义

main.cc

#include "threadpool.hpp"
#include "Task.hpp"
#include <ctime>int main()
{//ThreadPool<Task>* tp = new ThreadPool<Task>(5); //tp->start(); // 创建五个线程cout << "process running..." << endl;ThreadPool<Task>::GetInstance()->start();srand(time(nullptr));int len = opsum.size();while(true){// 1.构建任务int x = rand() % 10 + 1;usleep(10);int y = rand() % 5;char op = opsum[rand() % len];Task t(x, y, op);// 2.交给线程池处理ThreadPool<Task>::GetInstance()->push(t);cout << "main thread make task: " << t.GetTask() << endl;sleep(1);}
}

上面的验证我们确实只有一个单例对象，可是我们在获取单例对象的时候，也是多线程获取的呢？多个线程调用GetInstance呢？此时并发访问就会出现线程安全的问题，解决就要加锁。

但是我们会发现一个问题，除了第一次调用GetInstance会实例化对象，后面的线程访问的时候都要先加锁，判断，然后再解锁，最后返回，其实我们会发现后面的判断总会失败，所以我们这里可以再设计一下。

注意事项: 1. 加锁解锁的位置 2. 双重 if 判定, 避免不必要的锁竞争

五、STL,智能指针和线程安全

STL中的容器是否是线程安全的?

不是. 原因是, STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响. 而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶). 因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全.

智能指针是否是线程安全的?

对于 unique_ptr, 由于只是在当前代码块范围内生效, 因此不涉及线程安全问题.

对于 shared_ptr, 多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数.

六、其他常见的各种锁

悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
自旋锁，公平锁，非公平锁

自旋锁

概念：自旋锁是一种简单的同步原语，当一个线程尝试获取一个已经被其他线程持有的锁时，该线程不会立即放弃CPU的执行权限，而是不断地循环检查（自旋）锁的状态，直到锁变为可用状态。这种方式适用于锁被持有的时间非常短的情况，可以减少线程上下文切换的开销。

挂起等待锁（也称为阻塞锁）

概念：挂起等待锁，当一个线程尝试获取锁失败时，不是通过自旋等待，而是将该线程置于等待（或阻塞）状态，并让出CPU给其他线程使用。线程被操作系统挂起，直到锁变为可用状态时，由操作系统唤醒该线程，重新参与竞争锁。

区别

资源利用：自旋锁在等待时会持续占用CPU，而挂起等待锁则不会，它会让出CPU给其他任务执行。
适用场景：自旋锁适用于锁保护的临界区执行时间很短的情况；挂起等待锁适用于锁可能被长时间持有的情况。
性能影响：自旋锁在锁很快就释放的场景下性能较好，因为避免了线程上下文切换的开销；而挂起等待锁虽然增加了上下文切换的开销，但在锁持有时间较长时能更有效地利用系统资源。

七、读者写者问题

读写锁介绍

在编写多线程的时候，有一种情况是十分常见的。那就是，有些公共数据修改的机会比较少。相比较改写，它们读的机会反而高的多。通常而言，在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。那么有没有一种方法，可以专门处理这种多读少写的情况呢？有，那就是读写锁。

⭐注意：写独占，读共享，读锁优先级高

读写锁接口

设置读写优先

int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
/*
pref 共有 3 种选择
PTHREAD_RWLOCK_PREFER_READER_NP (默认设置) 读者优先，可能会导致写者饥饿情况
PTHREAD_RWLOCK_PREFER_WRITER_NP 写者优先，目前有 BUG，导致表现行为和
PTHREAD_RWLOCK_PREFER_READER_NP 一致
PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP 写者优先，但写者不能递归加锁
*/

初始化

int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
const pthread_rwlockattr_t*restrict attr);

加锁

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

解锁

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);