【Linux操作系统】-- 多线程(三)-- 线程池+单例模式
目录
线程池
场景
代码实现
线程安全的单例模式
懒汉实现方式和懒汉实现方式
饿汉方式实现单例模式
懒汉方式实现单例模式
实战代码演练单例模式
线程池
在C++中用户使用new/malloc都是向操作系统OS申请的,在系统的角度,就相当于new/malloc在底层封装了系统调用,当调用系统调用。
- 状态发生变化,就是在malloc申请调用系统调用,需要从用户态转变为内核态,申请完状态再从内核态转变为用户态。
- 在向OS申请的时候,OS并不保证有足够的内存空间给你,或者在申请的时候,OS正在跑其他代码,并不一定有合适的内存块给你。有可能要执行OS内的内存处理算法,把内存碎片合并,或者把不要的内存释放掉,总之它要做更多的工作来腾出你要的空间。对于用户来说,OS做什么工作跟用户根本没有关系,但是OS做的工作所花的实践都嫁接到了用户头上,是耗时的。
所以频繁的使用new/malloc是有可能降低工作效率的,每次执行一次new/malloc都会执行一次耗时动作。与其这样,那么我们不如直接一次跟OS要一大块内存,那么OS系统所做的耗时的动作只需要执行一遍,这样用户想用多少空间就用多少空间,直接从这块池子中拿到空间,而不用一次一次的去向OS申请空间,执行内存管理算法。我们将这一大块内存称为内存池。
那么我们申请到大块内存空间,这个内存空间需要用户来进行管理!第二,内存池主要还是要提高效率。
当我们处理一批任务的时候,如果一批任务到来的时候,通常我们先创建线程,并让线程处理任务,这个是我们在网络服务经常使用的。当任务到来的时候再去创建线程,就相当于我需要内存的时候再从OS申请空间,创建线程也是有成本的。所以我们可以预先创建出一把线程,任务一到,线程已经提前准备好了,我们直接将任务指派给某个线程,让线程去运行就可以。提前准备好的线程,用来随时处理任务,就称为线程池。
内存池是为了提高效率,那么线程池的创建也是为了提高效率。
场景
假设线程池预先创建出一批线程,另有一个线程用来生产任务,我们想把这个生产出来的任务派发给线程池。那么怎么派发呢?我们可以在线程池中维护一个任务队列,这个跟我们写阻塞队列一样,但是这次我们不写固定大小。当线程产生任务,将任务放入任务队列,线程池中的线程竞争式的抢夺,从队列中拿任务,拿到自己的上下文处理任务就可以。那么这就是可以处理任务的线程池。
代码实现
根据上面的场景,我们的线程池,需要有一个类,需要包含一个任务队列,还需要包含若干个线程。
首先里面的成员变量需要一个int num来告诉线程池有几个线程。用queue类型表示队列,里面放的任务类型用模板表示。构造函数中,初始化线程数量,我们定义一个全局变量g_num来初始化,知道数量之后,就创建线程。
那么我们写一个初始化线程来创建一批线程,其中这里我们就不保存线程id了,因为线程创建好之后,主线程就不管他们了,所有新创建的线程直接分离。在初始化创建线程时,我们让每个线程都执行Routinue任务函数。因为我们不想让主线程等待线程池的线程,我们在Routinue函数中分离线程。一旦分离,主线程就会向后走,新线程(线程池线程)继续执行自己的内容。
在类中执行类内成员方法是不可行的,我们不能在类内直接写一个Routinue函数,因为类内函数包含一个隐藏的this指针。想要在类内让线程执行类内方法,必须让线程执行静态方法!静态成员方法是没有this指针的,并且它也无法直接使用非静态成员变量。
下面我们写向队列放任务的函数pushtask。我们在放任务的同时,线程池的线程会抢占式的竞争队列中的任务,这个过程在Rontinue中实现,所以我们需要一把锁来保护临界资源。当我安全的向任务队列放任务,有可能很长时间都没有任务,导致线程们都休眠了,那么我们要判断以下对垒是否为空。当队列是空的时候就跳出循环break,然后解锁,这样的话会导致一个问题,当线程没有抢占任务就会被挂起,而抢占到锁的线程可能会一直抢占,因为从挂起状态到抢占状态需要一段时间,而某一个线程抢占完任务,解锁后,因为没有被挂起,抢占能力更强,导致一支枪展资源,只有一个线程来执行任务。所以没有有任务的时候我们再解锁。
static void* Routinue(void* args)
{
pthread_detach(pthread_self());//将线程分离掉
ThreadPool<T>* tp = (ThreadPool<T>*)args;
while(true)
{
Lock();
if(tp->IsEmpty())
{
break;
}
Unlock();
}
}
所以在队列为空的时候,需要将所有线程挂起。此时需要一个条件变量。当有任务的时候,需要在成员变量队列中拿任务,但是静态函数不可以使用非静态成员变量,但是可以使用非静态函数,所以我们洗一个PopTask,从队列里拿出数据。PopTask函数在拿出数据后,因为pop已经将数据从队列里删除,数据已经不是临界资源,所以可以在锁外面处理数据。t.Run();这样就形成了多个线程同时在执行任务。
因为担心条件不就绪,就执行挂起等待,或者多核cpu处理程序,误判本来队列不为空,判断为空,那么判空就会出现错误,所以我们将if改为while,变成轮询检测。
thread_pool.hpp
#pragma once
#include <iostream>
#include <string>
#include <queue>
#include <unistd.h>
#include <pthread.h>
namespace ns_threadpool
{
const int g_num = 5;
template <class T>
class ThreadPool
{
private:
int num_;
std::queue<T> task_queue_;
pthread_mutex_t mtx_;
pthread_cond_t cond_;
public:
void Lock()
{
pthread_mutex_lock(&mtx_);
}
void Unlock()
{
pthread_mutex_unlock(&mtx_);
}
bool IsEmpty()
{
return task_queue_.empty();
}
void Wait()
{
pthread_cond_wait(&cond_,&mtx_);
}
void WakeUp()
{
pthread_cond_signal(&cond_);
}
public:
ThreadPool(int num = g_num)
: num_(num)
{
pthread_mutex_init(&mtx_,nullptr);
pthread_cond_init(&cond_,nullptr);
}
~ThreadPool()
{
pthread_mutex_destroy(&mtx_);
pthread_cond_destroy(&cond_);
}
static void* Routinue(void* args)
{
pthread_detach(pthread_self());//将线程分离掉
ThreadPool<T>* tp = (ThreadPool<T>*)args;
while(true)
{
tp->Lock();
while(tp->IsEmpty())
{
tp->Wait();//挂起等待
}
//该队列一定有任务,需要从队列拿任务
// T t = task_queue_.front(); --不可以这样写,静态成员函数不可以使用非静态成员变量
T t;
tp->PopTask(&t);
tp->Unlock();
t();
}
}
void InitThreadPool()
{
pthread_t tid;
for (int i = 0; i < num_; i++)
{
pthread_create(&tid, nullptr, Routinue, (void*)this);//要执行对象,传this指针。
}
}
void PushTask(const T& in)
{
Lock();
task_queue_.push(in);
Unlock();
WakeUp();
}
void PopTask(T* out)
{
*out = task_queue_.front();
task_queue_.pop();
}
};
}
main.cc
#include "thread_pool.hpp"
#include "Task.hpp"
#include <ctime>
#include <cstdlib>
using namespace ns_threadpool;
using namespace ns_task;
int main()
{
ThreadPool<Task>* tp = new ThreadPool<Task>(10);
tp->InitThreadPool();//初始化线程池
srand((long long)time(nullptr));
const std::string ops = "+-*/%";
while(true)
{
//网络
Task t(rand()%20 + 1,rand()%10+1 ,"+-*/%"[rand()%ops.size()]);
tp->PushTask(t);
sleep(1);
}
return 0;
}
线程安全的单例模式
有时候,在做服务器开发的时候,会将很多数据加载到内存,这些数据往往只需要一个单例的类来管理这些数据,也就是这些数据在内存中只有一份。所以,如果我们想要一个数据在内存中只出现一次,我们就称之为单例模式。(一个类只能创建一个对象)
当我们定义对象,需要经历两个步骤:
- 开辟空间:开辟空间在编译器编译代码时,当程序加载到内存时,他会自动给你开辟空间。
- 给空间写入初始值:通常调用构造函数初始化。
开辟空间+填入数据,我们叫他初始化的过程,如果将这两个步骤分开,我们叫填入数据为赋值过程。定义对象的本质就是将对象加载到内存,那么单例模式就是让该对象在内存中存在一份,加载一次。
那么什么时候加载,什么时候创建呢?其中就有饿汉模式和懒汉模式,这两种模式。一般而言,我们的独享被设计成单例需要满足两个条件:
- 语义上,只需要一个
- 该对象内部存在大量的空间,保存了大量的数据,如果允许该对象存在多份,或者允许发生各种拷贝,内存中就会存在冗余数据。
懒汉实现方式和懒汉实现方式
通俗点讲,饿汉就是吃完饭立刻洗碗,懒汉就是吃完饭等下次吃饭的时候再洗碗。那么这两种方式有什么区别呢?懒汉最核心的方式就是延迟加载,我们遇见最典型的方式就是写时拷贝。那么饿汉方式有很多弊端,比如开辟空间立马就给你,但是有的空间你用不到,并且创建空间的时间更久,需要开辟更多空间,初始化更多数据。所以这就体现了懒汉模式的好处,需要的时候再做,不需要的时候不做。如果使用饿汉的时候,这会导致程序启动的时候非常慢,如果采用懒汉模式,刚开始启动的时候先不加载,先让程序跑起来,你用到数据的时候,再给你创建,此时通过延时加载的方式让代码启动时速度变快。
饿汉方式实现单例模式
这里有一个静态成员变量data,获取静态成员变量使用静态函数调用。那么我们知道,当构建出这个对象的时候,这个对象在类中已经被创建好了,static的成员函数/成员对象是属于类,而不属于对象的。也就是说下面这个代码编译形成可执行程序,加载到内存时,类只要被加载进来了,那么这个对象也早就存在了。所以当我们在创建这个对象的时候,当加载这个程序时,对象已经就有了。这也就是饿汉方式,创建对象,立马把对象加载出来。
template <typename T>
class Singleton
{
static T data;
public:
static T* GetInstance()
{
return &data;
}
};
懒汉方式实现单例模式
与饿汉不同,饿汉在编译时候就已经开辟好空间,而懒汉在用成员函数获取成员变量的时候,先要判断这个变量是否为空,如果为空就新建,如果已经有了对象,就返回这个存在的对象的地址。这样的话,我们在编译的时候还没有开辟空间,当我们想用这个对象,用懒汉的方式,用的时候再开辟即可。
template <typename T>
class Singleton
{
static T* inst;
public:
static T* GetInstance()
{
if(inst == NULL)
{
inst = new T();
}
return inst;
}
};
实战代码演练单例模式
在这篇文章的开始,我们讲到了线程池,而线程池数据多,在内存中也只需要一份,所以我们可以来写一个单例模式版的线程池。
因为是单例模式,所以我们要把构造函数变成私有的,并且不能有拷贝构造和赋值。并且成员变量需要有一个静态指针指向的这个单例,也就是说需要把那些能够创建对象的方法全部私有化,在类内定义一个私有的指针。因为是类内静态成员变量,需要在类外初始化。获取对象的时候不能直接用类创建对象,需要用一个方法获取到这个指向类的指针。
和以往的方式一样,创建对象后需要初始化,InitThreadPool,所以我们直接在thread_pool类中的GetInstance中创建对象后,直接初始化对象,这样更方便。
thread_pool.hpp
#pragma once
#include <iostream>
#include <string>
#include <queue>
#include <unistd.h>
#include <pthread.h>
namespace ns_threadpool
{
const int g_num = 5;
template <class T>
class ThreadPool
{
private:
//省略...
static ThreadPool<T> *ins_; //
private:
ThreadPool(int num = g_num)
: num_(num)
{
pthread_mutex_init(&mtx_, nullptr);
pthread_cond_init(&cond_, nullptr);
}
ThreadPool(const ThreadPool<T> &tp) = delete; //拷贝构造
ThreadPool<T> &operator=(ThreadPool<T> &tp) = delete; //赋值重载
public:
//省略...
public:
static ThreadPool<T>* GetInstance()
{
if (ins_ == nullptr)
{
ins_ = new ThreadPool<T>();
ins_->InitThreadPool();
}
return ins_;
}
~ThreadPool()
{
pthread_mutex_destroy(&mtx_);
pthread_cond_destroy(&cond_);
}
static void *Routinue(void *args)//静态函数为了传一个参数
{
pthread_detach(pthread_self()); //将线程分离掉
ThreadPool<T> *tp = (ThreadPool<T> *)args;
while (true)
{
tp->Lock();
while (tp->IsEmpty())
{
tp->Wait(); //挂起等待
}
//该队列一定有任务,需要从队列拿任务
// T t = task_queue_.front(); --不可以这样写,静态成员函数不可以使用非静态成员变量
T t;
tp->PopTask(&t);
tp->Unlock();
t();
}
}
void InitThreadPool()
{
pthread_t tid;
for (int i = 0; i < num_; i++)
{
pthread_create(&tid, nullptr, Routinue, (void *)this); //要执行对象,传this指针。
}
}
void PushTask(const T &in)
{
Lock();
task_queue_.push(in);
Unlock();
WakeUp();
}
void PopTask(T *out)
{
*out = task_queue_.front();
task_queue_.pop();
}
};
template <class T>
ThreadPool<T> *ThreadPool<T>::ins_ = nullptr;
}
main.cc
#include "thread_pool.hpp"
#include "Task.hpp"
#include <ctime>
#include <cstdlib>
using namespace ns_threadpool;
using namespace ns_task;
int main()
{
sleep(5);
srand((long long)time(nullptr));
const std::string ops = "+-*/%";
while(true)
{
//网络
Task t(rand()%20 + 1,rand()%10+1 ,ops[rand()%ops.size()]);
ThreadPool<Task>::GetInstance()->PushTask(t);
}
return 0;
}
单例本身会在任何场景,任何环境下调用,GetInstance势必会被多线程重入,导致线程安全问题。比如说,当单例第一次被创建,就在创建到new语句,开辟内存空间到一半,就被切走了,剩下的线程一看,发现此时单例还是空的,还没有被创建出来,这时就会再次重新创建,等到第一个单例被切回来的时候,又创建了一个。
所以可以定义一个static的锁,静态锁初始化可以用宏PTHREAD_MUTEX_INITIALIZER;初始化后就开始上锁,当线程抢占队列任务时候,先竞争锁,创建单例,创建初始化线程之后,解锁然后返回单例。每次上锁解决了单例不安全的问题,但如果队列任务为0的话,每次线程还要抢占锁再判断队列中有没有任务,这样非常消耗资源,所以我们做一个双判断,在锁的外面再加一个判空。
static ThreadPool<T> *GetInstance()
{
static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
//当前单例对象还没有被创建按,上锁
if (ins_ == nullptr) //双判定,减少锁的征用,提高获取单例的效率
{
pthread_mutex_lock(&lock);
if (ins_ == nullptr)
{
ins_ = new ThreadPool<T>();
ins_->InitThreadPool();
}
pthread_mutex_unlock(&lock);
}
return ins_;
}