C++实现MySQL数据库连接池

C++实现MySQL数据库连接池

涉及技术

MySQL数据库编程、单例模式、STL容器、C++11多线程（线程互斥、线程互斥、线程同步通信和unique_lock）、智能指针shared_ptr、lambda表达式、生产者-消费者线程模型。

项目背景

为了提升MySQL数据库（基于C/S设计（客户端-服务器））的访问瓶颈，除了在服务器端增加缓冲服务器缓存常用的数据之外（例如radis、其实也就是建立一个常访问的key-value对，便于快速索引），还可以增加连接池。来提高MySQL Server的访问效率，在高并发环境下，大量的TCP三次握手、MySQl Server连接验证、MySQL Server关闭连接回收资源和TCP四次挥手 所耗费的性能时间也是很明显的，增加连接池就是为了减少这一部分性能损耗（直接从数据池中获取可用连接，就不用重新建立连接）。（客户端和数据库服务器端的通信一般分为：建立TCP连接、MySQL连接的验证和指令通信，TCP四次挥手，更多TCP连接和断开可以查看一文彻底搞懂 TCP三次握手、四次挥手过程及原理）。

目前市场主流的连接池有druid，c3p0和apache dbcp连接池，他们对于短时间内大量的数据库增删改查操作性能的提升是很明显的，但是它们都是基于Java实现的。

连接池的功能

连接池一般包含了数据库连接所使用的ip地址、port端口、用户名和密码以及其他的性能参数，例如初始连接量、最大连接量、最大空闲时间、连接超时时间，该项目主要是基于C++实现的连接池，主要实现以上几个所有连接池都支持的通用功能。

初始连接量(initSize)：表示连接池事先会和MySQL Server创建initSize个数的connection连接，当应用发起MySQL访问时，不用再创建和MySQL Server新的连接，直接从连接池中获取一个可用的连接就可以，使用完成后，并不去释放connection，而是把当前connection再归还到连接池当中。

最大连接量(maxSize)：当并发访问MySQL Server的请求增多时，初始连接量已经不够使用了，此时会根据新的请求数量去创建更多的连接给应用去使用，但是新创建的连接数量上限是maxSize，不能无限制的创建连接，因为每一个连接都会占用一个socket资源，一般连接池和服务器程序是部署在一台主机上的，如果连接池占用过多的socket资源，那么服务器就不能接收太多的客户端请求了。当这些连接使用完成后，再次归还到连接池当中来维护。

最大空闲时间(maxIdleTime)：：当访问MySQL的并发请求多了以后，连接池里面的连接数量会动态增加，上限是maxSize个，当这些连接用完再次归还到连接池当中。如果在指定的maxIdleTime里面，这些新增加的连接都没有被再次使用过，那么新增加的这些连接资源就要被回收掉，只需要保持初始连接量initSize个连接就可以了。

连接超时时间(connectionTimeOut)：当MySQL的并发请求量过大，连接池中的连接数量已经到达maxSize了，而此时没有空闲的连接可供使用，那么此时应用从连接池获取连接无法成功，它通过阻塞的方式获取连接的时间如果超过connectionTimeout时间，那么获取连接失败，无法访问数据库。

MySQL Server参数介绍

功能实现设计

ConnectionPool.cpp和ConnectionPool.h：连接池代码实现
Connection.cpp和Connection.h：数据库操作代码、增删改查代码实现
连接池主要包含了以下功能点：
1.连接池只需要一个实例，所以ConnectionPool以单例模式进行设计
2.从ConnectionPool中可以获取和MySQL的连接Connection
3.空闲连接Connection全部维护在一个线程安全的Connection队列中，使用线程互斥锁保证队列的线程安全
4.如果Connection队列为空，还需要再获取连接，此时需要动态创建连接，上限数量是maxSize
5.队列中空闲连接时间超过maxIdleTime的就要被释放掉，只保留初始的initSize个连接就可以了，这个功能点肯定需要放在独立的线程中去做
6.如果Connection队列为空，而此时连接的数量已达上限maxSize，那么等待connectionTimeout时间如果还获取不到空闲的连接，那么获取连接失败，此处从Connection队列获取空闲连接，可以使用带超时时间的mutex互斥锁来实现连接超时时间
7.用户获取的连接用shared_ptr智能指针来管理，用lambda表达式定制连接释放的功能（不真正释放连接，而是把连接归还到连接池中）
8.连接的生产和连接的消费采用生产者-消费者线程模型来设计，使用了线程间的同步通信机制条件变量和互斥锁

实现代码参考

connection.h

#pragma once
#include <mysql.h>
#include <string>
#include<ctime>
using namespace std;// 数据库操作类
class Connection
{
public:// 初始化数据库连接Connection();// 释放数据库连接资源~Connection();// 连接数据库bool connect(string ip, unsigned short port, string user, string password,string dbname);// 更新操作 insert、delete、updatebool update(string sql);// 查询操作 selectMYSQL_RES* query(string sql);//刷新一下连接的起始的空闲时间void refreshAliveTime();//返回连接的存活时间clock_t getAliveTime();
private:MYSQL* _conn; // 表示和MySQL Server的一条连接clock_t _alivetime;//记录进入空闲状态后的起始时间
};

connection.cpp

#include"connection.h"
#include"public.h"// 初始化数据库连接
Connection::Connection()
{_conn = mysql_init(nullptr);
}
// 释放数据库连接资源
Connection::~Connection()
{if (_conn != nullptr)mysql_close(_conn);
}
// 连接数据库
bool Connection::connect(string ip, unsigned short port, string user, string password,string dbname)
{MYSQL* p = mysql_real_connect(_conn, ip.c_str(), user.c_str(),password.c_str(), dbname.c_str(), port, nullptr, 0);return p != nullptr;
}
// 更新操作 insert、delete、update
bool Connection::update(string sql)
{if (mysql_query(_conn, sql.c_str())){LOG("更新失败:" + sql);return false;}return true;
}
// 查询操作 select
MYSQL_RES* Connection::query(string sql)
{if (mysql_query(_conn, sql.c_str())){LOG("查询失败:" + sql);return nullptr;}return mysql_use_result(_conn);
}
//刷新一下连接的起始的空闲时间
void  Connection::refreshAliveTime()
{_alivetime = clock_t();
}//返回连接的存活时间
clock_t Connection::getAliveTime()
{return clock() - _alivetime;
}

commConnectionPool.h

#pragma once
#include<string>
#include<queue>
#include<mutex>
#include<atomic>
#include<thread> 
#include<memory>
#include<functional>
#include<condition_variable>
#include"connection.h"using namespace std;class ConnectionPool {
public://获取连接池对象实例，静态成员方法，不依赖于对象调用static ConnectionPool* getConnectionPool();//从线程池中获取线程shared_ptr<Connection> getConnection();
private:// 单例，构造函数私有化ConnectionPool();//加载配置文件bool loadConfigFile();//用来线程来产生连接void produceConnectionTask();//用来扫描连接的进程函数，防止很线程池中线程数量大于initSize而不归还资源void scannerConnectionTask();string _ip; //mysql的ip地址unsigned short _port; //mysql的端口号string _username; //mysql的连接用户名string _password; //mysql连接用户的密码string _dbname;int _initSize; //初始化连接池的数量int _maxSize; //最大化连接池的数量int _maxIdleTime; //连接池最大空闲时间int _connectionTimeout;//连接池最大获取时间queue<Connection*> _connectionQue; //连接池存储数据库连接队列，必须是多线程安全的mutex _queueMutex;//维护连接池队列线程安全的互斥锁atomic_int _connectionCnt;//记录所创建的connection数量condition_variable cv;//设置条件变量，用于连接生产线程和连接消费线程的实现
};

commConnectionPool.cpp

#include"commConnectionPool.h"
#include"public.h"//线程安全的懒汉单例模式接口
ConnectionPool* ConnectionPool::getConnectionPool() {//静态局部变量由编译器自动lock和unlockstatic ConnectionPool pool;return &pool;
}//从线程池中获取线程
shared_ptr<Connection> ConnectionPool::getConnection() {unique_lock<mutex> lock(_queueMutex);while (_connectionQue.empty()) {if (cv_status::timeout == cv.wait_for(lock, chrono::milliseconds(_connectionTimeout))){if (_connectionQue.empty()) {LOG("获取空闲连接超时，获取失败!");return nullptr;}}}/*shared_ptr智能指针析构时，会把connection资源直接给delete掉，相当于调用了Connection的析构函数，connect就会被关闭掉了，这里需要自定义shared_ptr的释放资源方式，把connection归还到队列中*/shared_ptr<Connection> sp(_connectionQue.front(),[&](Connection* pcon) {//这里是在服务器应用线程中调用的，所以一定要考虑线程安全unique_lock<mutex> lock(_queueMutex);//智能指针析构的时候会将指针重新输入到队列中_connectionQue.push(pcon);pcon->refreshAliveTime();});_connectionQue.pop();cv.notify_all();//消费完连接之后，通知生产者线程检查一下，如果生产队列为空后，就通知线程赶紧生产return sp;
}ConnectionPool::ConnectionPool() {if (!loadConfigFile()) {return;}for (int i = 0; i < _initSize; i++) {Connection* p = new Connection();p->connect(_ip, _port, _username, _password, _dbname);p->refreshAliveTime();_connectionQue.push(p);_connectionCnt++;}//启动一个新的线程，作为一个连接的生产者thread produce(std::bind(& ConnectionPool::produceConnectionTask, this));//分离线程（分离线程），主线程结束后该线程自动结束produce.detach();//启动一个定时线程，扫描超过maxIdleTime时间的空闲线程thread scanner(std::bind(&ConnectionPool::scannerConnectionTask, this));scanner.detach();
}//用来扫描连接的进程函数，防止很线程池中线程数量大于initSize而不归还资源
void ConnectionPool::scannerConnectionTask()
{while (true) {//通过sleep模拟定时效果this_thread::sleep_for(chrono::seconds(_maxIdleTime));//扫秒整个队列，释放多余的连接//队列要用互斥锁，防止多线程访问unique_lock<mutex> lock(_queueMutex);while (_connectionCnt > _initSize){Connection* p = _connectionQue.front();//如果队列头都没有超时的话，那么后面的connection肯定不会超时//每次回收返回队列都是插入在队尾的if (p->getAliveTime() >= _maxIdleTime){_connectionQue.pop();_connectionCnt--;delete p;}else {break;}}}
}//用来线程来产生连接
void ConnectionPool::produceConnectionTask() {while (true) {//所有的线程在创建时都被_queueMutex锁住了，共用一把锁，函数作用域结束后默认解锁//unique_lock无默认参数时会自动加锁unique_lock<mutex> lock(_queueMutex);while (!_connectionQue.empty()) {//condition_variable cv 必须和unique_lock一起使用cv.wait(lock);//队列不为空，此处生产者线程进入等待状态}if (_connectionCnt < _maxSize) {Connection* p = new Connection();p->connect(_ip, _port, _username, _password, _dbname);p->refreshAliveTime();_connectionQue.push(p);_connectionCnt++;}cv.notify_all();//通知消费者线程可以进行连接了}
}bool ConnectionPool::loadConfigFile() {//读取配置文件FILE* pf = fopen("./mysql.ini", "r");if (pf == nullptr) {LOG("mysql.ini file is not exist!");return false;}//feof()函数判断文件字节流是否到了末尾while (!feof(pf)) {char line[1024] = { 0 };//fgets获取文件一行，并指定行的最大值（包含最后的空字符）//如果成功，该函数返回相同的 str 参数。//如果到达文件末尾或者没有读取到任何字符，str 的内容保持不变，并返回一个空指针。//如果发生错误，返回一个空指针。fgets(line, 1024, pf);string str = line;int idx=str.find('=', 0);if (idx == -1) {//无效的配置continue;}int endx = str.find('\n', idx);string key = str.substr(0, idx);string value = str.substr(idx + 1, endx - idx - 1);if (key == "ip") {_ip = value;}else if (key == "port") {_port = atoi(value.c_str());}else if (key == "username") {_username = value;}else if (key == "password") {_password = value;}else if (key == "dbname") {_dbname = value;}else if (key == "initSize") {_initSize = atoi(value.c_str());}else if (key == "maxSize") {_maxSize = atoi(value.c_str());}else if (key == "maxIdleTime") {_maxIdleTime = atoi(value.c_str());}else if (key == "connectionTimeout") {_connectionTimeout = atoi(value.c_str());}}return true;
}

public.h

#pragma once
#include<iostream>#define LOG(str) std::cout<<__FILE__<<" : "<<__LINE__<<" : "<<__TIMESTAMP__<<" : "<<str<<endl;

main.cpp

#include<iostream>
#include<string>
#include<ctime>
#include<thread>
#include"connection.h"
#include"commConnectionPool.h"using namespace std;int main(int argc, char argv[]) {clock_t begin = clock();int number = 5000;bool is_use_connection_pool = true;/*if (!is_use_connection_pool){for (int i = 0; i < number; i++){Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");conn.update(sql);}}else {//获取静态唯一的连接池，也是静态变量和静态方法的好处ConnectionPool* pool = ConnectionPool::getConnectionPool();for (int i = 0; i < number; i++){shared_ptr<Connection> sp = pool->getConnection();char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");sp->update(sql);}}*/Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");//多线程-未使用连接池thread t1([]() {for (int i = 0; i < 250; i++){Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");conn.update(sql);}});thread t2([]() {for (int i = 0; i < 250; i++){Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");conn.update(sql);}});thread t3([]() {for (int i = 0; i < 250; i++){Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");conn.update(sql);}});thread t4([]() {for (int i = 0; i < 250; i++){Connection conn;conn.connect("127.0.0.1", 3306, "root", "xiehou", "chat");char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");conn.update(sql);}});//多线程-线程池/*thread t1([]() {ConnectionPool* pool = ConnectionPool::getConnectionPool();for (int i = 0; i < 1250; i++){shared_ptr<Connection> sp = pool->getConnection();char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");sp->update(sql);}});thread t2([]() {ConnectionPool* pool = ConnectionPool::getConnectionPool();for (int i = 0; i < 1250; i++){shared_ptr<Connection> sp = pool->getConnection();char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");sp->update(sql);}});thread t3([]() {ConnectionPool* pool = ConnectionPool::getConnectionPool();for (int i = 0; i < 1250; i++){shared_ptr<Connection> sp = pool->getConnection();char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");sp->update(sql);}});thread t4([]() {ConnectionPool* pool = ConnectionPool::getConnectionPool();for (int i = 0; i < 1250; i++){shared_ptr<Connection> sp = pool->getConnection();char sql[1024] = { 0 };// 项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')", "zhangsan", 20, "male");sp->update(sql);}});*///join表示要等子线程结束后，后面的主线程才能继续执行，也就是end计时要等t1~t4结束后才能执行，detach则不用等待t1.join();t2.join();t3.join();t4.join();clock_t end = clock();std::cout << (end - begin) << "ms" << endl;return 0;
}

mysql.ini

# 用于存储mysql的基本连接信息
ip=127.0.0.1
port=3306
username=root
password=xiehou
dbname=chat
initSize=10
maxSize=1024
# 单位是秒
maxIdleTime=60
# 单位是毫秒
connectionTimeout=100

压力测试

数据量	未使用连接池消耗时间（ms）	使用连接池消耗的时间（ms）
1000	单线程：10548 三线程：3472	单线程：4577 三线程：2858
5000	单线程：53145 三线程：17485	单线程：22877 三线程：14865

C++多进程/多线程

更多详细资料可以查看C++多线程详解（全网最全）。

主要涉及头文件thread、mutex、atomic、condition_variable、future。

MySQL数据库常用命令

MySQL数据库程序需要以;结尾，类似于C/C++程序一样。并且MySQL程序对大小写不敏感，也就是不太区分大小写。

数据库相关

创建数据库命令

create database name; //管理员用户
mysqladmin -u root -p create name  //普通用户，需要使用管理员权限

如果数据库存在则会创建失败，得到类似以下信息ERROR 1007 (HY000): Can't create database 'name'; database exists。
推荐的创建命令是CREATE DATABASE IF NOT EXISTS name DEFAULT CHARSET utf8 COLLATE utf8_general_ci;

删除数据库命令

drop database name

删除数据库的时候将会有提示信息，确定是否删除。

选择数据库

use name

显示数据库所有表格

show tables;

数据库表格相关

数据类型

MySQL 支持多种类型，大致可以分为三类：数值、日期/时间和字符串(字符)类型。

数值类型

MySQL 支持所有标准 SQL 数值数据类型。

这些类型包括严格数值数据类型(INTEGER、SMALLINT、DECIMAL 和 NUMERIC)，以及近似数值数据类型(FLOAT、REAL 和 DOUBLE PRECISION)。

关键字INT是INTEGER的同义词，关键字DEC是DECIMAL的同义词。

BIT数据类型保存位字段值，并且支持 MyISAM、MEMORY、InnoDB 和 BDB表。

作为 SQL 标准的扩展，MySQL 也支持整数类型 TINYINT、MEDIUMINT 和 BIGINT。

日期和时间类型

表示时间值的日期和时间类型为DATETIME、DATE、TIMESTAMP、TIME和YEAR。

每个时间类型有一个有效值范围和一个"零"值，当指定不合法的MySQL不能表示的值时使用"零"值。

字符串类型

字符串类型指CHAR、VARCHAR、BINARY、VARBINARY、BLOB、TEXT、ENUM和SET。

数据表格常用命令

创建数据表格

CREATE TABLE table_name (column_name column_type);

以下例子中我们将在数据库中创建数据表runoob_tbl，注意的是：表名和列名都不需要引号括起来：

CREATE TABLE IF NOT EXISTS `runoob_tbl`(`runoob_id` INT UNSIGNED AUTO_INCREMENT,`runoob_title` VARCHAR(100) NOT NULL,`runoob_author` VARCHAR(40) NOT NULL,`submission_date` DATE,`sex` ENUM('男','女')   DEFAULT NULL,PRIMARY KEY ( `runoob_id` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

实例解析：

如果你不想字段为 NULL 可以设置字段的属性为 NOT NULL， 在操作数据库时如果输入该字段的数据为NULL ，就会报错。
AUTO_INCREMENT定义列为自增的属性，一般用于主键，数值会自动加1。
PRIMARY KEY关键字用于定义列为主键。 您可以使用多列来定义主键，列间以逗号分隔。
ENGINE 设置存储引擎，CHARSET 设置编码。

mysql> create table user(-> id INT UNSIGNED AUTO_INCREMENT,-> name VARCHAR(50) NOT NULL,-> age INT NOT NULL,-> sex ENUM('male','female') NOT NULL,-> PRIMARY KEY (id)-> )ENGINE=InnoDB DEFAULT CHARSET=utf8;

删除数据表格

DROP TABLE table_name ;

插入数据

INSERT INTO table_name ( field1, field2,...fieldN )VALUES( value1, value2,...valueN );

如果数据是字符型，必须使用单引号或者双引号，如：“value”。

查询数据

SELECT column_name,column_name
FROM table_name
[WHERE Clause]
[LIMIT N][ OFFSET M]

• 查询语句中可以使用一个或者多个表，表之间使用逗号(,)分割，并使用WHERE语句来设定查询条件。
• SELECT命令可以读取一条或者多条记录。
• 可以使用星号（*）来代替其他字段，SELECT语句会返回表的所有字段数据
• 可以使用 WHERE语句来包含任何条件。
• 可以使用 LIMIT属性来设定返回的记录数。
• 可以通过OFFSET指定SELECT语句开始查询的数据偏移量。默认情况下偏移量为0。

where语句

以下是 SQL SELECT 语句使用 WHERE 子句从数据表中读取数据的通用语法：

SELECT field1, field2,...fieldN FROM table_name1, table_name2...
[WHERE condition1 [AND [OR]] condition2.....

• 查询语句中你可以使用一个或者多个表，表之间使用逗号, 分割，并使用WHERE语句来设定查询条件。
• 可以在 WHERE子句中指定任何条件。
• 可以使用 AND 或者 OR 指定一个或多个条件。
• WHERE 子句也可以运用于 SQL 的 DELETE 或者 UPDATE 命令。
• WHERE 子句类似于程序语言中的 if 条件，根据 MySQL 表中的字段值来读取指定的数据。

查看表格状态

desc table_name;

样例如下：

mysql> desc user;
+-------+-----------------------+------+-----+---------+----------------+
| Field | Type                  | Null | Key | Default | Extra          |
+-------+-----------------------+------+-----+---------+----------------+
| id    | int unsigned          | NO   | PRI | NULL    | auto_increment |
| name  | varchar(50)           | NO   |     | NULL    |                |
| age   | int                   | NO   |     | NULL    |                |
| sex   | enum('male','female') | NO   |     | NULL    |                |
+-------+-----------------------+------+-----+---------+----------------+
4 rows in set (0.01 sec)

常见问题

error C4996: ‘sprintf’: This function or variable may be unsafe. Consider using sprintf_s instead. ，在VS中设置，项目属性 -> C/C++ -> 预处理器 -> 预处理定义上加上_CRT_SECURE_NO_DEPRECATE和_SCL_SECURE_NO_DEPRECATE，使用分号隔开。