470-数据库连接池开发(连接池代码)

短命女 2022-09-07 14:52 335阅读 0赞

## 数据库连接池的配置文件 mysql.ini ##

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70]

## CommomConnectionPool.h ##

#pragma once
    #include <string>
    #include <queue>
    #include <mutex>
    #include <iostream>
    #include <atomic>
    #include <thread>
    #include <condition_variable>
    #include <memory>
    #include <functional>
    using namespace std;
    #include "Connection.h"
    
    /*
    实现连接池功能模块
    */
    class ConnectionPool
    {
        
    public:
    	//获取连接池对象实例
    	static ConnectionPool* getConnectionPool();
    	//给外部提供接口，从连接池中获取一个可用的空闲连接
    	shared_ptr<Connection> getConnection();//智能指针自动管理连接的释放 
    private:
    	//单例#1 构造函数私有化
    	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;//连接池获取连接的超时时间
    
    	queue<Connection*> _connectionQue;//存储mysql连接的队列，多线程从队列取放connection 
    	mutex _queueMutex;//维护连接队列的线程安全互斥锁
    	atomic_int _connectionCnt;//记录连接所创建的connection连接的总数量 
    	condition_variable cv;//设置条件变量，用于连接生产线程和连接消费线程的通信
    };

## CommomConnectionPool.cpp ##

#include "pch.h"
    #include "CommonConnectionPool.h"
    #include "public.h"
    
    //线程安全的懒汉单例函数接口
    ConnectionPool* ConnectionPool::getConnectionPool()
    {
        
    	static ConnectionPool pool;//lock和unlock
    	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;
    		}
    
    		//password=123456\n
    		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()
    {
        
    	//加载配置项了
    	if (!loadConfigFile())
    	{
        
    		return;
    	}
    
    	//创建初始数量的连接
    	for (int i = 0; i < _initSize; ++i)
    	{
        
    		Connection *p = new Connection();
    		p->connect(_ip, _port, _username, _password, _dbname);
    		p->refreshAliveTime();//刷新一下开始空闲的起始时间
    		_connectionQue.push(p);
    		_connectionCnt++;
    	}
    
    	//启动一个新的线程，作为连接的生产者 linux thread => pthread_create
    	thread produce(std::bind(&ConnectionPool::produceConnectionTask, this));//和this当前对象绑定在一起 
    	produce.detach();//守护线程，主线程结束了，这个线程就结束了 
    
    	//启动一个新的定时线程，扫描超过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
    		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;
    }
    
    //扫描超过maxIdleTime时间的空闲连接，进行对于的连接回收
    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--;
    				delete p;//调用~Connection()释放连接
    			}
    			else
    			{
        
    				break;//队头的连接没有超过_maxIdleTime，其它连接肯定没有
    			}
    		}
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xJTlpFWVU2NjY_size_16_color_FFFFFF_t_70]: /images/20220829/860350ffadb54669b4fb50cbab5ff893.png