文章目录
- 线程池
- 线程池代码
- 线程池单例模式
- STL,智能指针和线程安全
- 其他锁（了解）
- 读者写者问题（了解）
线程池
我们去处理任务时，一个任务对应一个创建一个线程进行处理，效率是比较低的。我们可以预先创建一批线程，任务队列里没有任务的时候，每个线程都休眠，当队里中有任务的时候，就可以环形线程进行处理了。唤醒线程的成本比创建整个线程的成本小，这就是线程池的逻辑思想。
线程池: 一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
线程池常见的应用场景如下：
需要大量的线程来完成任务，且完成任务的时间比较短。
对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。
线程池代码
整体框架结构：
Thread.hpp的简单封装线程，下面我们进行简单的验证
Thread类主要成员变量是线程名，函数，线程参数，参数ID以及对应编号
Thread类提供了一个无参构造，完成对成员变量name的赋值。同时，对外主要提供了start接口和join接口，对于join接口就是线程等待，而对于start接口就是创建线程的接口，在外部如果调用的话我们需要传入对应的函数以及线程对应的参数。
#pragma once
#include <pthread.h>
#include <iostream>
#include <cassert>
#include <string>
#include <functional>
namespace ThreadNs
{
typedef std::function<void*(void*)> func_t;
const int num =1024;
class Thread
{
private:
static void* start_routine(void*args)
{
Thread* _this = static_cast<Thread*>(args);
return _this->callback();
}
public:
Thread()
{
char namebuffer[num];
snprintf(namebuffer,sizeof namebuffer,"thread-%d",threadnum++);
name_ = namebuffer;
}
void start(func_t func,void*args = nullptr)
{
func_ = func;
args_ = args;
int n = pthread_create(&tid_,nullptr,start_routine,this);
assert(n==0);
(void)n;
}
void join()
{
int n = pthread_join(tid_,nullptr);
assert(n==0);
(void)n;
}
std::string threadname()
{
return name_;
}
~Thread()
{}
void* callback()
{
return func_(args_);
}
private:
std::string name_;
func_t func_;
void *args_;
pthread_t tid_;
static int threadnum;
};
int Thread::threadnum = 1;
}
//进行调用
using namespace ThreadNs;
void* handler(void*args)
{
string ret = static_cast<const char*>(args);
while(true)
{
cout<<ret<<endl;
sleep(1);
}
return nullptr;
}
int main()
{
Thread t1;
Thread t2;
t1.start(handler,(void*)"thread1");
t2.start(handler,(void*)"thread2");
t1.join();
t1.join();
return 0;
}
对于任务队列，可以由多个线程进行访问，我们就需要加锁保护了，把之前写过的锁的小组件引入进来：
LockGuard.hpp
#include <iostream>
#include <mutex>
class Mutex
{
public:
Mutex(pthread_mutex_t*lock_p=nullptr)
:lock_p_(lock_p)
{}
void lock()
{
if(lock_p_) pthread_mutex_lock(lock_p_);
}
void unlock()
{
if(lock_p_) pthread_mutex_unlock(lock_p_);
}
~Mutex()
{}
private:
pthread_mutex_t * lock_p_;
};
class LockGuard
{
public:
LockGuard(pthread_mutex_t*mutex)
:mutex_(mutex)
{
mutex_.lock();
}
~LockGuard()
{
mutex_.unlock();
}
private:
Mutex mutex_;
};
线程池代码如下：创建一批线程时，我们需要实现线程的运行函数static void*handlerTask,之所以是静态的，是因为我们要把这个运行函数传递给Thread类中的func_，不能有this指针，所以是静态成员函数。而没有this指针，我们无法访问ThreadPool里面的成员变量，所以需要封装接口供其调用。
ThreadPool.hpp
#pragma once
#include "Thread.hpp"
#include "LockGuard.hpp"
#include <vector>
#include <queue>
#include <pthread.h>
#include <mutex>
#include <unistd.h>
using namespace ThreadNs;
const int gnum = 3;
template <class T>
class ThreadPool;
template <class T>
class ThreadData
{
public:
ThreadPool<T> *threadpool;
std::string name;
public:
ThreadData(ThreadPool<T> *tp, const std::string &n) : threadpool(tp), name(n)
{ }
};
template <class T>
class ThreadPool
{
private:
static void *handlerTask(void *args)
{
ThreadData<T> *td = (ThreadData<T> *)args;
ThreadPool<T> *threadpool = static_cast<ThreadPool<T> *>(args);
while (true)
{
T t;
{
LockGuard lockguard(td->threadpool->mutex());
while(td->threadpool->isQueueEmpty())
{
td->threadpool->threadWait();
}
t = td->threadpool->pop();
}
std::cout << td->name << " 获取了一个任务" << t.toTaskString() << "并处理完成，结果是: " << t() << std::endl;
}
delete td;
return nullptr;
}
public:
void lockQueue() { pthread_mutex_lock(&_mutex); }
void unlockQueue() { pthread_mutex_unlock(&_mutex); }
bool isQueueEmpty() { return _task_queue.empty(); }
void threadWait() { pthread_cond_wait(&_cond, &_mutex); }
T pop()
{
T t = _task_queue.front();
_task_queue.pop();
return t;
}
void Push(const T &in)
{
LockGuard lockguard(&_mutex);
_task_queue.push(in);
pthread_cond_signal(&_cond);
}
pthread_mutex_t *mutex()
{
return &_mutex;
}
public:
ThreadPool(const int &num = gnum) : _num(num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for (int i = 0; i < _num; i++)
{
_threads.push_back(new Thread());
}
}
void run()
{
for (const auto &t : _threads)
{
ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
t->start(handlerTask, td);
std::cout << t->threadname() << "start..." << std::endl;
}
}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
for (const auto &t : _threads)
delete t;
}
private:
int _num;
std::vector<Thread *> _threads;
std::queue<T> _task_queue;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
};
我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的；现在我们想像之前处理各种数据的计算，那么先引入任务组件Task.hpp：
Task.hpp
#pragma once
#include <iostream>
#include <functional>
class Task
{
using func_t = std::function<int(int,int ,char)>;
public:
Task(){}
Task(int x,int y,char op,func_t func)
:_x(x),_y(y),_op(op),_callback(func)
{}
std::string operator()()
{
int result = _callback(_x,_y,_op);
char buffer[1024];
snprintf(buffer,sizeof buffer,"%d %c %d = %d",_x,_op,_y,result);
return buffer;
}
std::string toTaskString()
{
char buffer[1024];
snprintf(buffer,sizeof buffer,"%d %c %d = ?",_x,_op,_y);
return buffer;
}
private:
int _x;
int _y;
char _op;
func_t _callback;
};
const std::string oper = "+-*/%";
int mymath(int x,int y,char op)
{
int result = 0;
switch(op)
{
case '+':
result = x+y;
break;
case '-':
result = x-y;
break;
case '*':
result = x*y;
break;
case '/':
if(y==0)
{
std::cerr<<"div zero error!"<<std::endl;
result = -1;
}
else
{
result = x/y;
}
break;
case '%':
if(y==0)
{
std::cerr<<"mod zero error!"<<std::endl;
result = -1;
}
else
{
result = x%y;
}
break;
default:
break;
}
return result;
}
main.cc
#include "ThreadPool.hpp"
#include "Thread.hpp"
#include "Task.hpp"
#include <unistd.h>
#include <ctime>
int main()
{
ThreadPool<Task>* tp = new ThreadPool<Task>();
tp->run();
srand(time(0));
int x,y;
char op;
while(true)
{
x = rand()%10+1;
y = rand()%20+1;
op =oper[rand()%oper.size()];
Task t(x,y,op,mymath);
tp->Push(t);
sleep(1);
}
return 0;
}
线程池单例模式
单例模式是一种创建型设计模式，它保证一个类只有一个实例存在，并且提供一个全局访问点来访问该实例。饿汉方式和懒汉方式是两种常见的单例模式的实现方式。单例模式的主要特点包括：
只能有一个实例。
全局访问点，方便访问该实例。
在很多服务器开发场景中，经常需要让服务器加载很多的数据 （上百G） 到内存中，此时往往要用一个单例的类来管理这些数据。
我们要做的第一步就是把构造函数私有，再把拷贝构造和赋值运算符重载delete：
接下来就要在成员变量中定义一个静态指针，方便获取单例对象。:
在设置获取单例对象的函数的时候，注意要设置成静态成员函数，因为在获取对象前根本没有对象，无法调用非静态成员函数（无this指针）:
主函数进行调用：
不过也许会出现多个线程同时申请资源的场景，所以还需要一把锁来保护这块资源，而这把锁也得设置成静态，因为GetSingle()函数是静态的
运行结果:
线程池单例模式完整代码：
#pragma once
#include "Thread.hpp"
#include "LockGuard.hpp"
#include <vector>
#include <queue>
#include <pthread.h>
#include <mutex>
#include <unistd.h>
using namespace ThreadNs;
const int gnum = 3;
template <class T>
class ThreadPool;
template <class T>
class ThreadData
{
public:
ThreadPool<T> *threadpool;
std::string name;
public:
ThreadData(ThreadPool<T> *tp, const std::string &n) : threadpool(tp), name(n)
{
}
};
template <class T>
class ThreadPool
{
private:
static void *handlerTask(void *args)
{
ThreadData<T> *td = (ThreadData<T> *)args;
ThreadPool<T> *threadpool = static_cast<ThreadPool<T> *>(args);
while (true)
{
T t;
{
// td->threadpool->lockQueue();
LockGuard lockguard(td->threadpool->mutex());
while (td->threadpool->isQueueEmpty())
{
td->threadpool->threadWait();
}
t = td->threadpool->pop(); // pop的本质是将任务从公共队列中拿到当前线程独立的栈中
// td->threadpool->unlockQueue();
}
std::cout << td->name << " 获取了一个任务" << t.toTaskString() << "并处理完成，结果是: " << t() << std::endl;
}
delete td;
return nullptr;
}
ThreadPool(const int &num = gnum) : _num(num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for (int i = 0; i < _num; i++)
{
_threads.push_back(new Thread());
}
}
void operator=(const ThreadPool &) = delete;
ThreadPool(const ThreadPool &) = delete;
public:
void lockQueue() { pthread_mutex_lock(&_mutex); }
void unlockQueue() { pthread_mutex_unlock(&_mutex); }
bool isQueueEmpty() { return _task_queue.empty(); }
void threadWait() { pthread_cond_wait(&_cond, &_mutex); }
T pop()
{
T t = _task_queue.front();
_task_queue.pop();
return t;
}
pthread_mutex_t *mutex()
{
return &_mutex;
}
public:
void run()
{
for (const auto &t : _threads)
{
ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
t->start(handlerTask, td);
std::cout << t->threadname() << "start..." << std::endl;
}
}
void Push(const T &in)
{
LockGuard lockguard(&_mutex);
_task_queue.push(in);
pthread_cond_signal(&_cond);
}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
for (const auto &t : _threads)
delete t;
}
static ThreadPool<T> *getInstance()
{
if (nullptr == tp)
{
_lock.lock();
if (tp == nullptr)
{
tp = new ThreadPool<T>();
}
_lock.unlock();
}
return tp;
}
private:
int _num;
std::vector<Thread *> _threads;
std::queue<T> _task_queue;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
static ThreadPool<T> *tp;
static std::mutex _lock;
};
template <class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;
template <class T>
std::mutex ThreadPool<T>::_lock;
STL,智能指针和线程安全
STL的容器中不是线程安全的：STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响.而且对于不同的容器, 加锁方式的不同, 性能可能也不同。因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全.
对于 unique_ptr, 由于只是在当前代码块范围内生效, 因此不涉及线程安全问题.对于 shared_ptr, 多个对象需要共用一个引用计数变量, 所以会存在线程安全问题. 但是标准库实现的时候考虑到了这个问题, 基于原子操作(CAS)的方式保证 shared_ptr 能够高效, 原子的操作引用计数.
其他锁（了解）
悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS操作。
CAS操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试
自旋锁：使用自旋锁时，当多线程发生竞争锁的情况时，加锁失败的线程会忙等待（这里的忙等待可以用 while 循环等待实现），直到它拿到锁。而互斥锁加锁失败后，线程会让出 CPU 资源给其他线程使用，然后该线程会被阻塞挂起。如果成功申请临界资源的线程，临界区代码执行时间过长，自旋的线程会长时间占用 CPU 资源，所以自旋的时间和临界区代码执行的时间是成正比的关系。如果临界区代码执行的时间很短，就不应该使用互斥锁，而应该选用自旋锁。因为互斥锁加锁失败，是需要发生上下文切换的，如果临界区执行的时间比较短，那可能上下文切换的时间会比临界区代码执行的时间还要长。
读者写者问题（了解）
读写锁由读锁和写锁两部分构成，如果只读取共享资源用读锁加锁，如果要修改共享资源则用写锁加锁。
如果写锁没被写线程持有，多个读线程能够并发持有锁，提高共享资源的访问效率，因为读锁用于读取共享资源，所以多个线程持有读锁也不会破坏共享资源的数据。
一旦写锁被线程持有后，读线程获取锁的操作会被阻塞，而其他写线程的获取写锁的操作也会被阻塞
注意：写独占，读共享，读锁优先级高
读者写者问题和生产者消费者模型的本质区别就是消费者会取走数据，而读者不会取走数据。
读写锁接口
//初始化读写锁
pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
//销毁读写锁
pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
//读加锁
pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
//写加锁
pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
//解锁
pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
