C++数据库连接池

---

连接池

对于数据库操作都是在访问数据库的时候创建连接，访问完毕断开连接。但是如果在高并发情况下，有些需要频繁处理的操作就会消耗很多的资源和时间，比如：

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

数据库连接池负责分配、管理和释放数据库连接，它允许应用程序重复使用一个现有的数据库连接，而不是再重新建立一个。在并发程度比较高的时候，连接池技术尽可能多地重用了消耗内存地资源，大大节省了内存，提高了服务器地服务效率，能够支持更多的客户服务。通过使用连接池，将大大提高程序运行效率。

---

C API

• MYSQL *mysql_real_connect：连接mysql服务器。
• int mysql_query(MYSQL *mysql, const char *query); 执行sql语句
• MYSQL_RES *mysql_store_result(MYSQL *mysql); 获取结果集
• unsigned int mysql_num_fields(MYSQL_RES *result) 得到结果集的列数
• MYSQL_FIELD *mysql_fetch_fields(MYSQL_RES *result); 获取表头 -> 列名(字段名)
• unsigned long *mysql_fetch_lengths(MYSQL_RES *result); 得到结果集中字段的长度
• MYSQL_ROW mysql_fetch_row(MYSQL_RES *result); 遍历结果集
• void mysql_free_result(MYSQL_RES *result); 释放结果集
• void mysql_close(MYSQL *mysql); 关闭mysql实例
• const char *mysql_character_set_name(MYSQL *mysql) 为当前连接返回默认的字符集。
• int mysql_set_character_set(MYSQL *mysql, char *csname); 设置api使用的字符集
• my_bool mysql_autocommit(MYSQL *mysql, my_bool mode) mysql中默认会进行事务的提交。
• my_bool mysql_commit(MYSQL *mysql); 事务提交
• my_bool mysql_rollback(MYSQL *mysql) 数据回滚
• const char *mysql_error(MYSQL *mysql); 打印错误信息
• unsigned int mysql_errno(MYSQL *mysql); 返回错误编号
• Windows：libmysql.dll 库。
• Linux：libmysqlclient.so。

数据库步骤

数据库连接顺序

1. 初始化连接环境
2. 连接mysql的服务器，需要提供如下连接数据: IP，端口，用户名，密码，数据库名字
3. 数据库的增删查改操作
4. 事务处理：提交事务；或者回滚。
5. 数据库的读操作
6. 遍历结果集。
7. 释放资源。

数据库连接池

• 单例模式，只需要一个类对象。
• 所有的数据库连接应该维护到一个安全的队列中，直接使用STL的queue。
• 在需要的时候可以从连接池中得到一个或多个可用的数据库连接，对于连接的动态创建和销毁，用单独的线程去处理。
• 如果队列中没有多余的可用连接，需要动态的创建新连接。
• 如果队列中空闲的连接太多，需要动态的销毁一部分。
• 数据库操作完毕，需要将连接归还到连接池中。生产者和消费者模型：锁，条件变量。

数据库连接服务操作模块：

#pragma once
#include <iostream>
#include <mysql.h>
#include <chrono>
using namespace std;
using namespace chrono;
class MysqlConn
{
public:
    // 初始化数据库连接
    MysqlConn();
    // 释放数据库连接
    ~MysqlConn();
    // 连接数据库
    bool connect(string user, string passwd, string dbName, string ip, unsigned short port = 3306);
    // 更新数据库: insert, update, delete
    bool update(string sql);
    // 查询数据库
    bool query(string sql);
    // 遍历查询得到的结果集
    bool next();
    // 得到结果集中的字段值
    string value(int index);
    // 事务操作
    bool transaction();
    // 提交事务
    bool commit();
    // 事务回滚 
    bool rollback();
    // 刷新起始的空闲时间点
    void refreshAliveTime();
    // 计算连接存活的总时长
    long long getAliveTime();
private:
    void freeResult();
    MYSQL* m_conn = nullptr;
    MYSQL_RES* m_result = nullptr;
    MYSQL_ROW m_row = nullptr;
    steady_clock::time_point m_alivetime;// 绝对时钟判断连接存活时长
};


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

#include "MysqlConn.h"

MysqlConn::MysqlConn()
{
    m_conn = mysql_init(nullptr); //初始化对象
    mysql_set_character_set(m_conn, "utf8"); // 设置utf-8编码格式
}

MysqlConn::~MysqlConn()
{
    if (m_conn != nullptr)
    {
        mysql_close(m_conn); //关闭连接
    }
    freeResult(); // 释放一个结果集合使用的内存。
}

bool MysqlConn::connect(string user, string passwd, string dbName, string ip, unsigned short port)
{
    MYSQL* ptr = mysql_real_connect(m_conn, ip.c_str(), user.c_str(), passwd.c_str(), dbName.c_str(), port, nullptr, 0); //连接mysql服务器
    return ptr != nullptr;
}

bool MysqlConn::update(string sql)
{
    if (mysql_query(m_conn, sql.c_str())) //更新数据库
    {
        return false;
    }
    return true;
}

bool MysqlConn::query(string sql)
{
    freeResult();
    if (mysql_query(m_conn, sql.c_str()))
    {
        return false;
    }
    m_result = mysql_store_result(m_conn); // 保存结果集
    return true;
}

bool MysqlConn::next()
{
    if (m_result != nullptr)
    {
        m_row = mysql_fetch_row(m_result);
        if (m_row != nullptr)
        {
            return true;
        }
    }
    return false;
}

string MysqlConn::value(int index)
{
    int rowCount = mysql_num_fields(m_result);
    if (index >= rowCount || index < 0)
    {
        return string();
    }
    char* val = m_row[index];
    unsigned long length = mysql_fetch_lengths(m_result)[index];
    return string(val, length);
}

bool MysqlConn::transaction()
{
    return mysql_autocommit(m_conn, false); //手动提交，不自动提交事务。
}

bool MysqlConn::commit()
{
    return mysql_commit(m_conn); //提交事务
}

bool MysqlConn::rollback()
{
    return mysql_rollback(m_conn); // 事务回滚
}

void MysqlConn::refreshAliveTime()
{
    m_alivetime = steady_clock::now();
}

long long MysqlConn::getAliveTime()
{
    nanoseconds res = steady_clock::now() - m_alivetime;// 纳秒 较高精度
    milliseconds millsec = duration_cast<milliseconds>(res);// 毫秒 精度相对较低
    return millsec.count();
}

void MysqlConn::freeResult()
{
    if (m_result)
    {
        mysql_free_result(m_result);
        m_result = nullptr;
    }
}

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

连接池

#pragma once
#include <queue>
#include <mutex>
#include <condition_variable>
#include "MysqlConn.h"
using namespace std;
class ConnectionPool
{
public:
    static ConnectionPool* getConnectPool(); // 创建单例模式
    ConnectionPool(const ConnectionPool& obj) = delete; //禁止使用拷贝构造
    ConnectionPool& operator=(const ConnectionPool& obj) = delete; // 禁止使用拷贝赋值运算符
    shared_ptr<MysqlConn> getConnection();
    ~ConnectionPool();
private:
    ConnectionPool(); //构造函数设置为私有的，单例模式
    bool parseJsonFile();
    void produceConnection();
    void recycleConnection();
    void addConnection();

    string m_ip;// 数据库服务器ip地址
    string m_user;// 数据库服务器用户名
    string m_passwd;// 数据库服务器密码
    string m_dbName;// 数据库服务器的数据库名
    unsigned short m_port;// 数据库服务器绑定的端口
    int m_minSize;// 连接池维护的最小连接数
    int m_maxSize;// 连接池维护的最大连接数
    int m_timeout;// 连接池获取连接的超时时长
    int m_maxIdleTime;// 连接池中连接的最大空闲时长


    queue<MysqlConn*> m_connectionQ; // 连接队列
    mutex m_mutexQ; 
    condition_variable m_cond;
};


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

#include "ConnectionPool.h"
#include <json/json.h>
#include <fstream>
#include <thread>
using namespace Json;
ConnectionPool* ConnectionPool::getConnectPool()
{
    static ConnectionPool pool;
    return &pool;
}

bool ConnectionPool::parseJsonFile()
{
    ifstream ifs("dbconf.json");
    Reader rd;
    Value root;
    rd.parse(ifs, root);
    if (root.isObject())
    {
        m_ip = root["ip"].asString();
        m_port = root["port"].asInt();
        m_user = root["userName"].asString();
        m_passwd = root["password"].asString();
        m_dbName = root["dbName"].asString();
        m_minSize = root["minSize"].asInt();
        m_maxSize = root["maxSize"].asInt();
        m_maxIdleTime = root["maxIdleTime"].asInt();
        m_timeout = root["timeout"].asInt();
        return true;
    }
    return false;
}

// 生产连接线程函数
void ConnectionPool::produceConnection()
{
    while (true)
    {
        unique_lock<mutex> locker(m_mutexQ); // 锁
        while (m_connectionQ.size() >= m_minSize)
        {
            m_cond.wait(locker); // 条件变量
        }
        addConnection(); // 执行连接
        m_cond.notify_all(); // 通知消费者线程消费
    }
}


// 销毁连接
void ConnectionPool::recycleConnection()
{
    while (true)
    {
        this_thread::sleep_for(chrono::milliseconds(500));
        lock_guard<mutex> locker(m_mutexQ);
        while (m_connectionQ.size() > m_minSize)
        {
            MysqlConn* conn = m_connectionQ.front();
            if (conn->getAliveTime() >= m_maxIdleTime) // 判断连接存活时长，如果满足条件就行删除
            {
                m_connectionQ.pop();
                delete conn;
            }
            else
            {
                break;
            }
        }
    }
}

void ConnectionPool::addConnection()
{
    MysqlConn* conn = new MysqlConn;
    conn->connect(m_user, m_passwd, m_dbName, m_ip, m_port);
    conn->refreshAliveTime();
    m_connectionQ.push(conn);
}

// 取出可用连接
shared_ptr<MysqlConn> ConnectionPool::getConnection()
{
    unique_lock<mutex> locker(m_mutexQ);
    while (m_connectionQ.empty()) // 判断连接是否为空
    {
        if (cv_status::timeout == m_cond.wait_for(locker, chrono::milliseconds(m_timeout)))// 阻塞一段时间
        {
            if (m_connectionQ.empty())
            {
                //return nullptr;
                continue;
            }
        }
    }
    // lambda表达式定义shared_ptr的销毁函数。
    shared_ptr<MysqlConn> connptr(m_connectionQ.front(), [this](MysqlConn* conn) {
        lock_guard<mutex> locker(m_mutexQ);
        conn->refreshAliveTime();
        m_connectionQ.push(conn);
        });
    m_connectionQ.pop();
    m_cond.notify_all();// 唤醒生产者线程，附带着也会唤醒消费者线程
    return connptr;
}

ConnectionPool::~ConnectionPool()
{
    while (!m_connectionQ.empty())
    {
        MysqlConn* conn = m_connectionQ.front();
        m_connectionQ.pop();
        delete conn;
    }
}

ConnectionPool::ConnectionPool()
{
    // 加载配置文件
    if (!parseJsonFile())
    {
        return;
    }

    for (int i = 0; i < m_minSize; ++i) // 默认情况保证有minsize个数的连接就行了。
    {
        addConnection();
    }
    thread producer(&ConnectionPool::produceConnection, this);// 生产连接
    thread recycler(&ConnectionPool::recycleConnection, this);// 看有没有需要销毁的连接
    producer.detach();
    recycler.detach();
}

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

参考列表：
https://www.bilibili.com/video/BV1Fr4y1s7w4
https://blog.csdn.net/CrankZ/article/details/82874158