MySQL 数据库连接池

关键技术点：

  • MySQL 数据库编程
  • 单例模式
  • queue队列容器
  • C++11多线程编程、线程互斥、线程同步通信和 unique_lock
  • 基于CAS的原子整形
  • 智能指针shared_ptr
  • lambda表达式
  • 生产者-消费者线程模型

项目背景

为了提高 MySQL 数据库（基于C/S设计）的访问瓶颈，除了在服务器端增加缓存服务器缓存常用的数据之外（例如：redis），还可以增加连接池，来提高 MySQL Server 的访问效率，在高并发情况下，大量的 TCP 三次握手、 MySQL Server 连接认证、MySQL Server 关闭连接回收资源和 TCP 四次挥手所耗费的性能时间也是很明显的，增加连接池就是为了减少这一部分的性能损耗。

而常见的解决数据库访问瓶颈的方法有两种：

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

注：

在市场上比较流行的连接池包括阿里的 druid，c3p0 以及 apache dbcp 连接池，它们对于短时间内大量的数据库增删改查操作性能的提升是很明显的，但是它们有一个共同点就是，全部由 Java 实现的。

那么本项目就是为了在 C/C++ 项目中，提供 MySQL Server 的访问效率，实现基于 C++ 代码的数据库连接 池模块。

功能介绍

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

  1. 初始连接量（iniSize）：表示连接池实现和 MySQL Server 创建 iniSize 个数的 connection 连接，当应用发起 MySQL 访问时，不用再创建和 MySQL Server 新的连接，直接从连接池中获取一个可用的连接即可，使用完成后，并不去释放 connection，而是把当前 connection 在归还到连接池中。
  2. 最大量接量（maxSize）：当并发访问 MySQL Server 的请求增多时，初始连接量已经不够使用时，此时会根据新的请求数量创建更多的连接给应用去使用，但是新创建的连接数量上限是 maxSize，不能无限制的创建连接，因为每一个连接都会占用一个 socket 资源，一般连接池和服务器程序是部署在一台主机上的，如果连接池占用过多 socket 资源，那么服务器就不能接受太多的客户端请求了。当这些连接使用完成后，再次归还到连接池中来维护。
  3. 最大空闲时间（maxIdleTime）：当访问 MySQL 的并发请求多了以后，连接池里面的连接数量会动态增加，上限是 maxSize 个，当这些连接用完再次归还到连接池当中。如果在指定的 maxIdleTime 里面，这些新增加的连接都没有被再次使用过，那么新增加的这些连接资源就要被回收掉，只需要保持初始连接量 initSize 个连接就可以了。
  4. 连接超时时间（connectionTimeout）：当 MySQL 的并发请求量过大，连接池中的连接数量已经到达 maxSize 了，而此时没有空闲的连接可供使用，那么此时应用从连接池获取连接无法成功，它通过阻塞的方式获取连接的时间如果超过 connectionTimeout 时间，那么获取连接失败，无法访问数据库。

MySQL Server 参数介绍

1
show variables like 'max_connections';

该命令可以查看 MySQL Server 所支持的最大连接个数，超过 max_connections 数量的连接，MySQL Server 会直接拒绝，所以在使用连接池增加连接数量的时候，MySQL Server 的 max_connections 参数也要适当的进行调整，以适配连接池的连接上限。

功能实现设计

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. 连接的生产和连接的消费采用生产者-消费者线程模型来设计，使用了线程间的同步通信机制条件变量和互斥锁

开发平台选型

有关 MySQL 数据库编程、多线程编程、线程互斥和同步通信操作、智能指针、设计模式、容器等等这些技术在 C++ 语言层面都可以直接实现，因此该项目选择直接在 windows 平台上进行开发，当然放在 Linux 平台下用 g++ 也可以直接编译运行。

压力测试

验证数据的插入操作所花费的时间，第一次测试使用普通的数据库访问操作，第二次测试使用带连接池的数据库访问操作，对比两次操作同样数据量所花费的时间，性能压力测试结果如下：

数据量 未使用连接池花费时间 使用连接池花费时间
1000 单线程： 四线程： 单线程： 四线程：
5000 单线程： 四线程： 单线程： 四线程：
10000 单线程： 四线程： 单线程： 四线程：

MySQL 数据库编程

这里的 MySQL 数据库编程直接采用 oracle 公司提供的 MySQL C/C++ 客户端开发包，在 VS 上需要进行相应的头文件和库文件的配置，如下：

  1. 右键项目 - C/C++ - 常规 - 附加包含目录，填写mysql.h头文件的路径

找到 mysql.h 目录文件

image.png

在 VS 编译器设置中的 C/C++ - 常规 中包含该目录

image.png

  1. 右键项目 - 链接器 - 常规 - 附加库目录，填写libmysql.lib的路径

找到 libmysql.lib 根目录

image.png

在 VS 编译器设置中的 链接器 - 常规 中包含该目录

image.png

  1. 右键项目 - 链接器 - 输入 - 附加依赖项，填写libmysql.lib库的名字

image.png

  1. 把libmysql.dll动态链接库（Linux下后缀名是.so库）放在工程目录下

image.png

代码实现

1. 配置 MySQL

Connection.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
#pragma once
#include <mysql.h>
#include <string>
// 引入定义文件
#include "Common.h"
// 定义命名空间
using namespace std;

/*
	实现 MySQL 数据库的增删改查
*/

class Connection
{
public:
	// 定义数据库连接
	Connection();
	// 定义析构函数
	~Connection();
	// 连接数据库
	bool connect(
		string ip,				// ip 地址
		unsigned short port,	// 端口
		string user,			// 用户名
		string password,		// 密码
		string dbname			// 数据库名
	);

	// 更新操作 insert、delete、update
	bool update(string sql);
	// 查询操作 select
	MYSQL_RES* query(string sql);

private:
	MYSQL* _conn; // 表示和MySQL Server的一条连接
};

Connection.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#include "Connection.h"
#include "Common.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);
}

Common.h

1
2
3
4
5
6
7
#pragma once
#include <iostream>

// 定义公共日志
#define LOG(str) \
	cout << __FILE__ << ": " << __LINE__ << " " \
	__TIMESTAMP__ << " : " << str << endl;

main.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#include <iostream>
#include "Connection.h"
using namespace std;

int main()
{
	Connection con;
	char sql[1024] = { 0 };
	// 创建 sql 语句
	sprintf(sql, "insert into users(name, age, sex) values('%s', '%d', '%s')",
		"zs", 20, "male");
	// 连接
	con.connect("127.0.0.1", 3306, "root", "root", "chat");
	con.update(sql);

	return 0;
}

2. 连接池单例实现

Connection.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#pragma once
/*
	连接池功能模块
*/

class ConnectionPool
{
public:
	// 获取连接池对象实例
	static ConnectionPool* getConnectionPool();

private:
	// 单例 #1 -- 构造函数私有化 
	ConnectionPool();
};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
#include "ConnectionPool.h"

// 线程安全的懒汉单例模式
ConnectionPool* ConnectionPool::getConnectionPool()
{
	// 静态类型初始化，编译器自动 lock 和 unlock
	static ConnectionPool pool;
	return &pool;
}

3. 加载配置文件实现

mysql.ini

1
2
3
4
5
6
7
8
9
10
11
12
# 数据库连接池配置文件
ip=127.0.0.1
port=3306
username=root
password=root
dbname=chat
initSize=10
maxSize=1024
# 最大空闲时间 单位为：s/秒
maxIdleTime=60
# 连接超时时间 单位为：ms/毫秒
connectionTimeout=100

Connection.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#pragma once
#include"Connection.h"
#include<queue>
#include<mutex>
#include<atomic>
#include<functional>
/*
	连接池功能模块
*/

class ConnectionPool
{
public:
	// 获取连接池对象实例
	static ConnectionPool* getConnectionPool();

private:
	// 单例 #1 -- 构造函数私有化 
	ConnectionPool();

	// 加载方法
	// 1. 获取配置文件中相关数据库信息
	bool loadConfigFile();

	// 定义所需数据
	// 1. mysql 的 ip 地址
	string _ip;
	// 2. mysql 的端口号 3306
	unsigned short _port;
	// 3. mysql 登陆用户名
	string _username;
	// 4. mysql 登陆密码
	string _password;
	// 5. 连接的数据库名称
	string _dbname;
	// 6. 连接池的初始连接量
	int _initSize;
	// 7. 连接池的最大量接量
	int _maxSize;
	// 8. 连接池的最大空闲时间
	int _maxIdleTime;
	// 9. 连接池获取连接的超时时间
	int _connectionTimeout;

	// 1. 存储 mysql 连接队列
	queue<Connection*> _connectionQue;
	// 2. 维护连接队列的线程安全互斥锁
	mutex _queueMutex;
	// 3. 记录连接所创建的 connection 连接的总数量 
	atomic_int _connectionCnt;
};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
#include "ConnectionPool.h"

bool ConnectionPool::loadConfigFile()
{
	// 读取方式打开文件
	FILE* pf = fopen("mysql.ini", "r");
	// 判断文件是否存在
	if (pf == nullptr)
	{
		LOG("mysql.ini file is not exist!");
		return false;
	}

	// 遍历文件内容
	while (!feof(pf))
	{
		// 定义保存字段
		char line[1024] = { 0 };
		// 按行读取并赋值
		fgets(line, 1024, pf);
		string str = line;
		// 查找
		int idx = str.find('=', 0);
		// 无效的配置项
		if (idx == -1)
		{
			continue;
		}
		
		// 获取结尾 \n
		//    0     1   2     3
		// password = 123456 \n
		int endidx = str.find('\n', idx);
		// 获取索引 0 的数据
		string key = str.substr(0, idx);
		// 获取索引 2 的数据
		string value = str.substr(idx + 1, endidx - 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());
		}
	}
}

// 构造
ConnectionPool::ConnectionPool()
{
	// 加载配置项
	if (!loadConfigFile())
	{
		return;
	}
	// 创建初始数量的连接
	int i = 0;
	for (i = 0; i < _initSize; i++)
	{
		// 实例化连接
		Connection* p = new Connection();
		// 调用 connection
		p->connect(_ip, _port, _username, _password, _dbname);
		// 添加到队列中
		_connectionQue.push(p);
		// 连接量增加
		_connectionCnt++;
	}
}

4. 创建链接的生产者线程

ConnectionPool.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
#pragma once
#include "Connection.h"
#include <queue>
#include <mutex>
#include <thread>
#include <atomic>
#include <memory>
#include <functional>
#include <condition_variable>
/*
	连接池功能模块
*/

class ConnectionPool
{
public:
	// 获取连接池对象实例
	static ConnectionPool* getConnectionPool();

private:
	// 单例 #1 -- 构造函数私有化 
	ConnectionPool();

	// 加载方法
	// 2. 专门负责生产新连接（C 语言），运行在独立的线程中
	// 普通方法运行在成员对象需要把 this 指针进行绑定
	void produceConnectionTask();

	// 4. 设置条件变量，用于连接生产线程和连接消费线程的通信
	condition_variable cv;
};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#include "ConnectionPool.h"
#include "Common.h"

// 构造
ConnectionPool::ConnectionPool()
{
	// 加载配置项
	// ----

	// 启动一个新的线程，作为连接的生产者 linux thread => pthread_create
	// 需要一个线程函数，因为连接创建的是 C 接口函数，无法直接进行创建，所以创建了 produceConnectionTask
	thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));
}

// 运行在独立的线程中，专门负责生产新连接
void ConnectionPool::produceConnectionTask()
{
	// 循环，目的是反复查看
	for (;;)
	{
		// 创建互斥锁
		unique_lock<mutex> lock(_queueMutex);
		// 判断 连接队列是否为空
		if (!_connectionQue.empty())
		{
			// 不为空，则进入生产者线程中进入等待状态
			cv.wait(lock);
		}
		// 连接数量没有达到上限，继续创建新的连接
		if (_connectionCnt < _maxSize)
		{
			// 实例化连接
			Connection* p = new Connection();
			// 调用 connection
			p->connect(_ip, _port, _username, _password, _dbname);
			// 添加到队列中
			_connectionQue.push(p);
			// 连接量增加
			_connectionCnt++;
		}

		// 通知消费者线程，可以消费连接了
		cv.notify_all();
	}
}

5. 创建连接的消费者线程

ConnectionPool.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#pragma once
#include "Connection.h"
#include <queue>
#include <mutex>
#include <thread>
#include <atomic>
#include <memory>
#include <functional>
#include <condition_variable>
/*
	连接池功能模块
*/

class ConnectionPool
{
public:
	// 通过智能指针给外部提供接口，从连接池中获取一个可用的闲置连接
	// 智能指针会自动析构，所以无需在进行释放。
	shared_ptr<Connection> getConnection();

};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
// 给外部提供接口，从连接池中获取一个可用的空闲连接
shared_ptr<Connection> ConnectionPool::getConnection()
{ 
	// 互斥锁
	unique_lock<mutex> lock(_queueMutex);
	// 判断为空
	if (_connectionQue.empty())
	{
		// 等待线程
		/*
			问题：
			当在等在中时，有连接唤醒了队列，在等待中被其他线程抢走了，那么队列依旧为空
			在 wait_for 中有两个状态：
			- no_timeout：未超时唤醒
			- timeout：   超时唤醒
			所以这里需要进行一个判断
		*/
		if (cv_status::timeout == cv.wait_for(lock, chrono::milliseconds(_connectionTimeout)))
		{
			if (_connectionQue.empty())
			{
				LOG("获取空闲连接超时了...获取连接失败!");
				return nullptr;
			}
		}
	}
	/*
	shared_ptr智能指针析构时，会把connection资源直接delete掉，相当于
	调用connection的析构函数，connection就被close掉了。
	这里需要自定义shared_ptr的释放资源的方式，把connection直接归还到queue当中
	*/
	shared_ptr<Connection> sp(_connectionQue.front(),
		[&](Connection* pcon) {
			// 这里是在服务器应用线程中调用的，所以一定要考虑队列的线程安全操作
			unique_lock<mutex> lock(_queueMutex);
			// 回到队列中
			_connectionQue.push(pcon);
		});

	// 出队消费
	_connectionQue.pop();
	// 消费完连接以后，通知生产者线程检查一下，如果队列为空了，赶紧生产连接
	cv.notify_all();  

	return sp;
}

6. 最大空闲时间

Connection.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#pragma once
#include <ctime>
// 定义命名空间
using namespace std;

/*
	实现 MySQL 数据库的增删改查
*/

class Connection
{
public:
	// 刷新一下连接的起始的空闲时间点
	void refreshAliveTime() { _alivetime = clock(); }
	// 返回存活的时间
	clock_t getAliveeTime() const { return clock() - _alivetime; }

private:
	// 记录进入空闲状态后的起始存活时间
	clock_t _alivetime;
};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
// 构造
ConnectionPool::ConnectionPool()
{
	// 加载配置项
	if (!loadConfigFile())
	{
		return;
	}
	// 创建初始数量的连接
	int i = 0;
	for (i = 0; i < _initSize; i++)
	{
		// 实例化连接
		Connection* p = new Connection();
		// 调用 connection
		p->connect(_ip, _port, _username, _password, _dbname);
		// 刷新空闲起始时间
		p->refreshAliveTime();
		// 添加到队列中
		_connectionQue.push(p);
		// 连接量增加
		_connectionCnt++;
	}

	// 启动一个新的线程，作为连接的生产者 linux thread => pthread_create
	// 需要一个线程函数，因为连接创建的是 C 接口函数，无法直接进行创建，所以创建了 produceConnectionTask
	thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));
}

// 运行在独立的线程中，专门负责生产新连接
void ConnectionPool::produceConnectionTask()
{
	// 循环，目的是反复查看
	for (;;)
	{
		// 创建互斥锁
		unique_lock<mutex> lock(_queueMutex);
		// 判断 连接队列是否为空
		if (!_connectionQue.empty())
		{
			// 不为空，则进入生产者线程中进入等待状态
			cv.wait(lock);
		}
		// 连接数量没有达到上限，继续创建新的连接
		if (_connectionCnt < _maxSize)
		{
			// 实例化连接
			Connection* p = new Connection();
			// 调用 connection
			p->connect(_ip, _port, _username, _password, _dbname);
			// 刷新空闲起始时间
			p->refreshAliveTime();
			// 添加到队列中
			_connectionQue.push(p);
			// 连接量增加
			_connectionCnt++;
		}

		// 通知消费者线程，可以消费连接了
		cv.notify_all();
	}
}

// 给外部提供接口，从连接池中获取一个可用的空闲连接
shared_ptr<Connection> ConnectionPool::getConnection()
{ 
	// 互斥锁
	unique_lock<mutex> lock(_queueMutex);
	// 判断为空
	if (_connectionQue.empty())
	{
		// 等待线程
		/*
			问题：
			当在等在中时，有连接唤醒了队列，在等待中被其他线程抢走了，那么队列依旧为空
			在 wait_for 中有两个状态：
			- no_timeout：未超时唤醒
			- timeout：   超时唤醒
			所以这里需要进行一个判断
		*/
		if (cv_status::timeout == cv.wait_for(lock, chrono::milliseconds(_connectionTimeout)))
		{
			if (_connectionQue.empty())
			{
				LOG("获取空闲连接超时了...获取连接失败!");
				return nullptr;
			}
		}
	}
	/*
	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);
		});

	// 出队消费
	_connectionQue.pop();
	// 消费完连接以后，通知生产者线程检查一下，如果队列为空了，赶紧生产连接
	cv.notify_all();  

	return sp;
}

7. 回收超时连接线程

ConnectionPool.h

1
2
3
4
5
6
7
8
9
10
11
12
13
/*
	连接池功能模块
*/

class ConnectionPool
{
public:
	// --------

private:
	// 3. 扫描超过maxIdleTime时间的空闲连接，进行对于的连接回收
	void scannerConnectionTask();
};

ConnectionPool.cpp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
// 构造
ConnectionPool::ConnectionPool()
{
	// 加载配置项
	// -------
	
	// 启动一个新的线程，作为连接的生产者 linux thread => pthread_create
	// 需要一个线程函数，因为连接创建的是 C 接口函数，无法直接进行创建，所以创建了 produceConnectionTask
	thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));
	produce.detach();

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

// 扫描超过maxIdleTime时间的空闲连接，进行对于的连接回收 scannerConnectionTask
void ConnectionPool::scannerConnectionTask()
{
	// 循环扫描
	for (;;)
	{
		// 使用 sleep 模拟定时效果
		this_thread::sleep_for(chrono::seconds(_maxIdleTime));

		// 扫描整个队列，释放多余的连接
		unique_lock<mutex> lock(_queueMutex);
		// 循环遍历
		while (_connectionCnt > _initSize)
		{
			// 指向队列第一个
			Connection* p = _connectionQue.front();
			// 判断
			if (p->getAliveeTime() >= (_maxIdleTime * 1000))
			{
				// 出队
				_connectionQue.pop();
				// 数量递减
				_connectionCnt--;
				// 调用~Connection()释放连接
				delete p;
			}
			else
			{
				// 队头的连接没有超过_maxIdleTime，其它连接肯定没有
				break;
			}
		}
	}
}

连接池压力测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
#include <iostream>
#include "Connection.h"
#include "ConnectionPool.h"
using namespace std;

//int main()
//{
//	/*
//	*  测试 数据库连接、添加功能
//	Connection con;
//	char sql[1024] = { 0 };
//	// 创建 sql 语句
//	sprintf(sql, "insert into users(name, age, sex) values('%s', '%d', '%s')",
//		"zs", 20, "male");
//	// 连接
//	con.connect("127.0.0.1", 3306, "root", "root", "chat");
//	con.update(sql);
//	*/
//
//	// 测试配置文件
//	ConnectionPool* cp = ConnectionPool::getConnectionPool();
//	//cp->loadConfigFile();
//
//	return 0;
//}

int main() {
	clock_t begin = clock();
	ConnectionPool* p = ConnectionPool::getConnectionPool();

	thread t1([]() {
		for (int i = 0; i < 250; i++) {
			/*Connection conn;
			char sql[1024];
			sprintf(sql," insert into user values(NULL, '%d', '%d')",i,i);
			conn.connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn.update(sql);*/

			ConnectionPool* p = ConnectionPool::getConnectionPool();
			shared_ptr<Connection> conn = p->getConnection();
			char sql[1024];
			sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
			conn->connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn->update(sql);
		}
		
		});
	thread t2([]() {
		for (int i = 250; i < 500; i++) {
			/*Connection conn;
			char sql[1024];
			sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
			conn.connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn.update(sql);*/

			ConnectionPool* p = ConnectionPool::getConnectionPool();
			shared_ptr<Connection> conn = p->getConnection();
			char sql[1024];
			sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
			conn->connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn->update(sql);
		}
		
		});
	thread t3([]() {
		for (int i = 500; i < 750; i++) {
			/*	Connection conn;
				char sql[1024];
				sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
				conn.connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
				conn.update(sql);*/

			ConnectionPool* p = ConnectionPool::getConnectionPool();
			shared_ptr<Connection> conn = p->getConnection();
			char sql[1024];
			sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
			conn->connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn->update(sql);
		}
		});
		
	thread t4([]() {
		for (int i = 750; i < 1000; i++) {
			/*	Connection conn;
				char sql[1024];
				sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
				conn.connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
				conn.update(sql);*/

			ConnectionPool* p = ConnectionPool::getConnectionPool();
			shared_ptr<Connection> conn = p->getConnection();
			char sql[1024];
			sprintf(sql, " insert into user values(NULL, '%d', '%d')", i, i);
			conn->connect("127.0.0.1", 3306, "root", "root", "javaweb_login");
			conn->update(sql);

		}
		});

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

	clock_t end = clock();
	cout << end - begin << "ms" << endl;
	getchar();
}
#C++ #MySQL #对象池
MySQL 数据库连接池
http://example.com/2023/08/24/05.C++项目笔记/01.MySQL数据库连接池/实现笔记/
Author
Yakumo
Posted on
August 24, 2023
Licensed under