目录
1.生产者消费者模型;
2.基于环形队列的生产者消费者模型;
3.线程池;
4.STL, 智能指针, 线程安全;
5.读者写者问题.
前言:
本篇博客博主五一假期都在肝的一篇, 希望xdm点点三连, 博主感谢了 onz !!!
1.生产者消费者模型
321原则:(便于记忆)
3是指3种关系: 生产者和生产者; 消费者和消费者;生产者和消费者.
2是2种角色: 生产者和消费者;
1是指一个交易场所: 内存空间.
1.1为什么要使用生产者消费者模型?
生产者和消费者模型是通过一个容器来解决生产者和消费者强解耦问题, 生产者和消费者通过阻塞队列进行通讯, 阻塞队列就相当于一个缓冲区, 消费者是从阻塞队列里面拿数据, 生产者生产的数据不用给消费者,直接给到阻塞队列里面, 这样可以平衡生产者和消费者的能力.(忙闲不均)
1.2生产者消费者模型的优点
(1)解耦; (2)支持并发; (3)支持忙闲不均.
1.3基于阻塞队列的生产消费模型
阻塞队列当队列为空的时候, 消费者无法取数据而是被阻塞,直到生产新数据.当队列为满的时,生产者也不会再生产数据而是被阻塞, 直到消费数据.
1.4代码
这里封装四个模块进行编写.对每个模块进行详细解答.
首先的是BlockQueue.hpp, 封装一个阻塞队列.
条件变量_p_cond和_c_cond都是用来阻塞队列生产消费关系处理的.
#pragma once
#include<iostream>
#include<queue>
#include"LockGuard.hpp"
using namespace std;
const int defaultcap = 5;
template<class T>
class BlockQueue
{
public:
BlockQueue(int cap = defaultcap)
:_capacity(cap)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_p_cond, nullptr);
pthread_cond_init(&_c_cond, nullptr);
}
bool IsFull()
{
return _q.size() == _capacity;
}
bool IsEmpty()
{
return _q.size() == 0;
}
void Push(const T& in)//生产者生产数据
{
LockGuard lockguard(&_mutex);
while(IsFull())
{
pthread_cond_wait(&_p_cond, &_mutex);
}
_q.push(in);
pthread_cond_signal(&_c_cond);
}
void Pop(T* out)//消费者消费数据
{
LockGuard lockguard(&_mutex);
while(IsEmpty())
{
pthread_cond_wait(&_c_cond, &_mutex);
}
*out = _q.front();
_q.pop();
pthread_cond_signal(&_p_cond);
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_p_cond);
pthread_cond_destroy(&_c_cond);
}
private:
queue<T> _q;
int _capacity;
pthread_mutex_t _mutex;
pthread_cond_t _p_cond; //生产者的条件变量
pthread_cond_t _c_cond; /* 消费者的条件变量 */
};上面BlockQueue还需要封装了一个锁LockGuard;
这里就是封装了一个互斥量(锁)
#pragma once
#include<pthread.h>
class Mutex
{
public:
Mutex(pthread_mutex_t* lock)
:_lock(lock)
{}
void Lock()
{
pthread_mutex_lock(_lock);
}
void unlock()
{
pthread_mutex_unlock(_lock);
}
~ Mutex()
{}
private:
pthread_mutex_t* _lock;
};
class LockGuard
{
public:
LockGuard(pthread_mutex_t* lock)
:_mutex(lock)
{
_mutex.Lock();
}
~LockGuard()
{
_mutex.unlock();
}
private:
Mutex _mutex;
};生产者生产数据和消费者消费数据的数据, 这里我们Task.hpp来实现+-*/%的处理类.
#pragma once
#include<iostream>
#include<string>
#include<unistd.h>
using namespace std;
const int defaultvalue = 0;
const string opers = "+-*/%)(|";
enum
{
ok = 0,
div_zero,
mod_zero,
unkown
};
class Task
{
public:
Task()
{}
Task(int x, int y, char op)
:data_x(x)
,data_y(y)
,oper(op)
,result(defaultvalue)
,code(ok)
{}
void Run()
{
switch(oper)
{
case'+':
result = data_x + data_y;
break;
case'-':
result = data_x - data_y;
break;
case'*':
result = data_x * data_y;
break;
case'/':
{
if(data_y == 0)
code = div_zero;
else
result = data_x / data_y;
}
break;
case'%':
{
if(data_y == 0)
code = mod_zero;
else
result = data_x % data_y;
}
break;
default:
code = unkown;
break;
}
}
void operator()()
{
Run();
sleep(2);
}
string PrintTask()
{
string s;
s = to_string(data_x);
s += oper;
s += to_string(data_y);
s += "=?";
return s;
}
string PrintResult()
{
string s;
s = to_string(data_x);
s += oper;
s += to_string(data_y);
s += "=";
s += to_string(result);
s += "[";
s += to_string(code);
s += "]";
return s;
}
~Task()
{}
private:
int data_x;
int data_y;
char oper; //+-/*%
int result;
int code; /* 结果码 0: 可信, !0 不可信 */
};最后一个main.cc进行测试;
这里p表示生产者, c表示消费者; ThreadData是封装了阻塞队列以及线程名称.
消费者对于数据是消费那么就是pop数据, 并且还要的到数据结果; 生产者是push数据, 并且还要更新数据的信息.
#include<iostream>
#include<pthread.h>
#include<time.h>
#include<unistd.h>
#include"Blockqueue.hpp"
#include"Task.hpp"
using namespace std;
class ThreadData
{
public:
BlockQueue<Task>* bq;
string name;
};
void* consumer(void* args)
{
ThreadData* td = (ThreadData*)args;
while(true)
{
Task t;
td->bq->Pop(&t);
t.Run();
cout << "consumer data:" << t.PrintResult() << "," << td->name << endl;
}
return nullptr;
}
void* productor(void* args)
{
BlockQueue<Task>* bq = static_cast<BlockQueue<Task>*>(args);
while(true)
{
int data1 = rand() % 10;
usleep(rand()%123);
int data2 = rand() % 10;
usleep(rand()%123);
char oper = opers[rand() % opers.size()];
Task t(data1, data2, oper);
cout << "productor task:" << t.PrintTask() << endl;
bq->Push(t);
sleep(1);
}
return nullptr;
}
int main()
{
/* 形成随机数 */
srand((uint16_t)time(nullptr) ^ getpid() ^ pthread_self());
BlockQueue<Task>* bq = new BlockQueue<Task>();
pthread_t c[3], p[2];/* consumer productor */
ThreadData* td = new ThreadData();
td->bq = bq;
td->name = "thread-1";
/* 消费者1 */
pthread_create(&c[0], nullptr, consumer, td);
ThreadData* td1 = new ThreadData();
td1->bq = bq;
td1->name = "thread-2";
/* 消费者2 */
pthread_create(&c[1], nullptr, consumer, td1);
ThreadData* td2 = new ThreadData();
td2->bq = bq;
td2->name = "thread-3";
/* 消费者3 */
pthread_create(&c[2], nullptr, consumer, td2);
pthread_create(&p[0], nullptr, productor, bq);
pthread_join(c[0], nullptr);
pthread_join(c[1], nullptr);
pthread_join(c[2], nullptr);
pthread_join(p[0], nullptr);
pthread_join(p[1], nullptr);
return 0;
}2.基于环形队列的生产者消费者模型
先穿插讲一个日志, 如何写日志.
va_list: 宏定义, 是一个指向变长参数列表的指针.
va-start: 可变参数的起始地址, ap是参数va_list, last是最后固定参数的地址.
va_end: 结束使用变长参数列表, 用于清空va_list和va_start
#pragma once
#include<iostream>
#include<stdarg.h>
#include<time.h>
using namespace std;
enum
{
Debug = 0,
Info,
warning,
Error,
Fatal
};
string LevelToString(int level)
{
switch(level)
{
case Debug:
return "Debug";
case Info:
return "Info";
case warning:
return "warning";
case Error:
return "Error";
case Fatal:
return "Fatal";
default:
return "unkown";
}
}
class Log
{
public:
Log(){}
void LogMessage(int level, const char* format, ...)
{
char content[1024];
va_list args;//可变参数
va_start(args, format);
vsnprintf(content, sizeof(content), format, args);
va_end(args);
uint64_t currtime = time(nullptr);
printf("[%s][%s]%s\n", LevelToString(level).c_str(),
to_string(currtime).c_str(), content);
}
~Log(){}
private:
};#include<iostream>
#include"Log.hpp"
#include"RingQueue.hpp"
#include"Task.hpp"
#include<pthread.h>
#include<ctime>
using namespace std;
void* Productor(void* args)
{
RingQueue<Task>* rq = static_cast<RingQueue<Task>*>(args);
while(true)
{
int data1 = rand() % 10;
usleep(rand() % 123);
int data2 = rand() % 10;
usleep(rand() % 123);
char oper = opers[rand() % (opers.size())];
Task t(data1, data2, oper);
cout << "productor task: " << t.PrintTask() << endl;
rq->Push(t);
}
return nullptr;
}
void* consumer(void* args)
{
RingQueue<Task>* rq = static_cast<RingQueue<Task>*>(args);
while(true)
{
Task t;
rq->Pop(&t);
t.Run();
cout << "consumer done, data is : " << t.PrintResult() << endl;
}
}
int main()
{
/* Log log;
log.LogMessage(Debug, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(warning, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(Error, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(Info, "hello %d, %s, %f", 10, "study", 3.14);
*/
srand((uint64_t)time(nullptr) ^ pthread_self());
pthread_t c[3], p[2];
RingQueue<Task>* rq = new RingQueue<Task>();
pthread_create(&p[0], nullptr, Productor, rq);
pthread_create(&p[1], nullptr, Productor, rq);
pthread_create(&c[0], nullptr, consumer, rq);
pthread_create(&c[1], nullptr, consumer, rq);
pthread_create(&c[2], nullptr, consumer, rq);
pthread_join(p[0], nullptr);
pthread_join(p[1], nullptr);
pthread_join(c[0], nullptr);
pthread_join(c[1], nullptr);
pthread_join(c[2], nullptr);
return 0;
}RingQueue是采用信号量进行条件变量以及互斥量的控制.
首先认识一下信号量的接口:
int sem_wait(sem_t* sem): 相当于条件变量里面的pthread_cond_wait
int sem_post(sem_t* sem): 相当于mutex_unlock;
int sem_init(sem* sem): 相当于pthread_cond_init;
int sem_destroy(sem* sem): 相当于pthread_cond_destroy
下图更好的理解环形队列生产消费模型;
生产者不能将消费者包个圈, 因为消费者没有消费, 那么空间资源就不会释放, 并且消费者也不能越过生产者, 因为消费者消费完生产者生产的数据, 那么还需要的数据并还没有生产.
这样就有两个变量需要控制一个是空间资源_space_sem, _data_sem; 以及生产者消费者的位置也需要记录_p_step, _c_step.
#pragma once
#include <iostream>
#include <vector>
#include <pthread.h>
#include <semaphore.h>
#include "LockGuard.hpp"
const int defaultsize = 5;
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 size = defaultsize)
: _ringqueue(size), _size(size), _p_step(0), _c_step(0)
{
sem_init(&_space_sem, 0, size);
sem_init(&_data_sem, 0, 0);
pthread_mutex_init(&_p_mutex, nullptr);
pthread_mutex_init(&_c_mutex, nullptr);
}
void Push(const T &in)
{
P(_space_sem);
{
LockGuard lockGuard(&_p_mutex);
_ringqueue[_p_step] = in;
_p_step++;
_p_step %= _size;
}
V(_data_sem);
}
void Pop(T *out)
{
// 消费
P(_data_sem);
{
LockGuard lockGuard(&_c_mutex);
*out = _ringqueue[_c_step];
_c_step++;
_c_step %= _size;
}
V(_space_sem);
}
~RingQueue()
{
sem_destroy(&_space_sem);
sem_destroy(&_data_sem);
pthread_mutex_destroy(&_p_mutex);
pthread_mutex_destroy(&_c_mutex);
}
private:
std::vector<T> _ringqueue;
int _size;
int _p_step; // 生产者的生产位置
int _c_step; // 消费位置
sem_t _space_sem; // 生产者
sem_t _data_sem; // 消费者
pthread_mutex_t _p_mutex;
pthread_mutex_t _c_mutex;
};常见生产者消费者线程, 借助上次使用过的Task类, 模拟生产这个数据, 然后进行push数据.
消费者就是popTask里面的数据结果.
注意: 创建的线程一定要join.
#include<iostream>
#include"Log.hpp"
#include"RingQueue.hpp"
#include"Task.hpp"
#include<pthread.h>
#include<ctime>
using namespace std;
void* Productor(void* args)
{
RingQueue<Task>* rq = static_cast<RingQueue<Task>*>(args);
while(true)
{
int data1 = rand() % 10;
usleep(rand() % 123);
int data2 = rand() % 10;
usleep(rand() % 123);
char oper = opers[rand() % (opers.size())];
Task t(data1, data2, oper);
cout << "productor task: " << t.PrintTask() << endl;
rq->Push(t);
}
return nullptr;
}
void* consumer(void* args)
{
RingQueue<Task>* rq = static_cast<RingQueue<Task>*>(args);
while(true)
{
Task t;
rq->Pop(&t);
t.Run();
cout << "consumer done, data is : " << t.PrintResult() << endl;
}
}
int main()
{
/* Log log;
log.LogMessage(Debug, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(warning, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(Error, "hello %d, %s, %f", 10, "study", 3.14);
log.LogMessage(Info, "hello %d, %s, %f", 10, "study", 3.14);
*/
srand((uint64_t)time(nullptr) ^ pthread_self());
pthread_t c[3], p[2];
RingQueue<Task>* rq = new RingQueue<Task>();
pthread_create(&p[0], nullptr, Productor, rq);
pthread_create(&p[1], nullptr, Productor, rq);
pthread_create(&c[0], nullptr, consumer, rq);
pthread_create(&c[1], nullptr, consumer, rq);
pthread_create(&c[2], nullptr, consumer, rq);
pthread_join(p[0], nullptr);
pthread_join(p[1], nullptr);
pthread_join(c[0], nullptr);
pthread_join(c[1], nullptr);
pthread_join(c[2], nullptr);
return 0;
}3.线程池
3.1 线程池的概念:
线程池是一种线程使用的机制, 维护多个线程, 等待命令执行多线程并发的任务, 避免过多线程的调用带来太大开销, 导致影响性能. 好处是避免短时间的任务创建和销毁的开销, 保证内核的高效使用.
3.2线程池使用的场景:
(1)需要大量线程完成, 并且任务时间短的;
(2)对性能要求比较高的应用;
(3)接受突发性大量请求;
3.3线程池的种类:
(1)创建固定数量的线程,池, 循环从队列里面拿任务;
(2)获取到任务之后, 执行任务对象的任务接口;
3.4代码编写(单例模式的线程池)
#include <iostream>
#include <queue>
#include <vector>
#include <pthread.h>
#include <functional>
#include"Thread.hpp"
#include"LockGuard.hpp"
#include"Log.hpp"
#include"Task.hpp"
using namespace std;
class ThreadData
{
public:
ThreadData(const string& name)
:threadname(name)
{}
~ThreadData()
{}
string threadname;
};
static const int defaultnum = 5;
template<class T>
class ThreadPool
{
private:
ThreadPool(int thread_num = defaultnum)
:_thread_num(thread_num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for(int i = 0; i < _thread_num; i++)
{
string threadname = "thread-";
threadname += to_string(i + 1);
ThreadData td(threadname);
_threads.emplace_back(threadname, bind(&ThreadPool<T>::ThreadRun, this, placeholders::_1), td);
log.LogMessages(Info, "%s is create...\n", threadname.c_str());
}
}
ThreadPool(const ThreadPool<T>& td) = delete;
ThreadPool<T>& operator=(const ThreadPool<T>& td) = delete;
public:
static ThreadPool<T>* GetInstance()
{
if(instance == nullptr)
{
LockGuard lockguard(&sig_lock);
if(instance == nullptr)
{
log.LogMessages(Info, "创建单例成功!");
instance = new ThreadPool<T>();
}
}
return instance;
}
bool start()/* 启动线程池 */
{
for(auto& thread : _threads)
{
thread.Start();
log.LogMessages(Info, "%s is running ...\n", thread.ThreadName().c_str());
}
return true;
}
void ThreadWait(const ThreadData& td)
{
log.LogMessages(Debug, "no task, %s is sleeping...\n", td.threadname.c_str());
pthread_cond_wait(&_cond, &_mutex);
}
void ThreadWakeup()
{
pthread_cond_signal(&_cond);
}
void checkSelf()
{
/* 水位线检查 */
}
void ThreadRun(ThreadData& td) /* 模拟pop消费者消费数据 */
{
while(true)
{
checkSelf();
T t;
{
LockGuard lockguard(&_mutex);
while(_q.empty())
{
ThreadWait(td);
log.LogMessages(Debug, "thread %s is wakeup\n", td.threadname.c_str());
}
t = _q.front();
_q.pop();
}
Task t1;
log.LogMessages(Debug, "%s handler task %s done, result is : %s\n",
td.threadname, t1.PrintTask().c_str(), t1.PrintResult().c_str());
}
}
void Push(Task& in)
{
log.LogMessages(Debug, "other thread push a task, task is :%s \n", in.PrintTask().c_str());
LockGuard lockguard(&_mutex);
_q.push(in);
ThreadWakeup();
}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
void wait()
{
for(auto& thread : _threads)
{
thread.Join();
}
}
private:
queue<T> _q;
vector<Thread<ThreadData>> _threads;
int _thread_num;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
static ThreadPool<T>* instance;
static pthread_mutex_t sig_lock;
};
/* 对static变量外部初始化 */
template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;
template <class T>
pthread_mutex_t ThreadPool<T>::sig_lock = PTHREAD_MUTEX_INITIALIZER;#include <iostream>
#include <memory>
#include <ctime>
#include "ThreadPool.hpp"
#include "Task.hpp"
using namespace std;
int main()
{
sleep(1);
ThreadPool<Task>::GetInstance()->start();
srand((uint64_t)time(nullptr) ^ getpid());
while(true)
{
int x = rand() % 10;
usleep(100);
int y = rand() % 10;
usleep(100);
char oper = opers[rand() % opers.size()];
Task t1(x, y, oper);
ThreadPool<Task>::GetInstance()->Push(t1);
sleep(1);
}
ThreadPool<Task>::GetInstance()->wait();
return 0;
}4. STL, 智能指针, 线程安全
(1) STL不是线程安全的, 如果要在线程里面使用需要配合锁使用.
(2) 智能指针中unique_ptr, 不是线程安全的; shared_ptr因为是涉及到引用计数所以是原子的,也就是线程安全的.
5.读者写者问题
在编写多线程的时候, 往往会出现修改的机会少, 读取的机会多, 在读取的时候互斥量的一系列处理就效率非常慢, 然后根据这种情况特定还有一种读写锁.
这里先认识一下读写锁的接口:
int pthread_rwlock_init: 读写锁的初始化
int pthread_rwlock_destroy: 读写锁的销毁
int pthread_rwlock_rdlock: 读锁的互斥量
int pthread_rwlock_wrlock: 写锁的互斥量
int pthread_rwlock_unlock: 读写锁的解锁.
