首页 > 编程学习 > C++实现MySQL数据库连接池

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

发布时间:2022/5/14 18:51:59

ConnectionPool-MySQL

    • 项目背景
    • 功能点介绍
    • 关键技术点
    • MySQL数据库编程
    • 压力测试
    • 参考资料

项目背景

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

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

注:
目前市场上比较流行的连接池包括阿里的druidc3p0以及apache dbcp连接池,但他们有一个共同的特点:
都是拿Java实现的。

而本项目就是采用第二种方法:
基于C++实现数据库连接池

功能点介绍

一般的连接池都包含了数据库连接所用的ip地址、port端口号、username用户名、password密码以及其他一些性能参数:比如初始连接量、最大连接量、最大空闲时间、连接超时时间等,本项目重点实现上述通用功能:

初始连接量(initSize)
初始连接量表示连接池事先会和MySQL Server创建的initSize数量的Connection连接。在完成初始连接量之后,当应用发起MySQL访问时,不用创建新的MySQL Server连接,而是从连接池中直接获取一个连接,当使用完成后,再把连接归还到连接池中。
最大连接量(maxSize)
当并发访问MySQL Server的请求增加,初始连接量不够用了,此时会增加连接量,但是增加的连接量有个上限就是maxSIze。因为每一个连接都会占用一个socket资源,一般连接池和服务器都是部署在一台主机上,如果连接池的连接数量过多,那么服务器就不能响应太多的客户端请求了。
最大空闲时间(maxIdleTime)
当高并发过去,因为高并发而新创建的连接在很长时间(maxIdleTime)内没有得到使用,那么这些新创建的连接处于空闲,并且占用着一定的资源,这个时候就需要将其释放掉,最终只用保存iniSize个连接就行。
连接超时时间(connectionTimeOut)
当MySQL的并发访问请求量过大,连接池中的连接数量已经达到了maxSize,并且此时连接池中没有可以使用的连接,那么此时应用阻塞connectionTimeOut的时间,如果此时间内有使用完的连接归还到连接池,那么他就可以使用,如果超过这个时间还是没有连接,那么它获取数据库连接池失败,无法访问数据库。

关键技术点

  1. 单例模式:
    连接池只需要一个实例,故使用单例模式来实现,实现逻辑如下:
class ConnectionPool
{
public:
	// 获取连接池的对象实例
	static ConnectionPool* getConnectionPool();
};
// 线程安全的懒汉单例函数接口
ConnectionPool* ConnectionPool::getConnectionPool()
{
	// 当程序运行到此时才会生成对象实例
	static ConnectionPool pool; 
	return &pool;
}
  1. queue容器:
    连接池的数据结构是queue队列,最早生成的连接connection放在队头,此时记录一个起始时间,这一点在后面最大空闲时间时会发挥作用:如果队头都没有超过最大空闲时间,那么其他的一定没有。实现逻辑如下:
#include <queue>

class ConnectionPool
{
private:
	// 存储mysql连接的队列
	queue<Connection*> _connectionQue;  
};
// 创建初始数量的连接(构造函数初始化时)
for (int i = 0; i < _initSize; ++i)
{
	// 创建一个新的连接
	Connection* p = new Connection();
	p->connect(_ip, _port, _username, _password, _dbname);
	// 刷新一下开始空闲的起始时间
	p->refreshAliveTime();
	// 将这个新的连接放入连接池队列
	_connectionQue.push(p);
	// 连接数量加一
	_connectionCnt++;
}

// 连接数量没有达到上限,继续创建新的连接(并发量增大需要扩增连接时)
if (_connectionCnt < _maxSize)
{
	Connection* p = new Connection();
	p->connect(_ip, _port, _username, _password, _dbname);
	// 刷新一下开始空闲的起始时间
	p->refreshAliveTime();
	// 将这个新的连接放入连接池队列
	_connectionQue.push(p);
	// 连接数量加一
	_connectionCnt++;
}

// 扫描整个队列,释放多余的连接(高并发过后,新建的连接超过最大超时时间时)
unique_lock<mutex> lock(_queueMutex);
while (_connectionCnt > _initSize)
{
	Connection* p = _connectionQue.front();
	if (p->getAliveTime() >= (_maxIdleTime * 1000))
	{
		_connectionQue.pop();
		_connectionCnt--;
		// 调用~Connection()释放连接
		delete p;
	}
	else
	{
		// 如果队头的连接没有超过_maxIdleTime,其他连接肯定没有
		break;
	}
}
  1. C++11多线程编程:
    为了进行压力测试,用结果证明使用连接池之后对数据库的访问效率确实比不使用连接池的时候高很多,使用了多线程来进行数据库的访问操作,并且观察多线程下连接池对于性能的提升。实现逻辑如下:
#include <thread>

int main()
{
	thread t1([]() {
		for (int i = 0; i < 250; ++i)
		{	
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t2([]() {
		for (int i = 0; i < 250; ++i)
		{	
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t3([]() {
		for (int i = 0; i < 250; ++i)
		{	
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t4([]() {
		for (int i = 0; i < 250; ++i)
		{	
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});

	t1.join();
	t2.join();
	t3.join();
	t4.join();
	
	return 0;
}
  1. 线程互斥、线程同步通信、unique_lock:
    因为连接池涉及生产连接和消费使用连接,故他不是线程安全的,为了线程安全,我们引入线程互斥量(mutex)、线程同步通信以及唯一锁(unique_lock):实现逻辑如下:
#include <mutex>
#include <condition_variable>

class ConnectionPool
{
private:
	// 设置条件变量,用于连接生产线程和连接消费线程的通信
	condition_variable cv;				
	// 维护连接队列的线程安全互斥锁
	mutex _queueMutex;
};
for (;;)
{
	unique_lock<mutex> lock(_queueMutex);
	while (!_connectionQue.empty())
	{
		// 队列不为空,此处生产线程进入等待状态
		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();
}
  1. 基于CAS的原子整形:
    对于_connectionCnt(连接池内的连接数量)来说,是我们生产和消费线程工作都要参考的变量,所以这个变量必须保证其原子性,故我们在此使用C++提供的原子类:atomic_int,注意这个需要引入头文件#include <atomic>,实现逻辑如下:
#include <atomic>

class ConnectionPool
{
private:
	// 记录连接所创建的connection连接的总数量
	atomic_int _connectionCnt;			
};
// 生产新连接时:
_connectionCnt++;
// 当新连接超过最大超时时间后被销毁时
_connectionCnt--;
  1. shared_ptr:
    对于连接池中的每一个连接connection来说,用户只需要关心使用,去连接池里直接获取连接并且使用即可,并不用关心连接的生成和销毁,并且,对于使用完的连接来说,为了达到复用的目的,我们并不希望把它释放掉,而是使用完之后将其放回到连接池中供其他消费者使用,这个我们就可以使用智能指针来完成(用户不用关心其使用完之后的情况),不过我们需要修改其析构函数,使其重新放入连接池而不是释放掉:
    实现逻辑如下:
class ConnectionPool
{
public:
	// 给外部提供接口,从连接池中获取一个可用的空闲连接
	shared_ptr<Connection> getConnection();
};
/*
* shared_ptr智能指针析构时,会把connection资源直接delete掉,
* 相当于调用connection的析构函数,connection就被close掉了
* 这里需要自定义shared_ptr的释放资源的方式,把connection直接归还到queue中
*/
shared_ptr<Connection> sp(_connectionQue.front(),
	[&](Connection* pcon) {
		// 这里是在服务器应用线程中调用的,所以一定要考虑队列的线程安全操作
		unique_lock<mutex> lock(_queueMutex);
		pcon->refreshAliveTime();
		_connectionQue.push(pcon);
	});
  1. lambda表达式:
    我们在前面修改智能指针的析构函数时使用了lambda表达式,在这里就不继续做赘述了。
  2. 生产者-消费者线程模型:
    对于连接池来说,除了初始创建的连接外,高并发时需要生产新的连接、用户端需要对连接池里的连接进行消费,这个构成了我们的生产者-消费者线程模型,并且该模型和线程间通信、线程互斥等都是离不开的:
    主要实现如下:
#include <functional>

class ConnectionPool
{
public:
	// 给外部提供接口,从连接池中获取一个可用的空闲连接
	shared_ptr<Connection> getConnection();
private:
	// 运行在独立的线程中,专门负责生产新连接
	void produceConnectionTask();

	// 扫描超过maxIdleTime时间的空闲连接,进行多余的连接回收
	void scannerConnectionTask();

	queue<Connection*> _connectionQue;  // 存储mysql连接的队列
	mutex _queueMutex;					// 维护连接队列的线程安全互斥锁
	atomic_int _connectionCnt;			// 记录连接所创建的connection连接的总数量
	condition_variable cv;				// 设置条件变量,用于连接生产线程和连接消费线程的通信
};
// 启动一个新线程,作为连接的生产者,相当于Linux下的 pthread_create
thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));
produce.detach();

// 启动一个新的线程,扫描超过maxIdleTime时间的空闲连接,进行多余连接的回收
thread scanner(std::bind(&ConnectionPool::scannerConnectionTask, this));
scanner.detach();

MySQL数据库编程

我们都知道,要对MySQL数据库进行操作(增删改查),需要使用到SQL语句,那我们在C++下如何输入SQL语句来对其进行操作呢?
一个比较好的办法就是对其进行封装:
首先是头文件:

#include <mysql.h>
#include <string>
#include <ctime>
using namespace std;

/*
* 实现MySQL数据库操作
*/

class Connection
{
public:
	// 初始化数据库连接
	Connection();
	// 释放数据库连接资源
	~Connection();
	// 连接数据库
	bool connect(string ip,
		unsigned short port,
		string user,
		string password,
		string dbname);
	// 更新操作 insert、delete、update
	bool update(string sql);
	// 查询操作 select
	MYSQL_RES* query(string sql);

	// 刷新一下连接的起始的空闲时间点
	void refreshAliveTime() { _alivetime = clock(); }

	// 返回存活时间
	clock_t getAliveTime()const { return clock() - _alivetime; }
private:
	MYSQL* _conn; // 表示和MySQL Server的一条连接
	clock_t _alivetime; // 记录进入空闲状态后的起始存活时间
};

接着是实现文件:

#include "Connection.h"
using namespace std;

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;
}

bool Connection::update(string sql)
{
	// 更新操作 insert、delete、update
	if (mysql_query(_conn, sql.c_str()))
	{
		LOG("更新失败:" + sql);
		return false;
	}
	return true;
}

MYSQL_RES* Connection::query(string sql)
{
	// 查询操作 select
	if (mysql_query(_conn, sql.c_str()))
	{
		LOG("查询失败:" + sql);
		return nullptr;
	}
	return mysql_use_result(_conn);
}

注意:
在Windows常见的VS上使用数据库需要进行数据库的配置:
方法如下:
1、右键项目-C/C++ - 常规 - 附加包含目录,填写mysql.h头文件的路径:

在这里插入图片描述在这里插入图片描述在这里插入图片描述

2、右键项目 - 链接器 - 常规 - 附加库目录,填写libmysql.lib的路径:

在这里插入图片描述在这里插入图片描述

3、右键项目 - 链接器 - 输入 - 附加库目录,填写libmysql.lib库的名字:

在这里插入图片描述

4、把libmysql.dll动态链接库(Linux下后缀名是.so库)放在工程目录下:

在这里插入图片描述在这里插入图片描述在这里插入图片描述

压力测试

在完成了代码之后,我们来进行压力测试。我们一直在说,使用连接池之后访问数据库的效率得到了提高,那么事实究竟如何呢?我们用事实说话:
首先是未使用连接池:
数据量1000:

  • 单线程

测试代码如下:

int main()
{
	clock_t begin = clock();
	for (int i = 0; i < 1000; ++i)
	{
		Connection conn;
		char sql[1024] = { 0 };
		sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
			"zhang san", 20, "male");
		conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
		conn.update(sql);
	}
	clock_t end = clock();
	cout << (end - begin) << "ms" << endl;
	return 0;
}

运行结果:
在这里插入图片描述在这里插入图片描述

  • 四线程
    测试代码:
int main()
{
	Connection conn;
	conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
	clock_t begin = clock();

	thread t1([]() {
		for (int i = 0; i < 250; ++i)
		{
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t2([]() {
		for (int i = 0; i < 250; ++i)
		{
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t3([]() {
		for (int i = 0; i < 250; ++i)
		{
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});
	thread t4([]() {
		for (int i = 0; i < 250; ++i)
		{
			Connection conn;
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			conn.connect("127.0.0.1", 3306, "root", "31777", "chat");
			conn.update(sql);
		}
		});

	t1.join();
	t2.join();
	t3.join();
	t4.join();

	clock_t end = clock();
	cout << (end - begin) << "ms" << endl;
	return 0;
}

运行结果:
在这里插入图片描述在这里插入图片描述

接下来使用连接池:
数据量1000:

  • 单线程
    测试代码:
int main()
{
	clock_t begin = clock();
	ConnectionPool* cp = ConnectionPool::getConnectionPool();
	for (int i = 0; i < 1000; ++i)
	{
		shared_ptr<Connection> sp = cp->getConnection();
		char sql[1024] = { 0 };
		sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
			"zhang san", 20, "male");
		sp->update(sql);
	}
	
	clock_t end = clock();
	cout << (end - begin) << "ms" << endl;
	return 0;
}

运行结果:
在这里插入图片描述在这里插入图片描述

  • 四线程
    测试代码:
int main()
{
	clock_t begin = clock();

	thread t1([]() {
		ConnectionPool *cp = ConnectionPool::getConnectionPool();
		for (int i = 0; i < 250; ++i)
		{
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			shared_ptr<Connection> sp = cp->getConnection();
			sp->update(sql);
		}
		});
	thread t2([]() {
		ConnectionPool *cp = ConnectionPool::getConnectionPool();
		for (int i = 0; i < 250; ++i)
		{
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			shared_ptr<Connection> sp = cp->getConnection();
			sp->update(sql);
		}
		});
	thread t3([]() {
		ConnectionPool *cp = ConnectionPool::getConnectionPool();
		for (int i = 0; i < 250; ++i)
		{
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			shared_ptr<Connection> sp = cp->getConnection();
			sp->update(sql);
		}
		});
	thread t4([]() {
		ConnectionPool *cp = ConnectionPool::getConnectionPool();
		for (int i = 0; i < 250; ++i)
		{
			char sql[1024] = { 0 };
			sprintf(sql, "insert into user(name,age,sex) values('%s',%d,'%s')",
				"zhang san", 20, "male");
			shared_ptr<Connection> sp = cp->getConnection();
			sp->update(sql);
		}
		});

	t1.join();
	t2.join();
	t3.join();
	t4.join();

	clock_t end = clock();
	cout << (end - begin) << "ms" << endl;
	return 0;
}

运行结果:
在这里插入图片描述在这里插入图片描述

综上,汇总表格如下:

数据量未使用连接池使用连接池
1000单线程:2366ms 四线程:830ms单线程:1563ms 四线程:635ms
5000单线程:11704ms 四线程:4133ms单线程:6887ms 四线程:3055ms
10000单线程:24036ms 四线程:7925ms单线程:14303ms 四线程:6155ms

可以看到使用连接池后,不管是单线程还是多线程都对数据库的访问速度提升了不少。

参考资料

【1】施磊.MySQL数据库连接池.腾讯课堂.2020.11

Copyright © 2010-2022 ngui.cc 版权所有 |关于我们| 联系方式| 豫B2-20100000