数据库连接池：连接池功能点的实现

连接池功能

主要包含以下功能：

1. 单例模式的连接池，构造函数需私有化
2. 需要一个获取连接的函数
3. 加载配置文件的函数
4. 生产者线程，用来生产连接
5. 用来回收多余空闲连接的函数，独立的线程

数据成员有：

6. 数据库所需信息
7. 连接池的初始连接量和最大连接量
8. 连接池的最大等待时间
9. 连接池获取连接的超时时间
10. 存储MySQL连接的队列
11. 维护该连接队列的互斥量
12. 原子整型操作，记录连接个数
13. 条件变量，用于生产者和消费者线程之间的通信。

图示：

因此可编写出头文件信息：

连接池头文件

#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;		// 连接生产线程和消费线程之间的通信
};

功能实现

单例模式的连接池

这里采用懒汉单例模式，由编译器自己进行加锁操作

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

MySQL配置文件

在Windows下该文件的后缀为 MySQL.ini

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

读取配置文件

// 从配置文件中加载配置项
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;
}

生产者线程实现

// 生产者：运行在独立的线程，专门负责生产连接
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;
}

回收多余连接，独立线程

这里需要MySQL连接的类里实现两个函数：

clock_t _aliveTime; // 记录进入空闲状态后的存活时间

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

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

然后再来看回收函数

// 扫描超过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，后面的肯定没有
			}
		}
	}
}