[mysql 基于C++实现数据库连接池连接池的使用] 持续更新中-编程知识

目背景
常见的MySQL、Oracle、SQLServer等数据库都是基于C/S架构设计的，即（客户端/服务器）架构，也就是说我们对数据库的操作相当于一个客户端，这个客户端使用既定的API把SQL语句通过网络发送给服务器端，MySQL Server执行完SQL语句后将结果通过网络返回客户端。通过网络通信的话就要涉及到TCP/IP协议里的“三次握手”、“四次挥手”等，大量访问时，每一个用户的请求都会对应一次“三次握手”、“四次挥手”的过程，这个性能的消耗是相当严重的；
对于数据库本质上是对磁盘的操作，如果对数据库的访问过多，即（I/O）操作过多，会出现访问瓶颈。
而常见的解决数据库访问瓶颈的方法有两种：

一、为减少磁盘 I/O的次数，在数据库和服务器的应用中间加一层缓存数据库（例如：Redis、Memcache）;
二、增加连接池，来减少高并发情况下大量 TCP三次握手、MySQL Server连接认证、MySQL Server关闭连接回收资源和TCP四次挥手所耗费的性能。

mysqlconn.hpp 实现连接增删改查操作

#include <mysql/mysql.h>
#include <iostream>
#include <string>
#include <ctime>
#include <chrono>
#include <memory> #define INFO    1
#define WARNING 2
#define ERROR   3
#define FATAL   4#define LOG(level, message) Log(#level, message, __FILE__, __LINE__)void Log(std::string level, std::string message, std::string file_name, int line)
{std::cout<<"["<<level<<"]["<<time(nullptr)<<"]["<<message<<"]["<<file_name<<"]["<<line<<"]"<<std::endl;
}class mysqlconn{private:MYSQL *m_conn = nullptr;MYSQL_RES* m_res = nullptr;//查询结果集MYSQL_ROW m_row;//记录结构体void freeResult(){if(m_res){mysql_free_result(m_res);m_res = nullptr;}}std::chrono::steady_clock::time_point m_aliveTime;public:mysqlconn(){//获取一个MYSQL句柄m_conn = mysql_init(nullptr);//设置字符集mysql_set_character_set(m_conn,"utf8");}~mysqlconn(){freeResult();if(m_conn != nullptr){mysql_close(m_conn);}}bool query(std::string sql){freeResult();if(mysql_query(m_conn, sql.c_str())){return false;}m_res = mysql_store_result(m_conn);return true;}//更新 修改 删除bool update(std::string sql){return mysql_query(m_conn, sql.c_str());}//连接指定的数据库bool connect(std::string ip, std::string user, std::string passwd, std::string dbName,  unsigned int port){return mysql_real_connect(m_conn, ip.c_str(), user.c_str(), passwd.c_str(), dbName.c_str(), port,nullptr,0) != nullptr;}//遍历得到的结果集bool next(){if(m_res != nullptr){m_row = mysql_fetch_row(m_res);  //获取一行if(m_row != nullptr){return true;}}return false;}//获取结果集里的值std::string value(int index){int rowCount = mysql_num_fields(m_res);  //返回结果集中字段数目if(index >= rowCount || index < 0){return std::string();}char* ans = m_row[index];unsigned long length = mysql_fetch_lengths(m_res)[index];return std::string(ans,length);		}//事务处理提交方式bool transaction(){return mysql_autocommit(m_conn,false);}//事务提交bool commit(){return mysql_commit(m_conn);}//事务回滚bool rollback(){return mysql_rollback(m_conn);}//更新空闲时间点void refreshAliveTime(){m_aliveTime = std::chrono::steady_clock::now();}//计算连接空闲时长long long getAliveTime(){std::chrono::duration<double> diff = std::chrono::steady_clock::now() - m_aliveTime;       //nanosecods 纳秒return diff.count();}};

connpool.hpp 连接池

#include <mutex>
#include <condition_variable>
#include <queue>
#include <fstream>
#include <thread>#include "mysqlconn.hpp"class ConnectionPool
{private:std::string m_user;std::string m_passwd;std::string m_ip;std::string m_dbName;unsigned short m_port;//连接的上限和下限,自动维护线程池的连接数int m_minSize;int m_maxSize;//连接的超时时长int m_timeout;int m_maxIdleTime;//线程同步  std::mutex m_mutexQ;                     //互斥锁std::condition_variable m_cond;          //条件变量std::queue<mysqlconn *> m_connectionQ;    //共享资源public://对外接口,获取线程池//静态局部变量是线程安全的static ConnectionPool  *getConnectPool()    {static ConnectionPool pool;return &pool;}//获取线程池中的连接std::shared_ptr<mysqlconn>  getConnection(){//需要操作共享资源std::unique_lock<std::mutex> locker(m_mutexQ);//判断连接池队列为空while(m_connectionQ.empty()){if(std::cv_status::timeout == m_cond.wait_for(locker, std::chrono::milliseconds(m_timeout))){if(m_connectionQ.empty()){continue;}}}//自定义shared_ptr析构方法,重新将连接放回到连接池中,而不是销毁std::shared_ptr<mysqlconn> connptr(m_connectionQ.front(),[this](mysqlconn *conn){std::unique_lock<std::mutex> locker(m_mutexQ);conn->refreshAliveTime();m_connectionQ.push(conn);	});//弹出,放到了队尾m_connectionQ.pop();m_cond.notify_all();return connptr;}//防止外界通过拷贝构造函数和移动拷贝构造函数ConnectionPool(const ConnectionPool &obj) = delete;ConnectionPool& operator=(const ConnectionPool& obj) = delete;~ConnectionPool(){while(!m_connectionQ.empty()){mysqlconn *conn = m_connectionQ.front();m_connectionQ.pop();delete conn;}}
private://构造函数私有化ConnectionPool(){//加载配置文件if(!parseJsonFile()){return;}//创建最少连接数for(int i=0;i<m_minSize;++i){addConnect();}//创建子线程用于检测并创建新的连接std::thread producer(&ConnectionPool::produceConnection,this);//销毁连接,检测并销毁连接std::thread recycler(&ConnectionPool::recycleConnection,this);//设置线程分离producer.detach();recycler.detach();}//解析配置文件bool parseJsonFile(){    //可以通过配置文件配置数据 这里写死 m_ip      = "127.0.0.1";m_user    = "pig";m_passwd  = "test1234";m_dbName  = "test";m_port    = 3306;m_minSize = 10;m_maxSize = 100;m_timeout = 10;m_maxIdleTime = 20;return true;}//任务函数void produceConnection()   //生产数据库连接{//通过轮询的方式不断的去检测while(true) {//操作共享资源,需要加锁std::unique_lock<std::mutex> locker(m_mutexQ);//判断连接数是否达到容量,如果大于等于容量则需要阻塞一段时间while (m_connectionQ.size() >= m_maxSize)   {m_cond.wait(locker);}addConnect();m_cond.notify_all();        //唤醒消费者}}void recycleConnection()   //销毁数据库连接{while(true){//休眠一定的时长std::this_thread::sleep_for(std::chrono::milliseconds(500));std::unique_lock<std::mutex> locker(m_mutexQ);//让线程池中最少保持用于 m_minSize个线程while(m_connectionQ.size() > m_minSize){mysqlconn *recyConn = m_connectionQ.front();//如果超时则销毁if(recyConn->getAliveTime() >= m_maxIdleTime){m_connectionQ.pop();delete recyConn;} else{break;}}}}void addConnect()         //添加连接{mysqlconn *conn = new mysqlconn;conn->connect(m_ip,m_user,m_passwd,m_dbName,m_port);conn->refreshAliveTime();m_connectionQ.push(conn);}};

main.cpp 测试主函数单线程连接池多线程连接池

#include "connpool.hpp"void pthread1_no_pool()
{clock_t begin = clock();std::unique_ptr<mysqlconn> sp = std::make_unique<mysqlconn>();bool connflag = sp->connect("127.0.0.1","pig","test1234", "test",3306);if(connflag == false) return;for (int i = 0; i < 4 * 1000; ++i){sp->refreshAliveTime();char sql[1024] = { 0 };sprintf(sql, "insert into tb_file values('%d','%s','%s');",i, "pthread1_no_pool", "1.png");auto upflag = sp->update(sql);}clock_t end = clock();std::cout << "pthread1_no_pool:" << (end - begin) << "ms" << std::endl;}void pthread1_use_pool(){ConnectionPool *cp = ConnectionPool::getConnectPool();clock_t begin = clock();std::shared_ptr<mysqlconn> sp = cp->getConnection();for (int i = 0; i < 1000 * 4; ++i){char sql[1024] = { 0 };sprintf(sql, "insert into tb_file(id, name, file) values('%d','%s','%s');",i, "pthread1_use_pool", "1.png");sp->update(sql);}clock_t end = clock();std::cout <<"pthread1_use_pool:" << (end - begin) << "ms" << std::endl;}void pthread4_no_pool()
{clock_t begin = clock();std::thread tt[4];for(int n = 0; n < 4; n++){tt[n] = std::thread([=]{std::unique_ptr<mysqlconn> sp = std::make_unique<mysqlconn>();sp->connect("127.0.0.1","pig","test1234", "test",3306);for (int i = 0; i < 1000 * (n + 1); ++i){sp->refreshAliveTime();char sql[1024] = { 0 };sprintf(sql, "insert into tb_file values('%d','%s','%s');",i, "pthread1_no_pool", "1.png");sp->update(sql);}});}for(int i = 0; i < 4; i++){tt[i].join();}clock_t end = clock();std::cout <<"pthread4_no_pool:" << (end - begin) << "ms" << std::endl;}void work(ConnectionPool *cp , int l){std::shared_ptr<mysqlconn> sp = cp->getConnection();for (int i = l * 1000; i < 1000 * (l + 1); ++i){char sql[1024] = { 0 };sprintf(sql, "insert into tb_file values('%d','%s','%s');",i, "pthread1_use_pool", "1.png");auto upflag = sp->update(sql);if(upflag != 0){std::cout <<"pthread4_use_pool:" << upflag << sql << std::endl;continue;}}
}void pthread4_use_pool()
{ConnectionPool *cp = ConnectionPool::getConnectPool();clock_t begin = clock();std::thread tt[4];for(int i = 0; i < 4; i++){tt[i] = std::thread(work, cp, i);}for(int i = 0; i < 4; i++){tt[i].join();}clock_t end = clock();std::cout <<"pthread4_use_pool:" << (end - begin) << "ms" << std::endl;
}// g++ -o main main.cpp connpool.hpp mysqlconn.hpp -lmysqlclient -std=c++14 -lpthread
int main()
{/*单线程 不使用连接池*///LOG(INFO, "pthread1_no_pool test:");//pthread1_no_pool();/*单线程 使用连接池*///LOG(INFO, "pthread1_use_pool test:");//pthread1_use_pool();/*多线程 不使用连接池*/LOG(INFO, "pthread4_no_pool test:");pthread4_no_pool();/*多线程 使用连接池*///LOG(INFO, "pthread4_use_pool test:");//pthread4_use_pool();return 0;
}