连接池优化

连接池的作用，是为了在高并发情况下提高MySQL数据库的访问瓶颈。
当访问MySQL的并发请求多了以后，连接池里面的连接数量会动态增加，上限是maxSize个，当这些连接用完再次归还到连接池当中。如果在指定的maxIdleTime里面，这些新增加的连接都没有被再次使用过，那么新增加的这些连接资源就要被回收掉，只需要保持初始连接量initSize个连接就可以了。

前面实现的连接池受限于queue容器FIFO的特性，连接对象只能从queue头部取出，尾部插入。
在高并发流量潮过后，我们的连接池容量扩到了maxSize条连接。
虽然访问量降下来了，但不排除有这中情况：

把最大空闲时间maxIdleTime设为100s，最大容量maxSize设为100条，高并发过后各条连接的空闲时间如下：

因为queue容器的FIFO特性，队头的空闲时间是一定会比后面的长的，但是，获取连接时也是取得空闲时间最长的那个。
在这种情况下，如果1s内就有一条或多条的获取连接请求，此时queue内Connection的最大空闲时间就永远不会超过100s。 扫描线程Scanner检查队头连接就永远不会释放连接，连接池的大小就降不下来，很明显100条连接肯定是多余的。
假设1s来一个数据库操作请求，执行耗费时间0.5s，2条连接就足以，但用队列的方式管理100条连接一条连接都释放不了。
为了让连接池的容量能降下来，应该让忙的连接更忙，闲的连接更闲，扫描线程检查最闲的连接有没有超过最大空闲时间，如果超过就释放关闭连接。
因此，应该让连接从头部获取和归还，这样就能让忙的连接越忙，闲的连接越闲，头部连接空闲时间永远小于尾部，扫描线程在尾部检查空闲时间，一旦大于最大空闲时间且总创建连接的数量大于初始连接量就释放连接。

要实现这样的管理方式，需要在头部和尾部删除，在头部插入，而queue容器时没有尾部删除操作的，需要换另一种能高效头删、尾删和头插的数据结构。
在STL中，deque和list容器都可以满足这个要求，用这两个都可以实现。下面给出用list实现的连接池。

CommonConnectionPool.h：

#pragma once
/*
实现连接池功能模块
*/

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "Connection.h"
using namespace std;
class ConnectionPool {
public:
	//获取连接池对象实例
	static ConnectionPool* getConnectionPool();
	//给外部提供接口，从连接池中获取一个可用的空闲连接
	shared_ptr<Connection> getConnection();
private:
	ConnectionPool(); 
	bool loadConfigFile();//从配置文件加载配置项

	//运行在独立的线程中，专门负责生产新连接
	void produceConnectionTask();

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

	string _ip; //mysql ip地址
	unsigned short _port; //mysql端口号 3306
	string _username; //mysql用户名
	string _password; //mysql登录密码
	string _dbname; //连接的数据库名称
	int _initSize;	//连接池初始连接量
	int _maxSize;	//连接池最大连接量
	int _maxIdleTime;	//连接池最大空闲时间
	int _connectionTimeOut;	//连接池获取连接的超时时间

	list<Connection*> _connectionList; //存储Mysql连接的队列
	mutex _queueMutex; //维护连接队列的线程安全互斥锁
	atomic_int _connectionCnt; //记录连接所创建的connection连接的总数量
	condition_variable cv; //设置条件变量，用于连接生产线程和连接消费线程的通信
};
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

CommonConnectionPool.cpp：

#include "pch.h"
#include "CommonConnectionPool.h"
#include "public.h"
//线程安全的懒汉单例接口
ConnectionPool* ConnectionPool::getConnectionPool() {
	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 (idx == -1) {
			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") {
			int _connectionTimeOut = atoi(value.c_str());
		}
	}
	return true;
}

ConnectionPool::ConnectionPool() {
	//加载配置项
	if (!loadConfigFile()) return;

	//创建初始数量的连接
	for (int i = 0; i < _initSize; i++) {
		Connection* p = new Connection;

		p->connect(_ip, _port, _username, _password, _dbname);
		p->refreshAliveTime();
		_connectionList.push_front(p);
		_connectionCnt++;
	}

	//启动一个线程作为生产者线程
	thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));
	produce.detach();

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

void ConnectionPool::produceConnectionTask() {
	while (1) {
		unique_lock<mutex> lock(_queueMutex);
		while (!_connectionList.empty()) {
			cv.wait(lock); //队列不空，此处生产者线程进入等待状态
		}
		//连接数量没有到达上限，继续创建新连接
		if (_connectionCnt < _maxSize) {
			Connection* p = new Connection();
			p->connect(_ip, _port, _username, _password, _dbname);
			p->refreshAliveTime();
			_connectionList.push_back(p);
			_connectionCnt++;
		}
		//通知消费者线程，可以消费连接了
		cv.notify_all();
	}
}

shared_ptr<Connection> ConnectionPool::getConnection() {
	unique_lock<mutex> lock(_queueMutex);
	while (_connectionList.empty()) {
		if (cv_status::timeout == cv.wait_for(lock, chrono::milliseconds(_connectionTimeOut))) {
			if (_connectionList.empty()) {
				LOG("获取空闲连接超时...获取连接失败！");
					return nullptr;
			}
		}
	}
	shared_ptr<Connection> sp(_connectionList.front(), 
		[&](Connection* pcon) {
			unique_lock<mutex> lock(_queueMutex);
			pcon->refreshAliveTime();
			_connectionList.push_front(pcon);
		});
	_connectionList.pop_front();
	cv.notify_all(); // 消费完连接以后，通知生产者线程检查一下，如果队列为空了，赶紧生产连接
	return sp;
}

void ConnectionPool::scannerConnectionTask() {
	while (1) {
		//通过sleep模拟定时效果
		this_thread::sleep_for(chrono::seconds(_maxIdleTime));

		//扫描整个队列，释放多余的连接
		unique_lock<mutex> lock(_queueMutex);
		while (_connectionCnt > _initSize) {
			Connection* p = _connectionList.back();
			if (p->getAliveTime() >= (_maxIdleTime * 1000)) {
				_connectionList.pop_back();
				_connectionCnt--;
				delete p;
			}
			else {
				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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144