前面讲到了服务器的基本架构,接下来讲讲支持服务器得以高并发的线程池和数据库连接池的构建。
一、线程池的实现
线程池就是首先创建一些线程,它们的集合称为线程池。线程池在系统启动时即创建大量空闲的线程,程序将一个任务传给线程池,线程池就会启动一条线程来执行这个任务,执行结束以后,该线程并不会死亡,而是再次返回线程池中成为空闲状态,等待执行下一个任务。
为什么需要线程池?
从上面对线程池的描述可知,线程池是预先创建的,负责执行服务器分配的任务,任务完成之后,线程空闲,以待执行另外的任务。
这就体现了线程池的两个好处:
第一、用完线程不销毁,也就避免了因为线程创建和销毁过程造成的时间损耗,提高服务器的并发性。
第二、预先创建一个线程池,也就避免了因为大量客户端建立连接,而去新建线程,以至于造成服务器资源消耗殆尽从而崩溃的风险!提高了服务器的鲁棒性。
下面来看看,如何实现一个线程池:
template <typename T> class threadpool { public: /*thread_number是线程池中线程的数量,max_requests是请求队列中最多允许的、等待处理的请求的数量*/ threadpool(int actor_model, connection_pool *connPool, int thread_number = 8, int max_request = 10000); ~threadpool(); bool append(T *request, int state); bool append_p(T *request); private: /*工作线程运行的函数,它不断从工作队列中取出任务并执行之*/ static void *worker(void *arg);//为什么要用静态成员函数呢? void run(); private: int m_thread_number; //线程池中的线程数 int m_max_requests; //请求队列中允许的最大请求数 pthread_t *m_threads; //描述线程池的数组,其大小m_thread_number std::list<T *> m_workqueue; //请求队列 locker m_queuelocker; //保护请求队列的互斥锁 sem m_queuestat; //是否有任务需要处理 connection_pool *m_connPool; //数据库 int m_actor_model; //模型切换(这个切换是指?) }; template <typename T> threadpool<T>::threadpool( int actor_model, connection_pool *connPool, int thread_number, int max_requests) : m_actor_model(actor_model),m_thread_number(thread_number), m_max_requests(max_requests), m_threads(NULL),m_connPool(connPool) { if (thread_number <= 0 || max_requests <= 0) throw std::exception(); m_threads = new pthread_t[m_thread_number]; //pthread_t是长整型 if (!m_threads) throw std::exception(); for (int i = 0; i < thread_number; ++i) { if (pthread_create(m_threads + i, NULL, worker, this) != 0) { //线程创建成功则返回0,如线程池创建失败则应关闭线程池 delete[] m_threads; throw std::exception(); } if (pthread_detach(m_threads[i])) //主要是将线程属性更改为unjoinable,便于资源的释放,详见PPPPS { delete[] m_threads; throw std::exception(); } } } template <typename T> threadpool<T>::~threadpool() { delete[] m_threads; } template <typename T> bool threadpool<T>::append(T *request, int state) { m_queuelocker.lock(); if (m_workqueue.size() >= m_max_requests) { m_queuelocker.unlock(); return false; } request->m_state = state; m_workqueue.push_back(request); m_queuelocker.unlock(); m_queuestat.post(); return true; } template <typename T> bool threadpool<T>::append_p(T *request) { m_queuelocker.lock(); if (m_workqueue.size() >= m_max_requests) { m_queuelocker.unlock(); return false; } m_workqueue.push_back(request); m_queuelocker.unlock(); m_queuestat.post(); return true; } template <typename T> void *threadpool<T>::worker(void *arg) { threadpool *pool = (threadpool *)arg; pool->run(); return pool; } template <typename T> void threadpool<T>::run() { while (true) { m_queuestat.wait(); m_queuelocker.lock(); if (m_workqueue.empty()) { m_queuelocker.unlock(); continue; } T *request = m_workqueue.front(); m_workqueue.pop_front(); m_queuelocker.unlock(); if (!request) continue; if (1 == m_actor_model) { if (0 == request->m_state) { if (request->read_once()) { request->improv = 1; connectionRAII mysqlcon(&request->mysql, m_connPool); request->process(); } else { request->improv = 1; request->timer_flag = 1; } } else { if (request->write()) { request->improv = 1; } else { request->improv = 1; request->timer_flag = 1; } } } else { connectionRAII mysqlcon(&request->mysql, m_connPool); request->process(); } } }
PS:模板类的初始化以及函数定义的格式即是带模板格式进行定义,如上例所示;
PPS:线程池构造的基本原理即是预先申请一堆线程资源,而后根据新来的任务请求进行线程分配;
PPPS:从上述代码可以看出异常处理机制以及资源获取失败后释放资源链接的重要性。
PPPPS: linux线程pthread有两种状态joinable状态和unjoinable状态:如果线程是joinable状态,当线程函数自己返回退出时或pthread_exit时都不会释放线程所占用堆栈和线程描述符(总计8K多)。
只有当你调用了pthread_join(pthread_join()即是子线程合入主线程,主线程阻塞等待子线程结束,然后回收子线程资源。
之后这些资源才会被释放。若是unjoinable状态的线程,这些资源在线程函数退出时或pthread_exit时自动会被释放。
而unjoinable属性可以在pthread_create时指定,或在线程创建后在线程中pthread_detach(pthread_detach()即主线程与子线程分离,子线程结束后,资源自动回收), 如:pthread_detach(pthread_self()),将状态改为unjoinable状态,确保资源的释放。其实简单的说就是在线程函数头加上 pthread_detach(pthread_self())的话,线程状态改变,在函数尾部直接 pthread_exit线程就会自动退出。省去了给线程擦屁股的麻烦。
二、数据库连接池的实现
class connection_pool { public: MYSQL *GetConnection(); //获取数据库连接 bool ReleaseConnection(MYSQL *conn); //释放连接 int GetFreeConn(); //获取连接 void DestroyPool(); //销毁所有连接 //单例模式 这个是利用局部静态变量懒汉模式实现单例(多线程不友好) //对于多线程而言,多个线程可能同时访问到该静态变量,并发现其没有被初始化(C++实现机制是) //(该看静态变量内部一标志位是为1,为1则说明已被初始化) //多个线程同时判定其没有初始化而后均初始化就会造成错误(即它不是一个原子操作) static connection_pool *GetInstance(); void init(string url, string User, string PassWord, string DataBaseName, int Port, int MaxConn, int close_log); private: connection_pool(); ~connection_pool(); int m_MaxConn; //最大连接数 int m_CurConn; //当前已使用的连接数 int m_FreeConn; //当前空闲的连接数 locker lock; list<MYSQL *> connList; //连接池 sem reserve; public: string m_url; //主机地址 string m_Port; //数据库端口号 string m_User; //登陆数据库用户名 string m_PassWord; //登陆数据库密码 string m_DatabaseName; //使用数据库名 int m_close_log; //日志开关 }; class connectionRAII{ public: connectionRAII(MYSQL **con, connection_pool *connPool); ~connectionRAII(); private: MYSQL *conRAII; connection_pool *poolRAII; };
PS:RAII(Resource Acquisition Is Initialization),也称为“资源获取就是初始化”,是c++等编程语言常用的管理资源、避免内存泄露的方法。它保证在任何情况下,使用对象时先构造对象,最后析构对象。
其基本的实现思想:当我们在一个函数内部使用局部变量,当退出了这个局部变量的作用域时,这个变量也就别销毁了;当这个变量是类对象时,这个时候,就会自动调用这个类的析构函数,而这一切都是自动发生的,不要程序员显示的去调用完。
PS2:数据库连接池负责分配、管理和释放数据库连接,它允许应用程序重复使用一个现有的数据库连接,而不是再重新建立一个;释放空闲时间超过最大空闲时间的数据库连接来避免因为没有释放数据库连接而引起的数据库连接遗漏。这项技术能明显提高对数据库操作的性能。这跟线程池的理念基本一致,即实现申请一堆数据库连接,而后谁用谁拿走,不用就空闲等待下个线程使用。
PS3:数据库连接的获取和释放通过信号量来实现:初始化信号量为连接池中的数量,销毁则信号量+1,获取则信号量-1,如果信号量为0说明连接池为空,那就一直等待;