文章目录
Linux线程
7. 线程池
线程池是一种多线程编程中的技术和概念。
它是一种线程使用模式。是一组预先创建好的线程集合,这些线程处于等待状态,随时准备接受任务并执行。
7.1 线程池介绍
为什么使用线程池
线程过多会带来调度开销,进而影响缓存局部性和整体性能。而线程池维护着多个线程,等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用,还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
线程池的应用场景
(1)需要大量的线程来完成任务,且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务,使用线程池技术是非常合适的。因为单个任务小,而任务数量巨大,你可以想象一个热门网站的点击次数。 但对于长时间的任务,比如一个Telnet连接请求,线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
(2)对性能要求苛刻的应用,比如要求服务器迅速响应客户请求。
(3)接受突发性的大量请求,但不至于使服务器因此产生大量线程的应用。 突发性大量客户请求,在没有线程池情况下,将产生大量线程,虽然理论上大部分操作系统线程数目最大值不是问题,短时间内产生大量线程可能使内存到达极限,出现错误。
使用线程池的优点
(1)提高性能:避免了频繁创建和销毁线程的开销,因为线程的创建和销毁是比较耗时的操作。
(2)控制资源:可以限制线程的数量,防止过多的线程竞争系统资源,导致系统性能下降甚至崩溃。
(3)提高响应性:能够更快地响应新的任务请求,因为线程已经准备好,无需等待线程创建。
7.2 线程池的实现
线程池示例
(1)创建固定数量线程池,循环从任务队列中获取任务对象。
(2)获取到任务对象后,执行任务对象中的任务接口。
执行任务:
#pragma once #include <iostream> #include <string> std::string opers="+-*/%"; enum{ DivZero=1, ModZero, Unknown }; class Task { public: Task() {} Task(int x,int y,char op) :_data1(x),_data2(y),_oper(op),_result(0),_exitcode(0) {} void run() { switch (_oper) { case '+': _result=_data1+_data2; break; case '-': _result=_data1-_data2; break; case '*': _result=_data1*_data2; break; case '/': { if(_data2==0) _exitcode=DivZero; else _result=_data1/_data2; } break; case '%': { if(_data2==0) _exitcode=ModZero; else _result=_data1%_data2; } break; default: _exitcode=Unknown; break; } } //Task对象重载运算符(),()直接进行run函数 void operator()() { run(); } std::string GetResult() { std::string r=std::to_string(_data1); r+=_oper; r+=std::to_string(_data2); r+="="; r+=std::to_string(_result); r+="[code: "; r+=std::to_string(_exitcode); r+="]"; return r; } std::string GetTask() { std::string r=std::to_string(_data1); r+=_oper; r+=std::to_string(_data2); r+="=?"; return r; } ~Task() {} private: int _data1; int _data2; char _oper; int _result; int _exitcode; };
线程池:
#pragma once #include <iostream> #include <vector> #include <queue> #include <string> #include <pthread.h> #include <unistd.h> #include "Task.hpp" struct ThreadData { pthread_t tid; std::string name; }; static const int defaultnum=5; //默认线程数量 //实现我们的线程池 template<class T> class ThreadPool { public: void Lock() { pthread_mutex_lock(&_mutex); } void Unlock() { pthread_mutex_unlock(&_mutex); } void Wakeup() { pthread_cond_signal(&_cond); } void ThreadSleep() { pthread_cond_wait(&_cond,&_mutex); } bool IsQueueEmpty() { return _tasks.empty(); } public: //注意我们线程调用的函数要求参数和返回值都是void* //但是handler在类中默认有this指针->参数不匹配,可以bind或者声明static或放在类外 static void *Handler(/*ThreadPool *this,*/void *args) { ThreadPool<T> *tp=static_cast<ThreadPool<T>*>(args); while(true) { tp->Lock(); while(tp->IsQueueEmpty()) //判断任务是否为空 { tp->ThreadSleep(); //条件变量 } T t=tp->Pop(); //取出任务 tp->Unlock(); t(); //处理任务 std::cout<<" run, "<<"result: "<< t.GetResult()<<std::endl; } return nullptr; } void Start() //启动线程池 { int num=_threads.size(); for(int i=0;i<num;i++) { _threads[i].name="thread-"+std::to_string(i+1); pthread_create(&(_threads[i].tid),nullptr,Handler,this); } } void Push(const T &t) //向任务队列放入任务 { Lock(); _tasks.push(t); //放入任务 Wakeup(); //唤醒线程 Unlock(); } T Pop() //取出任务 { T t=_tasks.front(); _tasks.pop(); return t; } ThreadPool(int num=defaultnum) :_threads(num) { pthread_mutex_init(&_mutex,nullptr); pthread_cond_init(&_cond,nullptr); } ~ThreadPool() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_cond); } private: std::vector<ThreadData> _threads; //线程池 std::queue<T> _tasks; //任务队列 pthread_mutex_t _mutex; //锁 pthread_cond_t _cond; //条件变量 };
运行函数:
#include <iostream> #include "ThreadPool.hpp" int main() { ThreadPool<Task> *tp=new ThreadPool<Task>(5); tp->Start(); srand(time(nullptr) ^ getpid()); while(true) { //1. 构建任务 int x=rand()%10+1; usleep(10); int y=rand()%5; char op=opers[rand()%opers.size()]; Task t(x,y,op); tp->Push(t); //2. 交给线程池处理 std::cout<<"main thread make task: "<<t.GetTask()<<std::endl; sleep(1); } return 0; }
7.3 线程安全的线程池
7.3.1 单例模式的概念
单例模式是一种常见的软件设计模式。
概念:单例模式确保一个类只有一个实例存在,并提供一个全局访问点来获取该实例。
特点包括
唯一性:保证一个类在整个应用程序中只有一个实例。
全局访问:提供了一种全局访问这个唯一实例的方式,方便在程序的任何地方使用。
延迟初始化:通常实例的创建是延迟的,即在首次使用时才创建实例,以提高性能和资源利用率。
单例模式的优点
节省系统资源:避免了频繁创建和销毁对象带来的资源消耗。
统一管理:对唯一的实例进行集中管理和控制,方便维护和修改。
保证一致性:在整个应用中,对于共享的数据或状态,通过单例模式可以保证其一致性。
7.3.2 饿汉和懒汉模式
饿汉模式:
在类加载时就创建单例对象。
优点:线程安全, 因为对象在类加载时就已经创建好了,不存在多线程并发创建的问题。简单直接,实现较为简单。
缺点:无论是否使用,对象都会在类加载时创建,可能会造成一定的资源浪费。
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; } };
懒汉模式实现线程安全的线程池
线程安全的线程池:
#pragma once #include <iostream> #include <vector> #include <queue> #include <string> #include <pthread.h> #include <unistd.h> #include "Task.hpp" struct ThreadData { pthread_t tid; std::string name; }; static const int defaultnum=5; //默认线程数量 //实现我们的线程池 template<class T> class ThreadPool { public: void Lock() { pthread_mutex_lock(&_mutex); } void Unlock() { pthread_mutex_unlock(&_mutex); } void Wakeup() { pthread_cond_signal(&_cond); } void ThreadSleep() { pthread_cond_wait(&_cond,&_mutex); } bool IsQueueEmpty() { return _tasks.empty(); } std::string GetThreadName(pthread_t tid) { for (const auto &ti : _threads) { if (ti.tid == tid) return ti.name; } return "None"; } public: //注意我们线程调用的函数要求参数和返回值都是void* //但是handler在类中默认有this指针->参数不匹配,可以bind或者声明static或放在类外 static void *Handler(/*ThreadPool *this,*/void *args) { ThreadPool<T> *tp=static_cast<ThreadPool<T>*>(args); std::string name = tp->GetThreadName(pthread_self()); while(true) { tp->Lock(); while(tp->IsQueueEmpty()) //判断任务是否为空 { tp->ThreadSleep(); //条件变量 } T t=tp->Pop(); //取出任务 tp->Unlock(); t(); //处理任务 std::cout<<name<<" run, "<<"result: "<< t.GetResult()<<std::endl; } return nullptr; } void Start() //启动线程池 { int num=_threads.size(); for(int i=0;i<num;i++) { _threads[i].name="thread-"+std::to_string(i+1); pthread_create(&(_threads[i].tid),nullptr,Handler,this); } } void Push(const T &t) //向任务队列放入任务 { Lock(); _tasks.push(t); //放入任务 Wakeup(); //唤醒线程 Unlock(); } T Pop() //取出任务 { T t=_tasks.front(); _tasks.pop(); return t; } static ThreadPool<T> *GetInstance() //获取单例对象 { if(nullptr==_tp) //创建单例对象后,不会再有申请和释放锁的操作 { pthread_mutex_lock(&_lock); //保护临界资源 if(_tp==nullptr) { std::cout<<"singleton create done"<<std::endl; _tp=new ThreadPool<T>(); } pthread_mutex_unlock(&_lock); } return _tp; } private: ThreadPool(const ThreadPool<T>&)=delete; const ThreadPool<T>& operator=(const ThreadPool<T>&)=delete; ThreadPool(int num=defaultnum) :_threads(num) { pthread_mutex_init(&_mutex,nullptr); pthread_cond_init(&_cond,nullptr); } ~ThreadPool() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_cond); } private: std::vector<ThreadData> _threads; //线程池 std::queue<T> _tasks; //任务队列 pthread_mutex_t _mutex; //锁 pthread_cond_t _cond; //条件变量 static ThreadPool<T> *_tp; //获取单例指针 static pthread_mutex_t _lock; //锁 }; template<class T> ThreadPool<T> *ThreadPool<T>::_tp=nullptr; template<class T> pthread_mutex_t ThreadPool<T>::_lock=PTHREAD_MUTEX_INITIALIZER;
运行函数:
#include <iostream> #include "ThreadPool.hpp" int main() { //ThreadPool<Task> *tp=new ThreadPool<Task>(5); //tp->Start(); sleep(2); //懒汉模式 ThreadPool<Task>::GetInstance()->Start(); srand(time(nullptr) ^ getpid()); while(true) { //1. 构建任务 int x=rand()%10+1; usleep(10); int y=rand()%5; char op=opers[rand()%opers.size()]; Task t(x,y,op); //tp->Push(t); ThreadPool<Task>::GetInstance()->Push(t); //2. 交给线程池处理 std::cout<<"main thread make task: "<<t.GetTask()<<std::endl; sleep(1); } return 0; }
STL中的容器是否是线程安全的?
不是。原因是 STL 的设计初衷是将性能挖掘到极致,而一旦涉及到加锁保证线程安全,会对性能造成巨大的影响。
而且对于不同的容器,加锁方式的不同,性能可能也不同(例如hash表的锁表和锁桶)。
因此 STL 默认不是线程安全。如果需要在多线程环境下使用,往往需要调用者自行保证线程安全。
智能指针是否是线程安全的?
对于 unique_ptr,由于只是在当前代码块范围内生效,因此不涉及线程安全问题。
对于 shared_ptr,多个对象需要共用一个引用计数变量,所以会存在线程安全问题。但是标准库实现的时候考虑到了这个问题,基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数。
8. 常见锁使用汇总
8.1 互斥锁(Mutex)
确保在同一时刻只有一个线程能够访问被保护的资源。
例如,多个线程同时操作一个共享的全局变量时,使用互斥锁来保证数据的一致性。
初始化互斥锁
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
作用:初始化一个互斥锁。
参数:
mutex:指向要初始化的互斥锁的指针。
attr:互斥锁的属性指针,通常为 NULL(使用默认属性)。
返回值:成功返回 0,失败返回错误码。
加锁
int pthread_mutex_lock(pthread_mutex_t *mutex);
作用:获取互斥锁,如果锁已被占用则阻塞等待。
参数:mutex:要加锁的互斥锁指针。
返回值:成功返回 0,失败返回错误码。
尝试加锁
int pthread_mutex_trylock(pthread_mutex_t *mutex);
作用:尝试获取互斥锁,如果锁可用则获取并返回 0,否则立即返回 EBUSY。
参数:mutex:要尝试加锁的互斥锁指针。
返回值:成功返回 0,锁不可用返回 EBUSY。
解锁
int pthread_mutex_unlock(pthread_mutex_t *mutex);
作用:释放已获取的互斥锁。
参数:mutex:要解锁的互斥锁指针。
返回值:成功返回 0,失败返回错误码。
销毁互斥锁
int pthread_mutex_destroy(pthread_mutex_t *mutex);
作用:销毁指定的互斥锁。
参数:mutex:要销毁的互斥锁指针。
返回值:成功返回 0,失败返回错误码。
8.2 条件变量(Condition Variable)
通常与互斥锁配合使用,用于线程之间的等待和通知。
比如一个线程需要等待某个条件满足后才能继续执行,而另一个线程在条件满足时通知它。
初始化条件变量
int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);
作用:初始化一个条件变量。
参数:
cond:指向要初始化的条件变量的指针。
attr:条件变量的属性指针,通常为 NULL(使用默认属性)。
返回值:成功返回 0,失败返回错误码。
等待条件变量
int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
作用:阻塞当前线程,直到指定的条件变量被唤醒。
参数:
cond:要等待的条件变量指针。
mutex:与条件变量关联的互斥锁指针。
返回值:成功返回 0,失败返回错误码。
定时等待条件变量
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);
作用:阻塞当前线程,直到指定的条件变量被唤醒或到达指定的超时时间。
参数:
cond:要等待的条件变量指针。
mutex:与条件变量关联的互斥锁指针。
abstime:指定的超时时间。
返回值:成功返回 0,超时返回 ETIMEDOUT,失败返回其他错误码。
唤醒一个等待条件变量的线程
int pthread_cond_signal(pthread_cond_t *cond);
作用:唤醒至少一个等待指定条件变量的线程。
参数:cond:要唤醒的条件变量指针。
返回值:成功返回 0,失败返回错误码。
唤醒所有等待条件变量的线程
int pthread_cond_broadcast(pthread_cond_t *cond);
作用:唤醒所有等待指定条件变量的线程。
参数:cond:要唤醒的条件变量指针。
返回值:成功返回 0,失败返回错误码。
销毁条件变量
int pthread_cond_destroy(pthread_cond_t *cond);
作用:销毁指定的条件变量。
参数:cond:要销毁的条件变量指针。
返回值:成功返回 0,失败返回错误码。
8.3 信号量(Semaphore)
用于控制同时访问某一资源的线程数量。
例如限制同时访问数据库连接的线程数量。
初始化信号量
int sem_init(sem_t *sem, int pshared, unsigned int value);
作用:初始化一个信号量。
参数:
sem:指向要初始化的信号量的指针。
pshared:表示信号量的共享属性,0 表示线程间共享,非 0 表示进程间共享。
value:信号量的初始值。
返回值:成功返回 0,失败返回 -1。
等待信号量
int sem_wait(sem_t *sem);
作用:等待信号量的值大于 0,然后将其减 1。
参数:sem:要操作的信号量指针。
返回值:成功返回 0,失败返回 -1。
尝试等待信号量
int sem_trywait(sem_t *sem);
作用:尝试等待信号量,如果信号量的值大于 0,则将其减 1 并立即返回;否则返回错误。
参数:sem:要操作的信号量指针。
返回值:成功返回 0,信号量不可用返回 -1 并设置 errno 为 EAGAIN。
释放信号量
int sem_post(sem_t *sem);
作用:将信号量的值增加 1。
参数:sem:要操作的信号量指针。
返回值:成功返回 0,失败返回 -1。
获取信号量的值
int sem_getvalue(sem_t *sem, int *sval);
作用:获取信号量的当前值,并将其存储在 sval 指向的变量中。
参数:
sem:要操作的信号量指针。
sval:用于存储信号量值的整数指针。
返回值:成功返回 0,失败返回 -1。
销毁信号量
int sem_destroy(sem_t *sem);
作用:销毁指定的信号量。
参数:sem:要销毁的信号量指针。
返回值:成功返回 0,失败返回 -1。
8.4 自旋锁(Spin Lock)
线程在获取锁失败时,会一直循环尝试获取,而不是阻塞等待。
适用于锁被持有的时间很短的情况,能避免线程切换的开销,但如果锁被长时间持有,会浪费 CPU 资源。
初始化自旋锁
int spinlock_init(spinlock_t *lock, const spinlockattr_t *attr);
作用:初始化指定的自旋锁。
参数:
lock:指向要初始化的自旋锁的指针。
attr:自旋锁属性指针,可为 NULL(使用默认属性)。
返回值:成功返回 0,失败返回错误码。
销毁自旋锁
int spinlock_destroy(spinlock_t *lock);
作用:销毁指定的自旋锁。
参数:lock:要销毁的自旋锁指针。
返回值:成功返回 0,失败返回错误码。
尝试获取自旋锁(读)
int spinlock_rdlock(spinlock_t *lock);
作用:尝试获取自旋锁的读锁。
参数:lock:指向要获取读锁的自旋锁的指针。
返回值:成功返回 0,失败返回错误码。
尝试获取自旋锁(写)
int spinlock_wrlock(spinlock_t *lock);
作用:尝试获取自旋锁的写锁。
参数:lock:指向要获取写锁的自旋锁的指针。
返回值:成功返回 0,失败返回错误码。
释放自旋锁
int spinlock_unlock(spinlock_t *lock);
作用:释放指定的自旋锁。
参数:lock:要释放的自旋锁指针。
返回值:成功返回 0,失败返回错误码。
8.5 读写锁(Read-Write Lock)
区分读操作和写操作。允许多个线程同时进行读操作,但在写操作时,不允许其他线程进行读或写操作。
适用于读操作频繁而写操作较少的场景,比如共享数据的读取次数远多于修改次数的情况。
初始化读写锁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
作用:初始化指定的读写锁。
参数:
rwlock:指向要初始化的读写锁的指针。
attr:读写锁属性指针,可为 NULL(使用默认属性)。
返回值:成功返回 0,失败返回错误码。
销毁读写锁
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
作用:销毁指定的读写锁。
参数:rwlock:要销毁的读写锁指针。
返回值:成功返回 0,失败返回错误码。
获取读写锁的读锁
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
作用:尝试获取指定读写锁的读锁。
参数:rwlock:指向要获取读锁的读写锁的指针。
返回值:成功返回 0,失败返回错误码。
获取读写锁的写锁
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
作用:尝试获取指定读写锁的写锁。
参数:rwlock:指向要获取写锁的读写锁的指针。
返回值:成功返回 0,失败返回错误码。
释放读写锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
作用:释放指定的读写锁。
参数:rwlock:要释放的读写锁指针。
返回值:成功返回 0,失败返回错误码。
设置读写锁的优先级
int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
作用:设置读写锁的优先级。
参数:
attr:读写锁属性指针。
pref:优先级选择,有以下 3 种:
PTHREAD_RWLOCK_PREFER_READER_NP
(默认设置)读者优先,可能会导致写者饥饿情况。
PTHREAD_RWLOCK_PREFER_WRITER_NP
写者优先,目前有 BUG,导致表现行为和 PTHREAD_RWLOCK_PREFER_READER_NP
一致。
PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP
写者优先,但写者不能递归加锁。
返回值:成功返回 0,失败返回错误码。
其他概念:
悲观锁:在每次取数据时,总是担心数据会被其他线程修改,所以会在取数据前先加锁(读锁,写锁,行锁等),当其他线程想要访问数据时,被阻塞挂起。
乐观锁:每次取数据时候,总是乐观的认为数据不会被其他线程修改,因此不上锁。但是在更新数据前,会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式:版本号机制和CAS操作。
公平锁:公平锁按照线程请求锁的先后顺序来分配锁。先请求的线程先获取,保证了顺序公平。保证顺序,适合要求严格公平的场景。但性能开销大,高并发时吞吐量可能受影响。
非公平锁:非公平锁不按请求顺序分配锁,锁释放时竞争的线程都可能获取,不一定是先请求的。性能好,高并发时吞吐量可能高。但可能导致线程饥饿,行为不太确定。
CAS操作:当需要更新数据时,判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败,失败则重试,一般是一个自旋的过程,即不断重试。