本章主要讲解学习Linux线程章节的后一部分,主要介绍生产消费者模型以及线程池等等的学习
交易产所是被多个生产者和消费者共同所见的,即被多个执行流同时访问,为了避免交易产所中的数据混乱,我们需要将该临界资源用互斥锁保护起来。其中,所有的生产者和消费者都会竞争式的申请锁,因此生产者和生产者、消费者和消费者、生产者和消费者之间都存在互斥关系
注:互斥关系保证的是数据的访问正常,而同步关系是为了让多线程(生产和消费者)之间协同起来
main.cc:
#include "BlockQueue.hpp"
#include<cstdio>
#include<ctime>
#include<unistd.h>
#define NUM 5
void* consumer(void* c)
{
blockqueue<int>* bq=(blockqueue<int>*)c;
while(true)
{
//取出数据
int t=0;
bq->Get(&t);
std::cout<<"consumer["<<pthread_self()<<"]:生产数据->"<<t<<std::endl;
}
return nullptr;
}
void* producter(void* p)
{
srand((unsigned)time(nullptr));
blockqueue<int>* bq=(blockqueue<int>*)p;
while(true)
{
//生成数据和放入数据
int t=rand()%100+1;
bq->Put(t);
std::cout<<"producter["<<pthread_self()<<"]:消费数据->"<<t<<std::endl;
sleep(1);
}
return nullptr;
}
int main()
{
blockqueue<int>* bq=new blockqueue<int>(NUM);
pthread_t cid,pid;
pthread_create(&cid,nullptr,consumer,(void*)bq);
pthread_create(&pid,nullptr,producter,(void*)bq);
pthread_join(cid,nullptr);
pthread_join(pid,nullptr);
delete bq;
}
BlockQueue.hpp:
#include <iostream>
#include <queue>
#include <pthread.h>
template<class T>
class blockqueue
{
private:
int _cap;//容量
std::queue<T> _q;//交易场所-临界资源
//同步与互斥
pthread_cond_t _data;
pthread_cond_t _space;
pthread_mutex_t _lock;
public:
blockqueue(int cap):_cap(cap)
{
pthread_cond_init(&_data,nullptr);
pthread_cond_init(&_space,nullptr);
pthread_mutex_init(&_lock,nullptr);
}
bool IsFull()
{
return _q.size()==_cap;
}
bool IsEmpty()
{
return _q.size()==0;
}
//获取数据
void Get(T* out)
{
pthread_mutex_lock(&_lock);
while(IsEmpty())//唤醒后还需检测
{
pthread_cond_wait(&_data,&_lock);//同步
}
//有数据
*out=_q.front();
_q.pop();
if(_q.size()<=_cap/2)//空间过半
pthread_cond_signal(&_space);//唤醒该条件变量下的线程
pthread_mutex_unlock(&_lock);
}
//放数据
void Put(const T& in)
{
pthread_mutex_lock(&_lock);
while(IsFull())
{
pthread_cond_wait(&_space,&_lock);
}
_q.push(in);
if(_q.size()>=_cap/2)//数据过半
pthread_cond_signal(&_data);//有数据,唤醒该条件变量下的等待线程
pthread_mutex_unlock(&_lock);
}
~blockqueue()
{
pthread_cond_destroy(&_data);
pthread_cond_destroy(&_space);
pthread_mutex_destroy(&_lock);
}
};
pthread_cond_wait
函数时就需要传入当前线程手中的互斥锁,此时当该线程被挂起时就会自动释放手中的互斥锁,而当该线程被唤醒时又会自动获取到该互斥锁
main.cc:
#include "BlockQueue.hpp"
#include<cstdio>
#include<ctime>
#include<unistd.h>
#define NUM 10
void* consumer(void* c)
{
blockqueue<int>* bq=(blockqueue<int>*)c;
while(true)
{
//取出数据
int t=0;
bq->Get(&t);
std::cout<<"consumer["<<pthread_self()<<"]:生产数据->"<<t<<std::endl;
}
return nullptr;
}
void* producter(void* p)
{
srand((unsigned)time(nullptr));
blockqueue<int>* bq=(blockqueue<int>*)p;
while(true)
{
//生成数据和放入数据
int t=rand()%100+1;
bq->Put(t);
std::cout<<"producter["<<pthread_self()<<"]:消费数据->"<<t<<std::endl;
sleep(1);
}
return nullptr;
}
int main()
{
blockqueue<int>* bq=new blockqueue<int>(NUM);
pthread_t cid1,cid2,cid3,cid4,pid1,pid2;
pthread_create(&cid1,nullptr,consumer,(void*)bq);
pthread_create(&cid2,nullptr,consumer,(void*)bq);
pthread_create(&cid3,nullptr,consumer,(void*)bq);
pthread_create(&cid4,nullptr,consumer,(void*)bq);
pthread_create(&pid1,nullptr,producter,(void*)bq);
pthread_create(&pid2,nullptr,producter,(void*)bq);
pthread_join(cid1,nullptr);
pthread_join(cid2,nullptr);
pthread_join(cid3,nullptr);
pthread_join(cid4,nullptr);
pthread_join(pid1,nullptr);
pthread_join(pid2,nullptr);
delete bq;
}
BlockQueue.hpp:
#include <iostream>
#include <queue>
#include <pthread.h>
template<class T>
class blockqueue
{
private:
int _cap;//容量
std::queue<T> _q;//交易场所-临界资源
//同步与互斥
pthread_cond_t _data;
pthread_cond_t _space;
pthread_mutex_t _lock;//生产-消费的互斥
pthread_mutex_t c_lock;//消费之间互斥
pthread_mutex_t p_lock;//生产之间互斥
public:
blockqueue(int cap):_cap(cap)
{
pthread_cond_init(&_data,nullptr);
pthread_cond_init(&_space,nullptr);
pthread_mutex_init(&_lock,nullptr);
pthread_mutex_init(&c_lock,nullptr);
pthread_mutex_init(&p_lock,nullptr);
}
bool IsFull()
{
return _q.size()==_cap;
}
bool IsEmpty()
{
return _q.size()==0;
}
//获取数据
void Get(T* out)
{
pthread_mutex_lock(&c_lock);
pthread_mutex_lock(&_lock);
while(IsEmpty())//唤醒后还需检测
{
pthread_cond_wait(&_data,&_lock);//同步
}
//有数据
*out=_q.front();
_q.pop();
if(_q.size()<=_cap/2)//空间过半
pthread_cond_signal(&_space);//唤醒该条件变量下的线程
pthread_mutex_unlock(&_lock);
pthread_mutex_unlock(&c_lock);
}
//放数据
void Put(const T& in)
{
pthread_mutex_lock(&p_lock);
pthread_mutex_lock(&_lock);
while(IsFull())
{
pthread_cond_wait(&_space,&_lock);
}
_q.push(in);
if(_q.size()>=_cap/2)//数据过半
pthread_cond_signal(&_data);//有数据,唤醒该条件变量下的等待线程
pthread_mutex_unlock(&_lock);
pthread_mutex_unlock(&p_lock);
}
~blockqueue()
{
pthread_cond_destroy(&_data);
pthread_cond_destroy(&_space);
pthread_mutex_destroy(&_lock);
pthread_mutex_destroy(&c_lock);
pthread_mutex_destroy(&p_lock);
}
};
注:当然可以只使用一把互斥锁,这里为了更好表示生产与消费者们之间的关系使用了三个互斥锁
环形队列采用数组模拟,用模运算来模拟环状特性
RingQueue.hpp:
#include<iostream>
#include<semaphore.h>
#include<pthread.h>
#include<vector>
#define NUM 10
template<class T>
class RingQueue
{
private:
int _cap;//容量
std::vector<T> _rq;//临界资源
//保证消费者和生产者之间的同步与互斥关系
sem_t _space;//描述空间资源
sem_t _data;//描述数据资源
//记录下标位置-同样也是一种临界资源
int c_index;
int p_index;
//保证多线程下消费者之间和生产者之间的互斥关系(使用的下标是临界资源,并且对下标的操作不是原子的)
pthread_mutex_t c_lock;
pthread_mutex_t p_lock;
public:
RingQueue(int cap=NUM):_cap(cap),_rq(cap)
{
sem_init(&_space,0,_cap);
sem_init(&_data,0,0);
pthread_mutex_init(&c_lock,nullptr);
pthread_mutex_init(&p_lock,nullptr);
}
~RingQueue()
{
sem_destroy(&_space);
sem_destroy(&_data);
pthread_mutex_destroy(&c_lock);
pthread_mutex_destroy(&p_lock);
}
void Put(const T& in)
{
sem_wait(&_space);//申请空间资源-预定
pthread_mutex_lock(&c_lock);//生产者之间竞争互斥锁-排队
_rq[c_index++]=in;
c_index%=_cap;
sem_post(&_data);//发布数据资源
pthread_mutex_unlock(&c_lock);//解锁
}
void Get(T* out)
{
sem_wait(&_data);//申请数据资源
pthread_mutex_lock(&p_lock);//消费者竞争互斥锁-排队
*out=_rq[p_index++];
p_index%=_cap;
sem_post(&_space);//发布空间资源
pthread_mutex_unlock(&p_lock);//解锁
}
};
main.cc:
#include "RingQueue.hpp"
#include<cstdio>
#include<ctime>
#include<unistd.h>
class Task
{
private:
int _cnt;
public:
Task(){}
Task(int cnt):_cnt(cnt){}
int Run()
{
int sum=0;
for(int i=1;i<=_cnt;i++)
sum+=i;
return sum;
}
void Show()
{
printf("Thread:%p is running...Task done: counting from 1 to %d",pthread_self(),_cnt);
}
};
void* consumer(void* c)
{
RingQueue<Task>* rq=(RingQueue<Task>*)c;
while(true)
{
//取出数据+计算结果
Task t;
rq->Get(&t);
int ret=t.Run();
std::cout<<"consumer->";
t.Show();
printf(" ret=%d\n",ret);
sleep(1);
}
return nullptr;
}
void* producter(void* p)
{
srand((unsigned)time(nullptr));
RingQueue<Task>* rq=(RingQueue<Task>*) p;
while(true)
{
//生成数据和放入数据
int t=rand()%100+1;
rq->Put(Task(t));
std::cout<<"producter->Thread:"<<pthread_self()<<" is running...Task made: count from 1 to "<<t<<std::endl;
//sleep(1);
}
return nullptr;
}
int main()
{
RingQueue<int>* rq=new RingQueue<int>(NUM);
pthread_t cid1,cid2,cid3,cid4,pid1,pid2;
pthread_create(&cid1,nullptr,consumer,(void*)rq);
pthread_create(&cid2,nullptr,consumer,(void*)rq);
pthread_create(&cid3,nullptr,consumer,(void*)rq);
pthread_create(&cid4,nullptr,consumer,(void*)rq);
pthread_create(&pid1,nullptr,producter,(void*)rq);
pthread_create(&pid2,nullptr,producter,(void*)rq);
pthread_join(cid1,nullptr);
pthread_join(cid2,nullptr);
pthread_join(cid3,nullptr);
pthread_join(cid4,nullptr);
pthread_join(pid1,nullptr);
pthread_join(pid2,nullptr);
delete rq;
}
示例:WEB服务器完成网页请求这样的任务,使用线程池技术是非常合适的,因为单个任务小,而任务数量巨大;但对于长时间的任务,比如一个Telnet连接请求,线程池的优点就不明显了,因为Telnet会话时间比线程的创建时间大多了
示例:要求服务器迅速响应客户请求接受突发性的大量请求,突发性大量客户请求,在没有线程池情况下,将产生大量线程,虽然理论上大部分操作系统线程数目最大值不是问题,短时间内产生大量线程可能使内存到达极限,出现错误
ThreadPool.hpp:
#pragma once
#include<pthread.h>
#include<iostream>
#include<queue>
#define NUM 10
template<class T>
class ThreadPool
{
private:
int _num;//线程数量
std::queue<T> task_queue;//任务阻塞队列->临界资源
//线程同步与互斥
pthread_mutex_t _lock;
pthread_cond_t _cond;
public:
//封装接口便于传入的对象调用
void Lock()
{
pthread_mutex_lock(&_lock);
}
void UnLock()
{
pthread_mutex_unlock(&_lock);
}
void CondWait()
{
pthread_cond_wait(&_cond,&_lock);
}
void CondWake()
{
pthread_cond_signal(&_cond);
}
bool IsEmpty()
{
return task_queue.size()==0;
}
public:
ThreadPool(int num):_num(num)
{
pthread_mutex_init(&_lock,nullptr);
pthread_cond_init(&_cond,nullptr);
}
~ThreadPool()
{
pthread_mutex_destroy(&_lock);
pthread_cond_destroy(&_cond);
}
void InitThreadPool()
{
pthread_t tid;
for(int i=0;i < _num;i++)//创建一批线程
{
pthread_create(&tid,nullptr,Routine,this);//传入线程池地址
pthread_detach(tid);//线程分离
}
}
//线程执行例程
static void* Routine(void* arg)
{
ThreadPool* tp=(ThreadPool*)arg;
//检测是否有任务-没任务等待;有任务进行获取并执行
while(true)
{
//检查
while(tp->IsEmpty())
tp->CondWait();
//取出任务
T task;
tp->Get(&task);
task();//执行逻辑-仿函数
}
}
void Put(const T& t)
{
Lock();
task_queue.push(t);
CondWake();//有任务进行唤醒
UnLock();
}
void Get(T* t)
{
//取任务已经是获取到锁资源了
*t=task_queue.front();
task_queue.pop();
}
};
main.cc:
#include "ThreadPool.hpp"
#include<cstdio>
#include<ctime>
#include<unistd.h>
class Task
{
private:
int x;
int y;
char op;
public:
Task(){}
Task(int _x,int _y,char _op):x(_x),y(_y),op(_op){}
void operator()()
{
Handler();
}
void Handler()
{
int res;
switch(op)
{
case '+':
res=x+y;
break;
case '-':
res=x-y;
break;
case '*':
res=x*y;
break;
case '/':
res=x/y;
break;
default:
break;
}
printf("thread:%p done the task: %d %c %d = %d\n",pthread_self(),x,op,y,res);
}
};
int main()
{
srand((unsigned)time(nullptr));
ThreadPool<Task>* tp=new ThreadPool<Task>(NUM);
tp->InitThreadPool();
char ch[]="+-*/";
while(true)
{
int x=rand()%1000+1;
int y=rand()%1000+1;
int op=ch[rand()%4];
tp->Put(Task(x,y,op));
sleep(1);
}
delete tp;
}
设计模式(Design Pattern)是一套被反复使用、多数人知晓的、经过分类的、代码设计经验的总结
为了代码可重用性、让代码更容易被他人理解、保证代码可靠性;设计模式使代码编写真正工程化
一个类只能创建一个对象,即单例模式,该模式可以保证系统中该类只有一个实例,并提供一个访问它的全局访问点,该实例被所有程序模块共享
在某个服务器程序中,该服务器的配置信息存放在一个文件中,这些配置数据由一个单例对象统一读取,然后服务进程中的其他对象再通过这个单例对象获取这些配置信息,这种方式简化了在复杂环境下的配置管理
饿汉模式和懒汉模式
当程序启动时就创建一个唯一的实例对象
template <typename T>
class Singleton
{
public:
static Singleton& GetInstance()//获取实例地址
{
return _s;
}
private:
Singleton();//构造私有化,禁止随意构造
//delete拷贝构造和赋值函数,防拷贝赋值
Singleton(const Singleton& s) = delete;
Singleton& operator=(const Singleton& s) = delete;
static T _s;//在类里的变量都是声明,在cpp文件中进行定义
};
类里面的成员变量只是声明,而静态成员对象需要在类外进行定义,并且不能在.h文件中定义,如果多个.cpp文件包含该头文件,那么则会报重复定义的错误
实现简单
懒汉模式则是需要的时候在第一次调用的时候进行创建
template <typename T>
class Singleton
{
public:
//提供获取对象以及释放对象的静态方法
static Singleton& GetInstance()
{
//提高效率,避免多次锁住及解锁
if (_s == nullptr)
{
//保证线程安全
_m.lock();//锁住
if (_s == nullptr)
{
_s = new Singleton;
}
_m.unlock();//解锁
}
return *_s;
}
static void DelInstance()
{
//提高效率,避免多次锁住及解锁
if (_s != nullptr)//设置volatile关键字避免被编译器优化
{
//保证线程安全
_m.lock();//锁住
if (_s != nullptr)
{
delete _s;
_s = nullptr;
}
_m.unlock();//解锁
}
}
vector<int> _v;
private:
Singleton() {};//要有函数体,否则只是声明,当new的时候找不到对应的实体
Singleton(const Singleton& s) = delete;
Singleton& operator=(const Singleton& s) = delete;
volatile static T* _s;//储存实例对象地址
static mutex _m;//互斥锁
};
对于懒汉模式需要注意的是要保证线程安全,当多个进行调用GetInstance()/DelInstance()时,可能多次进行new和delete,可能造成数据的丢失
无启动负载;可以自由控制多个单例类的定义顺序
实现复杂
ThreadPool.hpp:
#pragma once
#include<pthread.h>
#include<iostream>
#include<queue>
#define NUM 10
template<class T>
class ThreadPool
{
private:
std::queue<T> task_queue;//任务阻塞队列->临界资源
//线程同步与互斥
pthread_mutex_t _lock;
pthread_cond_t _cond;
private:
static ThreadPool<T>* _inst;
//避免构造调用
ThreadPool()
{
pthread_mutex_init(&_lock,nullptr);
pthread_cond_init(&_cond,nullptr);
}
//防拷贝构造
ThreadPool(const ThreadPool<T>&)=delete;
ThreadPool<T>& operator=(const ThreadPool<T>&)=delete;
public:
//封装接口便于传入的对象调用
void Lock()
{
pthread_mutex_lock(&_lock);
}
void UnLock()
{
pthread_mutex_unlock(&_lock);
}
void CondWait()
{
pthread_cond_wait(&_cond,&_lock);
}
void CondWake()
{
pthread_cond_signal(&_cond);
}
bool IsEmpty()
{
return task_queue.size()==0;
}
public:
static ThreadPool<T>* GetInstance()
{
//静态变量,全局只有一个,只在最开始初始化
static pthread_mutex_t mtx=PTHREAD_MUTEX_INITIALIZER;//静态初始化
if(_inst==nullptr)
{
pthread_mutex_lock(&mtx);
if(_inst==nullptr)
{
_inst=new ThreadPool<T>();
}
pthread_mutex_unlock(&mtx);
}
return _inst;
}
~ThreadPool()
{
pthread_mutex_destroy(&_lock);
pthread_cond_destroy(&_cond);
}
void InitThreadPool(int num)
{
pthread_t tid;
for(int i=0;i < num;i++)//创建一批线程
{
pthread_create(&tid,nullptr,Routine,this);//传入线程池地址
pthread_detach(tid);//线程分离
}
}
//线程执行例程
static void* Routine(void* arg)
{
ThreadPool* tp=(ThreadPool*)arg;
//检测是否有任务-没任务等待;有任务进行获取并执行
while(true)
{
//检查
while(tp->IsEmpty())
tp->CondWait();
//取出任务
T task;
tp->Get(&task);
task();//执行逻辑-仿函数
}
}
void Put(const T& t)
{
Lock();
task_queue.push(t);
CondWake();//有任务进行唤醒
UnLock();
}
void Get(T* t)
{
//取任务已经是获取到锁资源了
*t=task_queue.front();
task_queue.pop();
}
};
//静态成员初始化
template<class T>
ThreadPool<T>* ThreadPool<T>::_inst=nullptr;
main.cc:
#include "ThreadPool.hpp"
#include<cstdio>
#include<ctime>
#include<unistd.h>
class Task
{
private:
int x;
int y;
char op;
public:
Task(){}
Task(int _x,int _y,char _op):x(_x),y(_y),op(_op){}
void operator()()
{
Handler();
}
void Handler()
{
int res;
switch(op)
{
case '+':
res=x+y;
break;
case '-':
res=x-y;
break;
case '*':
res=x*y;
break;
case '/':
res=x/y;
break;
default:
break;
}
printf("thread:%p done the task: %d %c %d = %d\n",pthread_self(),x,op,y,res);
}
};
int main()
{
srand((unsigned)time(nullptr));
ThreadPool<Task>* tp=ThreadPool<Task>::GetInstance();
tp->InitThreadPool(NUM);
char ch[]="+-*/";
while(true)
{
int x=rand()%1000+1;
int y=rand()%1000+1;
int op=ch[rand()%4];
tp->Put(Task(x,y,op));
sleep(1);
}
delete tp;
}
STL中的容器不是线程安全的
注:写独占,读共享,读锁优先级高
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
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_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);
这种代码段加锁,会极大地降低我们程序的效率。读写锁可以专门处理这种多读少写的情况
注:写独占,读共享,读锁优先级高
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
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_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);