• 数据库连接池

    前言

            Mysql数据库是一个C/S模型，通过客户端来访问服务端。底层使用的是TCP协议。

            在连接数据库和断开数据库时，需要进行一下几个步骤：

    1. 建立通信连接的TCP三次握手。
    2. 数据库服务器的连接认证。
    3. 数据库服务器关闭连接时的资源回收。
    4. 断开通信连接的TCP四次挥手。

            当在我们在程序中需要频繁的访问数据库，也就是需要频繁的连接数据库和断开数据库时，会需要频繁的进行上面的操作。在高并发的情况下会消耗很多时间和资源。

            使用连接池可以减少这部分的消耗。指挥在创建连接池时，建立连接，在销毁连接池对象时，断开连接。需要连接数据库时，在连接池中取出连接，不需要使用数据库时，将连接放会连接池即可。

    连接池的实现

            1. 技术点

    • 多线程编程
    • 线程同步与互斥。互斥锁和条件变量的使用。
    • 处理时间和日期的chrono库。
    • 智能指针
    • lambda表达式
    • MySql C语言库的使用。
    • 单例模式
    • 生产者消费者模型
    • Jsoncpp库的使用

            2. 连接池的设计

            在网络通信中，服务器可能收到多个客户端的请求。连接池部署在服务器上，并且服务端会部署各种组件，其中数据库组件可能部署在另外一台服务器上，需要操作数据库，就需要连接另外一台服务器。

            使用线程池和连接池可以大大的提高效率。

            多线程，可以用来处理不用的任务，比如：接受客户端的请求，处理客户端的请求，从连接池中取出连接等。

            对于连接池，首先在连接池中，与数据库建立若干个连接，需要使用连接数据库时，从连接池中取，不需要连接时，将连接放入连接池中。

     实现的功能点：

    • 连接池只需要一个实例，所以连接池是一个单例模式的类。
    • 所有数据库连接需要维护到一个安全的队列中
      • 方便数据库的连接的添加和删除
      • 由于连接池是只有一个实例，所以需要是线程安全的，需要使用互斥锁来保护队列数据的压入和弹出。
    • 在需要的时候可以从数据库中得到一个或者多个可用的数据库连接。
      • 如果有可用连接直接取出
      • 如果没有，阻塞等待一段时间，在重试
    • 如果队列中没有多余可用连接，需要动态创建。不能超过最大连接数。
    • 如果队列中空闲连接太多，需要动态销毁一部分。
    • 数据库操作完毕，需要将连接归还到连接池中。

    细节分析：

    • 数据库连接的存储：使用生产者消费者模型，将连接保存到队列中。
      • 生产者：专门生产数据库连接的线程。
      • 消费者：需要访问数据库的线程。
      • 处理生产者消费者模型，需要通过条件变量来使用线程的同步。
    • 连接池连接的动态创建：交给一个单独的线程来处理。
    • 连接池连接的动态销毁：交给一个动态的线程处理。
    • 连接池的默认连接数量：连接池中提供了用连接的最小数量。如果不够，就动态创建，如果太多就动态销毁。
    • 连接池的最大连接数量：能够创建的最大有效数据库的连接数量。
    • 最大空闲时间：创建的数据库连接在指定的一定时间内，未被使用，需要进行销毁。
    • 连接超时：当连接池中没有可用连接，消费者线程无法获取连接，阻塞等待的时间。

            3. 封装MySql API的类

            释放数据库的资源，需要释放创建数据库的资源和结果集的资源。

            使用的C语言MySql接口：用C语言API(常用)操作MySql数据库_两片空白的博客-CSDN博客

    1. #pragma once
    2.  
    3.  
    4. #include 
    5. #include 
    6. #include 
    7. using namespace std;
    8. using namespace chrono;
    9.  
    10. class MysqlConn
    11. {
    12. public:
    13. 	//初始化数据库
    14. 	MysqlConn();
    15. 	//释放数据库资源
    16. 	~MysqlConn();
    17. 	//连接数据库
    18. 	bool connect(string user, string passwd, string dbName, string ip, unsigned short port = 3306);
    19. 	//更新数据库 update delete insert
    20. 	bool update(string sql);
    21. 	//查询数据库
    22. 	bool query(string sql);
    23. 	//遍历数据库，主要是查找结果集中的下一行数据
    24. 	bool next();
    25. 	//得到结果集中需要行的index列的字段
    26. 	string value(int index);
    27. 	//事务操作
    28. 	bool transaction();
    29. 	//提交事务
    30. 	bool commit();
    31. 	//事务回滚
    32. 	bool roolback();
    33. 	//刷新连接等待时间
    34. 	void refreshAliveTime();
    35. 	//获得连接等待时间
    36. 	long long getAliveTime();
    37. private:
    38. 	//释放结果集的资源
    39. 	void releaseRes();
    40.  
    41. 	MYSQL* m_conn = nullptr;
    42. 	MYSQL_RES* m_res = nullptr;
    43. 	MYSQL_ROW m_row = nullptr;
    44. 	steady_clock::time_point m_aliveTime;
    45.  
    46. };
    1. #include "MysqlConn.h"
    2.  
    3. MysqlConn::MysqlConn()
    4. {
    5. 	m_conn = mysql_init(nullptr);
    6. }
    7.  
    8. MysqlConn::~MysqlConn()
    9. {
    10. 	if (m_conn)
    11. 	{
    12. 		mysql_close(m_conn);
    13. 	}
    14. 	releaseRes();
    15. }
    16.  
    17. bool MysqlConn::connect(string user, string passwd, string dbName, string ip, unsigned short port)
    18. {
    19. 	MYSQL* ptr = mysql_real_connect(m_conn, ip.c_str(), user.c_str(), passwd.c_str(), dbName.c_str(), port, nullptr, 0);
    20. 	if (ptr)
    21. 	{
    22. 		return true;
    23. 	}
    24. 	return false;
    25. }
    26.  
    27. bool MysqlConn::update(string sql)
    28. {
    29. 	if (mysql_query(m_conn, sql.c_str()))
    30. 	{
    31. 		return false;
    32. 	}
    33. 	return true;
    34. }
    35.  
    36. bool MysqlConn::query(string sql)
    37. {
    38. 	//在查询结果集时，需要清空上一次结果集的内存
    39. 	releaseRes();
    40. 	if (mysql_query(m_conn, sql.c_str()) == 0)
    41. 	{
    42. 		m_res = mysql_store_result(m_conn);
    43. 		return true;
    44. 	}
    45. 	return false;
    46. }
    47.  
    48. bool MysqlConn::next()
    49. {
    50. 	if (m_res != nullptr)
    51. 	{
    52. 		m_row = mysql_fetch_row(m_res);
    53. 		if (m_row)
    54. 		{
    55. 			return true;
    56. 		}
    57. 	}
    58. 	return false;
    59. }
    60.  
    61. //参数时这一行的第几列
    62. string MysqlConn::value(int index)
    63. {
    64. 	//得到当前行中，列的数量
    65. 	int resNum = mysql_num_fields(m_res);
    66. 	if (index < 0 || index >= resNum)
    67. 	{
    68. 		return string();
    69. 	}
    70.  
    71. 	char* str = m_row[index];
    72. 	unsigned long length = mysql_fetch_lengths(m_res)[index];//获得index列字段的长度
    73. 	return string(str, length);
    74. }
    75.  
    76. bool MysqlConn::transaction()
    77. {
    78. 	return mysql_autocommit(m_conn, false);
    79. }
    80.  
    81. bool MysqlConn::commit()
    82. {
    83. 	return mysql_commit(m_conn);
    84. }
    85.  
    86. bool MysqlConn::roolback()
    87. {
    88. 	return mysql_rollback(m_conn);
    89. }
    90.  
    91. void MysqlConn::refreshAliveTime()
    92. {
    93. 	m_aliveTime = steady_clock::now();
    94. }
    95.  
    96. long long MysqlConn::getAliveTime()
    97. {
    98. 	nanoseconds res = steady_clock::now() - m_aliveTime;
    99. 	milliseconds millsec = duration_cast(res);
    100. 	return millsec.count();
    101. }
    102.  
    103. void MysqlConn::releaseRes()
    104. {
    105. 	if (m_res)
    106. 	{
    107. 		mysql_free_result(m_res);
    108. 	}
    109. }

      数据库连接池类

    • 使用队列来保存个连接，创建minSize个连接。
    • 使用两个单独的线程，一个用来创建连接，一个用来销毁连接，使队列中的线程个数维持在minSize个。
    • 队列是临界资源，pop和push时需要使用互斥锁加锁。
    • 队列为空，针对消费者和队列连接个数等于minSize，针对生产者都需要时用条件变量等待。当生产者生产连接到队列，消费者消费连接，需要唤醒消费者和生产者。
    • 当使用连接个数大于maxSize，也需要使用条件变量等待。
    • 销毁连接，当连接个数大于minSize，并且连接空闲时长大于maxIdleTime。
    1. #pragma once
    2.  
    3. #include "MysqlConn.h"
    4. #include 
    5. #include 
    6. #include 
    7.  
    8. class ConnectionPool
    9. {
    10. public:
    11. 	ConnectionPool* getConnectionPool();
    12. 	ConnectionPool(const ConnectionPool& pool) = delete;
    13. 	ConnectionPool& operator=(const ConnectionPool& pool) = delete;
    14. 	shared_ptr getConnection();
    15. private:
    16. 	ConnectionPool();
    17. 	~ConnectionPool();
    18. 	bool parseJsonFile();
    19. 	void productConnection();
    20. 	void recycleConnection();
    21. 	void addConnection();
    22.  
    23. 	string m_ip;
    24. 	string m_userName;
    25. 	string m_passwd;
    26. 	string m_dbName;
    27. 	unsigned short m_port;
    28.  
    29. 	int m_maxSize;//最大使用连接数
    30. 	int m_minSize;//连接池中最少连接数
    31. 	int m_timeout;
    32. 	int m_maxIdleTime;
    33. 	int m_useCount;//使用的连接数
    34. 	queue m_pool;
    35. 	mutex m_lock;
    36. 	condition_variable m_cond;
    37. };
    1. #include "ConnectionPool.h"
    2. #include 
    3. #include 
    4. #include 
    5.  
    6. using namespace Json;
    7. ConnectionPool::ConnectionPool()
    8. {
    9. 	//读取配置
    10. 	if (!parseJsonFile())
    11. 	{
    12. 		return;
    13. 	}
    14.  
    15. 	//生成连接
    16. 	for (int i = 0; i < m_minSize; ++i)
    17. 	{
    18. 		addConnection();
    19. 	}
    20.  
    21. 	//判断连接个数，是否需要创建连接或者销毁连接
    22. 	//单独用两个线程处理，处理的是连接池中的连接数
    23. 	thread produtor(&ConnectionPool::productConnection, this);
    24. 	thread recyclor(&ConnectionPool::recycleConnection, this);
    25. 	produtor.detach();
    26. 	recyclor.detach();
    27. }
    28.  
    29. ConnectionPool::~ConnectionPool()
    30. {
    31. 	while (!m_pool.empty())
    32. 	{
    33. 		MysqlConn* conn = m_pool.front();
    34. 		m_pool.pop();
    35. 		delete conn;
    36. 	}
    37. }
    38.  
    39. ConnectionPool* ConnectionPool::getConnectionPool()
    40. {
    41. 	static ConnectionPool pool;
    42. 	return &pool;
    43. }
    44.  
    45. bool ConnectionPool::parseJsonFile()
    46. {
    47. 	ifstream ifs("dbConfig.json");
    48. 	Reader rd;
    49. 	Value root;
    50. 	rd.parse(ifs, root);
    51. 	if (root.isObject())
    52. 	{
    53. 		m_ip = root["ip"].asString();
    54. 		m_port = root["port"].asInt();
    55. 		m_userName = root["userName"].asString();
    56. 		m_passwd = root["passwd"].asString();
    57. 		m_dbName = root["dbName"].asString();
    58. 		m_minSize = root["minSize"].asInt();
    59. 		m_maxSize = root["maxSize"].asInt();
    60. 		m_maxIdleTime = root["maxIdleTime"].asInt();
    61. 		m_timeout = root["timeout"].asInt();
    62.  
    63. 		return true;
    64. 	}
    65. 	return false;
    66. }
    67.  
    68. void ConnectionPool::productConnection()
    69. {
    70. 	while (true)
    71. 	{
    72. 		//对整个局域加锁
    73. 		unique_lock locker(m_lock);
    74. 		while (m_pool.size() >= m_minSize)
    75. 		{
    76. 			m_cond.wait(locker);
    77. 		}
    78. 		addConnection();
    79. 		//唤醒消费者，生产者只有一个线程，就是当前线程
    80. 		m_cond.notify_all();
    81. 	}
    82. }
    83.  
    84. //当连接池的连接个数大于minSize并且等待时长大于最大等待时间，需要销毁
    85. void ConnectionPool::recycleConnection()
    86. {
    87. 	while (true)
    88. 	{
    89. 		this_thread::sleep_for(milliseconds(500));
    90. 		lock_guard locker(m_lock);
    91. 		while (m_pool.size() > m_minSize)
    92. 		{
    93. 			MysqlConn* conn = m_pool.front();
    94. 			if (conn->getAliveTime() >= m_maxIdleTime)
    95. 			{
    96. 				//加锁
    97. 				m_pool.pop();
    98. 				delete conn;
    99. 				conn = nullptr;
    100. 			}
    101. 			else
    102. 			{
    103. 				break;
    104. 			}
    105. 		}
    106. 	}
    107. }
    108.  
    109. void ConnectionPool::addConnection()
    110. {
    111. 	MysqlConn* conn = new MysqlConn;
    112. 	conn->connect(m_userName, m_passwd, m_dbName, m_ip, m_port);
    113. 	conn->refreshAliveTime();
    114. 	m_pool.push(conn);
    115. }
    116.  
    117. shared_ptr ConnectionPool::getConnection()
    118. {
    119. 	//加锁
    120. 	unique_lock locker(m_lock);
    121. 	while (m_pool.empty() || m_useCount >= m_maxSize)
    122. 	{
    123. 		//等待一段时间
    124. 		if (cv_status::no_timeout == m_cond.wait_for(locker, milliseconds(m_timeout)))
    125. 		{
    126. 			//一段时间内没有被唤醒
    127. 			if (m_pool.empty() || m_useCount >= m_maxSize)
    128. 			{
    129. 				continue;
    130. 			}
    131. 		}
    132. 	}
    133. 	//从连接池中取连接
    134. 	shared_ptr conn(m_pool.front(), [this](MysqlConn* conn) {
    135. 		//删除器，将连接放回连接池
    136. 		lock_guard lock(m_lock);
    137. 		conn->refreshAliveTime();
    138. 		m_pool.push(conn);
    139. 		m_useCount--;
    140. 		m_cond.notify_all();
    141. 		});
    142.  
    143. 	//lock_guard lock(m_lock);上面加了
    144. 	m_pool.pop();
    145. 	m_useCount++;
    146. 	//唤醒生产者，由于生产者和消费者使用的同一个环境变量，会将消费者也唤醒
    147. 	//但是不影响，会继续判断
    148. 	m_cond.notify_all();
    149. 	return conn;
    150. }

    测试

    1. #include 
    2. #include "MysqlConn.h"
    3. #include "ConnectionPool.h"
    4.  
    5. //测试数据库API
    6. void query()
    7. {
    8. 	MysqlConn conn;
    9. 	conn.connect("root", "Root_123", "testdb", "114.55.92.82", 3306);
    10. 	string sql = "insert into person value(2, 20, \"man\", \"jj\")";
    11. 	conn.query(sql);
    12.  
    13. 	sql = "select * from person";
    14. 	bool flag = conn.query(sql);
    15. 	cout << "flag value" << flag << endl;
    16.  
    17. 	while (conn.next())
    18. 	{
    19. 		cout << conn.value(0) << ","
    20. 			<< conn.value(1) << ","
    21. 			<< conn.value(2) << ","
    22. 			<< conn.value(3) << ","
    23. 			<< endl;
    24. 	}
    25. }
    26.  
    27. void op1(int begin, int end)
    28. {
    29. 	for (int i = begin; i < end; ++i)
    30. 	{
    31. 		MysqlConn conn;
    32. 		conn.connect("root", "Root_123", "testdb", "114.55.92.82", 3306);
    33. 		//string sql;
    34. 		char sql[1024] = { 0 };
    35. 		sprintf(sql, "insert into person value(%d, 20, \"man\", \"jj\")", i);
    36. 		conn.update(sql);
    37. 	}
    38. }
    39.  
    40. void op2(ConnectionPool* pool, int begin, int end)
    41. {
    42. 	for (int i = begin; i < end; ++i)
    43. 	{
    44. 		shared_ptr ptr = pool->getConnection();
    45. 		char sql[1024] = {0};
    46. 		sprintf(sql, "insert into person value(%d, 20, \"man\", \"jj\")", i);
    47. 		ptr->update(sql);
    48. 	}
    49. }
    50.  
    51. //单线程
    52. void test1()
    53. {
    54. #if 1
    55. 	//非连接池 单线程 耗时108848850600纳秒 108848毫秒
    56. 	steady_clock::time_point begin = steady_clock::now();
    57. 	op1(0, 1000);
    58. 	steady_clock::time_point end = steady_clock::now();
    59. 	auto length = end - begin;
    60. 	cout << "非连接池 单线程 耗时" << length.count() << "纳秒 "
    61. 		<< length.count() / 1000000 << "毫秒" << endl;
    62. #else 
    63. 	//连接池 单线程 耗时16893997100纳秒 16893毫秒
    64. 	ConnectionPool* pool = ConnectionPool::getConnectionPool();
    65. 	steady_clock::time_point begin = steady_clock::now();
    66. 	op2(pool, 0, 1000);
    67. 	steady_clock::time_point end = steady_clock::now();
    68. 	auto length = end - begin;
    69. 	cout << "连接池 单线程 耗时" << length.count() << "纳秒 "
    70. 		<< length.count() / 1000000 << "毫秒" << endl;
    71. #endif 
    72. }
    73.  
    74. void test2()
    75. {
    76. #if 0
    77. 	//非连接池 多线程 耗时20575826400纳秒 20575毫秒
    78. 	
    79. 	//如果同一时间多个线程连接数据库，会导致一些线程连接失败
    80. 	MysqlConn conn;
    81. 	conn.connect("root", "Root_123", "testdb", "114.55.92.82", 3306);
    82.  
    83. 	steady_clock::time_point begin = steady_clock::now();
    84. 	thread t1(op1, 0, 200);
    85. 	thread t2(op1, 200, 400);
    86. 	thread t3(op1, 400, 600);
    87. 	thread t4(op1, 600, 800);
    88. 	thread t5(op1, 800, 1000);
    89. 	t1.join();
    90. 	t2.join();
    91. 	t3.join();
    92. 	t4.join();
    93. 	t5.join();
    94. 	steady_clock::time_point end = steady_clock::now();
    95. 	auto length = end - begin;
    96. 	cout << "非连接池 多线程 耗时" << length.count() << "纳秒 "
    97. 		<< length.count() / 1000000 << "毫秒" << endl;
    98. #else
    99. 	//连接池 多线程 耗时4014272600纳秒 4014毫秒
    100. 	ConnectionPool* pool = ConnectionPool::getConnectionPool();
    101. 	steady_clock::time_point begin = steady_clock::now();
    102. 	thread t1(op2, pool, 0, 200);
    103. 	thread t2(op2, pool, 200, 400);
    104. 	thread t3(op2, pool, 400, 600);
    105. 	thread t4(op2, pool, 600, 800);
    106. 	thread t5(op2, pool, 800, 1000);
    107. 	t1.join();
    108. 	t2.join();
    109. 	t3.join();
    110. 	t4.join();
    111. 	t5.join();
    112. 	steady_clock::time_point end = steady_clock::now();
    113. 	auto length = end - begin;
    114. 	cout << "连接池 多线程 耗时" << length.count() << "纳秒 "
    115. 		<< length.count() / 1000000 << "毫秒" << endl;
    116. #endif 
    117. }
    118.  
    119.  
    120. int main()
    121. {
    122. 	test2();
    123. 	return 0;
    124. }

    在vs中引用第三方库

    当前项目中会需要使用mysql和jsoncpp

    • 进入项目属性
    • 在下面两项中加入头文件路径和库文件路径

    •  在下面加上引用第三方库名

    遇到的问题

    • 编译时报错，找不到libmysql.lib库。

    找到对应动态库，将对应动态库加到项目对应exe文件下。

    • 随后报下面错误

            找到下载的mysql代码，在bin目录下，找到对应库文件，拷贝到项目的对应exe文件下

  • 相关阅读:
    PLC信号发生器(余弦信号)
    目标检测Neck：FPN（Feature Pyramid Network）与PAN（附python代码）
    LCA 的若干种求法
    通过 360 反馈提高团队绩效
    机器学习(公式推导与代码实现)--sklearn机器学习库
    vue3 + ts 项目实站 【二】 vue-router 安装. （后台管理系统)
    Go标准库Context
    MySQL数据库基础知识(一)
    Shell编程
    Fama-French三因子和五因子模型和Stata代码（内附原始数据）
  • 原文地址：https://blog.csdn.net/weixin_57023347/article/details/126694343