数据库连接池：代码目录

前言

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

头文件

public.h

#pragma once
#include<iostream>

#define LOG(str) \ std::cout << __FILE__ << ":" << __LINE__ << " " << \ __TIMESTAMP__ << " : " << str << std::endl;

Connection.h

// 实现MySQL数据库的操作
#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、update
	bool update(string sql);

	// 查询操作 select
	MYSQL_RES* query(string sql);

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

	// 返回存活的时间
	clock_t getAliveTime() const;

private:
	MYSQL* _conn;		// 表示和MySQL Server的一条连接
	clock_t _aliveTime; // 记录进入空闲状态后的存活时间
};

CommonConnectionPool.h

#pragma once
// 实现连接池功能模块
#include<iostream>
#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();

	// 回收线程函数，负责回收多余空闲连接
	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; // 存储MySQL连接的队列
	
	mutex _queueMutex;			// 维护连接队列的线程安全互斥锁
	atomic_int _connectionCnt;	// 记录连接所创建的connection的总连接数，不能超过_maxSize
	condition_variable cv;		// 连接生产线程和消费线程之间的通信
};

源文件

Connection.cpp

#include <mysql.h>
#include <string>
#include "Connection.h"
#include "public.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;
}

// 更新操作 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();
}

// 返回存活的时间
clock_t Connection::getAliveTime() const
{
    
	return clock() - _aliveTime;
}

mysql.ini

# 数据库连接池的配置文件
ip=127.0.0.1
port=3306
username=root
password=123456
dbname=chat
initSize=10
maxSize=1024
# 最大空闲时间默认是秒
maxIdleTime=60
# 连接超时时间，单位是毫秒
connectionTimeOut=100

CommonConnectionPool.cpp

#include "CommonConnectionPool.h"
#include "public.h"

// 线程安全的懒汉单例模式函数接口
ConnectionPool* ConnectionPool::getConnectionPool()
{
    
	// 静态局部变量，编译器自动lock和unlock
	static ConnectionPool pool; 
	return& pool;
}

// 从配置文件中加载配置项
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 (-1 == idx) // 无效的配置
		{
    
			continue;
		}

		int endidx = str.find('\n', idx);
		string key = str.substr(0, idx);
		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());
		}
	}
	return true;
}

// 连接池的构造函数
ConnectionPool::ConnectionPool()
{
    
	// 1.加载配置项
	if (!loadConfigFile())
	{
    
		return;
	}

	// 2.创建初始数量连接
	for (int i = 0; i < _initSize; ++i)
	{
    
		Connection* p = new Connection();
		p->connect(_ip, _port, _username, _password, _dbname);
		
		p->refreshAliveTime(); // 刷新一下开始空闲的起始时间
		_connectionQue.push(p);
		_connectionCnt++;
	}

	// 3.启动新的线程作为生产者线程
	thread produce(std::bind( & ConnectionPool::produceConnectionTask, this));
	produce.detach(); // 守护线程（分离线程）

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

// 生产者：运行在独立的线程，专门负责生产连接
void ConnectionPool::produceConnectionTask()
{
    
	// 循环（一直再监听）
	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();	// 通知消费者线程， 可以消费连接了
	}
}

// 给外部提供接口，提供一个空闲的连接
shared_ptr<Connection> ConnectionPool::getConnection()
{
    
	unique_lock<mutex> lock(_queueMutex);
	while (_connectionQue.empty())
	{
    
		// 不是直接sleep
		// 超时醒来发现还是空的，就返回nullptr
		if (cv_status::timeout == cv.wait_for(lock, std::chrono::milliseconds(_connectionTimeOut)))
		{
    
			if (_connectionQue.empty())
			{
    
				LOG("获取空闲连接超时了...获取连接失败！");
				return nullptr;
			}
		}
	}

	/* 由于shared_ptr析构的时候，会把Connection的资源直接delete掉， 相当于调用了Connection的析构函数，Connection就被close掉了， 所以这里需要自定义智能指针释放资源的方式，改用将资源还到队列里 */
	shared_ptr<Connection> sp(_connectionQue.front(),
		[&](Connection* pcon) {
    
			// 这里是在服务器应用线程中调用的，需要考虑线程安全
			unique_lock<mutex> lock(_queueMutex);

			pcon->refreshAliveTime();	// 刷新一下开始空闲的起始时间
			_connectionQue.push(pcon);
		});
	_connectionQue.pop();

	// 消费完队列里最后一个Connection，就通知生产者线程生产连接
	cv.notify_all(); 
	return sp;
}

// 扫描超过maxIdleTime的空闲连接，进行多余的连接回收
void ConnectionPool::scannerConnectionTask()
{
    
	for (; ;)
	{
    
		// 通过sleep模拟定时效果
		this_thread::sleep_for(std::chrono::seconds(_maxIdleTime));

		// 扫描整个队列，释放多于连接
		unique_lock<mutex> lock(_queueMutex);
		while (_connectionCnt > _initSize)
		{
    
			Connection* p = _connectionQue.front();
			if (p->getAliveTime() >= (_maxIdleTime * 1000))
			{
    
				_connectionQue.pop();
				_connectionCnt--;
				delete p;	// 调用 ~Connection();
			}
			else 
			{
    
				break; // 队头的连接没有超过_maxIdleTime，后面的肯定没有
			}
		}
	}
}