^-^
后端技术_mysql连接池实现
最近更新：2024-09-23   |   字数总计：2.7k   |   阅读估时：12分钟   |   阅读量：次
  1. 池化技术
  2. MySQL连接池实现
    1. 接口设计
    2. 连接池接口实现
    3. 单个连接接口实现
    4. 功能测试及使用
    5. 连接池配合线程池及性能测试
    6. Mysql与redis自动重连
      1. Mysql
      2. Redis
  3. 连接池连接数量设置
  4. 同步与异步连接
  5. 其他参考

池化技术

  1. 目的：提高响应、复用资源
  2. 数据库连接池：
    1. 资源复用
    2. 更快响应速度
    3. 统一管理连接，避免数据库连接泄漏
  3. 数据库连接过程：tcp握手、mysql认证三次握手、sql执行、mysql关闭、tcp四次挥手
    1. 池化后：每次仅执行sql，握手认证均托管给连接池
  4. 与长连接的关系：
    1. 连接池中的连接就是长连接，连接池的本质就是管理长连接
    2. 长连接是解决单个客户端和服务器之间建立链接所带来的性能问题
    3. 连接池是解决应用程序与数据库之间频繁操作时带来的创建链接性能问题
  5. 运行机制：
    1. 从连接池获取或创建可用连接
    2. 使用完毕后归还连接
    3. 系统关闭时再断开连接及释放资源

MySQL连接池实现

接口设计

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
// DBPool.h

class CDBConn;

class CDBPool {	// 只是负责管理连接CDBConn，真正干活的是CDBConn
public:
	CDBPool() {} // 如果在构造函数做一些可能失败的操作，需要抛出异常，外部要捕获异常
	CDBPool(const char* pool_name, const char* db_server_ip, uint16_t db_server_port,
			const char* username, const char* password, const char* db_name, 
			int max_conn_cnt);
	virtual 	~CDBPool();

	int 		Init();		// 连接数据库，创建连接
	CDBConn* 	GetDBConn(const int timeout_ms = 0);	// 获取连接资源
	void 		RelDBConn(CDBConn* pConn);	// 归还连接资源

	const char* GetPoolName() { return m_pool_name.c_str(); }
	const char* GetDBServerIP() { return m_db_server_ip.c_str(); }
	uint16_t 	GetDBServerPort() { return m_db_server_port; }
	const char* GetUsername() { return m_username.c_str(); }
	const char* GetPasswrod() { return m_password.c_str(); }
	const char* GetDBName() { return m_db_name.c_str(); }
private:
	string 		m_pool_name;	// 连接池名称
	string 		m_db_server_ip;	// 数据库ip
	uint16_t	m_db_server_port; // 数据库端口
	string 		m_username;  	// 用户名
	string 		m_password;		// 用户密码
	string 		m_db_name;		// db名称
	int			m_db_cur_conn_cnt;	// 当前启用的连接数量
	int 		m_db_max_conn_cnt;	// 最大连接数量
	list<CDBConn*>	m_free_list;	// 空闲的连接

	list<CDBConn*>	m_used_list;		// 记录已经被请求的连接
	std::mutex m_mutex;
    std::condition_variable m_cond_var;
	bool m_abort_request = false;
	int wait_cout = 0;  // ./test_dbpool 4 1 1
};

class CDBConn {
public:
	CDBConn(CDBPool* pDBPool);
	virtual ~CDBConn();
	int Init();

	// 创建表
	bool ExecuteCreate(const char* sql_query);
	// 删除表
	bool ExecuteDrop(const char* sql_query);
	// 查询
	CResultSet* ExecuteQuery(const char* sql_query);

    /**
    *  执行DB更新，修改
    *
    *  @param sql_query     sql
    *  @param care_affected_rows  是否在意影响的行数，false:不在意；true:在意
    *
    *  @return 成功返回true 失败返回false
    */
	bool ExecuteUpdate(const char* sql_query, bool care_affected_rows = true);
	uint32_t GetInsertId();

	// 开启事务
	bool StartTransaction();
	// 提交事务
	bool Commit();
	// 回滚事务
	bool Rollback();
	// 获取连接池名
	const char* GetPoolName();
	MYSQL* GetMysql() { return m_mysql; }
private:
	CDBPool* 	m_pDBPool;	// to get MySQL server information
	MYSQL* 		m_mysql;	// 对应一个连接
	char		m_escape_string[MAX_ESCAPE_STRING_LEN + 1];
};

连接池接口实现

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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
CDBPool::CDBPool(const char *pool_name, const char *db_server_ip, uint16_t db_server_port,
				 const char *username, const char *password, const char *db_name, int max_conn_cnt)
{
	m_pool_name = pool_name;
	m_db_server_ip = db_server_ip;
	m_db_server_port = db_server_port;
	m_username = username;
	m_password = password;
	m_db_name = db_name;
	m_db_max_conn_cnt = max_conn_cnt;	// 
	m_db_cur_conn_cnt = MIN_DB_CONN_CNT; // 最小连接数量
}
/* 释放连接池 */
CDBPool::~CDBPool()
{
	std::lock_guard<std::mutex> lock(m_mutex);
	m_abort_request = true;
	m_cond_var.notify_all();		// 通知所有在等待的

	for (list<CDBConn *>::iterator it = m_free_list.begin(); it != m_free_list.end(); it++)
	{
		CDBConn *pConn = *it;
		delete pConn;
	}

	m_free_list.clear();
}
/* 创建连接池 */
int CDBPool::Init()
{
	// !!!首先创建固定最小的连接数量
	for (int i = 0; i < m_db_cur_conn_cnt; i++)
	{
		CDBConn *pDBConn = new CDBConn(this);
		int ret = pDBConn->Init();
		if (ret)
		{
			delete pDBConn;
			return ret;
		}

		m_free_list.push_back(pDBConn);
	}

	// log_info("db pool: %s, size: %d\n", m_pool_name.c_str(), (int)m_free_list.size());
	return 0;
}

/*
 取连接
 !!!重点实现
 先从free_list里面看有没有可用的
    1.有 => 之间返回
    2.没有
        是否达到了最大的连接数量
            1. 是
                1. timeout为0 => 直接休眠死等，等待被条件唤醒or退出指令
                2. timeout>0 => 等待超时，超时后返回NULL or 返回连接 or 退出指令
            2. 不是 => 创建新的连接 => 返回
*/
CDBConn *CDBPool::GetDBConn(const int timeout_ms)
{
	std::unique_lock<std::mutex> lock(m_mutex);
	if(m_abort_request) 
	{
		log_warn("have aboort\n");
		return NULL;
	}

	if (m_free_list.empty())		// 2 当没有连接可以用时
	{
		// 第一步先检测 当前连接数量是否达到最大的连接数量 
		if (m_db_cur_conn_cnt >= m_db_max_conn_cnt) // 等待的逻辑
		{
			// 如果已经到达了，看看是否需要超时等待
			if(timeout_ms <= 0)		// 死等，直到有连接可以用 或者 连接池要退出
			{
				log_info("wait ms:%d\n", timeout_ms);
				m_cond_var.wait(lock, [this] 
				{
					log_info("wait:%d, size:%lu\n", wait_cout++, m_free_list.size());
					// 当前连接数量小于最大连接数量 或者请求释放连接池时退出
					return (!m_free_list.empty()) | m_abort_request;
				});
			} else {
				// return如果返回 false，继续wait(或者超时),  如果返回true退出wait
				// 1.m_free_list不为空
				// 2.超时退出
				// 3. m_abort_request被置为true，要释放整个连接池
				m_cond_var.wait_for(lock, std::chrono::milliseconds(timeout_ms), [this] {
					log_info("wait_for:%d, size:%lu\n", wait_cout++, m_free_list.size());
					return (!m_free_list.empty()) | m_abort_request;
				});
				// 带超时功能时还要判断是否为空
				if(m_free_list.empty()) 	// 如果连接池还是没有空闲则退出
				{
					return NULL;
				}
			}

			if(m_abort_request) 
			{
				log_warn("have aboort\n");
				return NULL;
			}
		}
		else // 还没有到最大连接则创建连接
		{
			CDBConn *pDBConn = new CDBConn(this);	//新建连接
			int ret = pDBConn->Init();
			if (ret)
			{
				log_error("Init DBConnecton failed\n\n");
				delete pDBConn;
				return NULL;
			}
			else
			{
				m_free_list.push_back(pDBConn);
				m_db_cur_conn_cnt++;
				// log_info("new db connection: %s, conn_cnt: %d\n", m_pool_name.c_str(), m_db_cur_conn_cnt);
			}
		}
	}

	CDBConn *pConn = m_free_list.front();	// 获取连接
	m_free_list.pop_front();	// STL 吐出连接，从空闲队列删除
	//   pConn->setCurrentTime();  // 伪代码
	// m_used_list.push_back(pConn);		// 

	return pConn;
}

/* 归还连接给连接池 
    1. 判断连接不为空
    2. 是否重复归还
*/
void CDBPool::RelDBConn(CDBConn *pConn)
{
	if(!pConn) {
		log_error("pConn is null");
		return;
	}

	std::lock_guard<std::mutex> lock(m_mutex);
	list<CDBConn *>::iterator it = m_free_list.begin();
	for (; it != m_free_list.end(); it++)	// 避免重复归还
	{
		if (*it == pConn)	
		{
			break;  // 说明这个连接已经归还过
		}
	}

	if (it == m_free_list.end())
	{
		// m_used_list.remove(pConn);
		m_free_list.push_back(pConn);
		m_cond_var.notify_one();		// 通知取队列
	} 
	else 
	{
		log_error("RelDBConn failed\n"); // 不再次回收连接
	}
}

单个连接接口实现

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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
CDBConn::CDBConn(CDBPool *pPool)
{
	m_pDBPool = pPool;
	m_mysql = NULL;
}

CDBConn::~CDBConn()
{
	if (m_mysql)
	{
		mysql_close(m_mysql);
	}
}

int CDBConn::Init()
{
	m_mysql = mysql_init(NULL);	// mysql_标准的mysql c client对应的api
	if (!m_mysql)
	{
		log_error("mysql_init failed\n");
		return 1;
	}

	int reconnect = 1;
	mysql_options(m_mysql, MYSQL_OPT_RECONNECT, &reconnect);	// 配合mysql_ping实现自动重连
	mysql_options(m_mysql, MYSQL_SET_CHARSET_NAME, "utf8mb4");	// ！！utf8mb4和utf8区别，有生僻字及表情需要使用utf8mb4

	// ip 端口 用户名 密码 数据库名
	if (!mysql_real_connect(m_mysql, m_pDBPool->GetDBServerIP(), m_pDBPool->GetUsername(), m_pDBPool->GetPasswrod(),
							m_pDBPool->GetDBName(), m_pDBPool->GetDBServerPort(), NULL, 0))
	{
		log_error("mysql_real_connect failed: %s\n", mysql_error(m_mysql));
		return 2;
	}

	return 0;
}

const char *CDBConn::GetPoolName()
{
	return m_pDBPool->GetPoolName();
}

bool CDBConn::ExecuteCreate(const char *sql_query)
{
	mysql_ping(m_mysql);
	// mysql_real_query 实际就是执行了SQL
	if (mysql_real_query(m_mysql, sql_query, strlen(sql_query)))
	{
		log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
		return false;
	}

	return true;
}
bool CDBConn::ExecuteDrop(const char *sql_query)
{
	mysql_ping(m_mysql);	// 如果端开了，能够自动重连

	if (mysql_real_query(m_mysql, sql_query, strlen(sql_query)))
	{
		log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
		return false;
	}

	return true;
}

CResultSet *CDBConn::ExecuteQuery(const char *sql_query)
{
	mysql_ping(m_mysql);

	if (mysql_real_query(m_mysql, sql_query, strlen(sql_query)))
	{
		log_error("mysql_real_query failed: %s, sql: %s\n", mysql_error(m_mysql), sql_query);
		return NULL;
	}
	// 返回结果
	MYSQL_RES *res = mysql_store_result(m_mysql);	// 返回结果
	if (!res)
	{
		log_error("mysql_store_result failed: %s\n", mysql_error(m_mysql));
		return NULL;
	}

	CResultSet *result_set = new CResultSet(res);	// 存储到CResultSet
	return result_set;
}

/*
1.执行成功，则返回受影响的行的数目，如果最近一次查询失败的话，函数返回 -1

2.对于delete,将返回实际删除的行数.

3.对于update,如果更新的列值原值和新值一样,如update tables set col1=10 where id=1;
id=1该条记录原值就是10的话,则返回0。

mysql_affected_rows返回的是实际更新的行数,而不是匹配到的行数。
*/
bool CDBConn::ExecuteUpdate(const char *sql_query, bool care_affected_rows)
{
	mysql_ping(m_mysql);

	if (mysql_real_query(m_mysql, sql_query, strlen(sql_query)))
	{
		log_error("mysql_real_query failed: %s, sql: %s\n", mysql_error(m_mysql), sql_query);
		//g_master_conn_fail_num ++;
		return false;
	}

	if (mysql_affected_rows(m_mysql) > 0)
	{
		return true;
	}
	else
	{ // 影响的行数为0时
		if (care_affected_rows)
		{ // 如果在意影响的行数时, 返回false, 否则返回true
			log_error("mysql_real_query failed: %s, sql: %s\n\n", mysql_error(m_mysql), sql_query);
			return false;
		}
		else
		{
			log_warn("affected_rows=0, sql: %s\n\n", sql_query);
			return true;
		}
	}
}

bool CDBConn::StartTransaction()
{
	mysql_ping(m_mysql);

	if (mysql_real_query(m_mysql, "start transaction\n", 17))
	{
		log_error("mysql_real_query failed: %s, sql: start transaction\n", mysql_error(m_mysql));
		return false;
	}

	return true;
}

bool CDBConn::Rollback()
{
	mysql_ping(m_mysql);

	if (mysql_real_query(m_mysql, "rollback\n", 8))
	{
		log_error("mysql_real_query failed: %s, sql: rollback\n", mysql_error(m_mysql));
		return false;
	}

	return true;
}

bool CDBConn::Commit()
{
	mysql_ping(m_mysql);

	if (mysql_real_query(m_mysql, "commit\n", 6))
	{
		log_error("mysql_real_query failed: %s, sql: commit\n", mysql_error(m_mysql));
		return false;
	}

	return true;
}
uint32_t CDBConn::GetInsertId()
{
	return (uint32_t)mysql_insert_id(m_mysql);
}

功能测试及使用

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
// test_curd.cpp
void testCurd()
{
    const char *db_pool_name    = DB_POOL_NAME;
    const char* db_host         = DB_HOST_IP;
    int   db_port               = DB_HOST_PORT;
    const char* db_dbname       = DB_DATABASE_NAME;
    const char* db_username     = DB_USERNAME;
    const char* db_password     = DB_PASSWORD;
    int db_maxconncnt           = DB_POOL_MAX_CON;
    CDBPool* pDBPool = new CDBPool(db_pool_name, db_host, db_port, 
        db_username, db_password, db_dbname, db_maxconncnt);
    if (pDBPool->Init()) {
        printf("init db instance failed: %s", db_pool_name);
        return;
    }

    CDBConn* pDBConn = pDBPool->GetDBConn();
    if(pDBConn)
    {
        bool ret = pDBConn->ExecuteDrop(DROP_IMUSER_TABLE);
        if(ret)
        {
            printf("DROP_IMUSER_TABLE ok\n");
        }
        // 1. 创建表
        ret = pDBConn->ExecuteCreate(CREATE_IMUSER_TABLE);
        if(ret)
        {
            printf("CREATE_IMUSER_TABLE ok\n");
        }

        // 2. 插入内容
        ret = insertUser(pDBConn);
        if(ret)
        {
            printf("insertUser ok -------\n\n");
        }
        // 3. 查询内容
        ret = queryUser(pDBConn);
        if(ret)
        {
            printf("queryUser ok -------\n\n");
        }
        // 4. 修改内容
        ret = updateUser(pDBConn);
        if(ret)
        {
            printf("updateUser ok -------\n\n");
        }
        ret = queryUser(pDBConn);
        if(ret)
        {
            printf("queryUser ok -------\n\n");
        }
        //归还连接
        pDBPool->RelDBConn(pDBConn);
    }
    else
    {
        printf("pDBConn is null\n");
    }
    delete pDBPool;
}

连接池配合线程池及性能测试

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
// test_dbpool.cpp
// 使用连接池的方式
void *workUsePool(void *arg, int id)    // 任务
{
    // printf("workUsePool id:%d\n", id);
    CDBPool *pDBPool = (CDBPool *)arg;
    CDBConn *pDBConn = pDBPool->GetDBConn(2000); // 获取连接，超时2000ms
    if (pDBConn)
    {
        bool ret = insertUser(pDBConn, id); // 插入用户信息
        if (!ret)
        {
            printf("insertUser failed\n");
        }
        pDBPool->RelDBConn(pDBConn);     // 然后归还连接
    }
    else
    {
        printf("GetDBConn failed\n");
    }
   
    // printf("exit id:%d\n", id);
    return NULL;
}

// 线程池启用线程数、db最大连接数、任务数量
int testWorkUsePool(int thread_num, int db_maxconncnt, int task_num)
{
    const char *db_pool_name = DB_POOL_NAME;
    const char *db_host = DB_HOST_IP;
    int db_port = DB_HOST_PORT;
    const char *db_dbname = DB_DATABASE_NAME;
    const char *db_username = DB_USERNAME;
    const char *db_password = DB_PASSWORD;

    // 每个连接池都对应一个对象
    CDBPool *pDBPool = new CDBPool(db_pool_name, db_host, db_port,
                                   db_username, db_password, db_dbname, db_maxconncnt);
    if (pDBPool->Init())
    {
        printf("init db instance failed: %s", db_pool_name);
        return -1;
    }

    CDBConn *pDBConn = pDBPool->GetDBConn(); // 获取连接
    if (pDBConn)
    {
        bool ret = pDBConn->ExecuteDrop(DROP_IMUSER_TABLE); // 删除表
        if (ret)
        {
            // printf("DROP_IMUSER_TABLE ok\n");
        }
        // 1. 创建表
        ret = pDBConn->ExecuteCreate(CREATE_IMUSER_TABLE); // 创建表
        if (ret)
        {
            // printf("CREATE_IMUSER_TABLE ok\n");
        }
        else
        {
            printf("ExecuteCreate failed\n");
            return -1;
        }
    }
    pDBPool->RelDBConn(pDBConn); // 一定要归还

    printf("task_num = %d, thread_num = %d, connection_num:%d, use_pool:1\n",
           task_num, thread_num, db_maxconncnt);

    ZERO_ThreadPool threadpool;
    threadpool.init(thread_num); // 设置线程数量
    threadpool.start();          // 启动线程池
    uint64_t start_time =  get_tick_count();
    for (int i = 0; i < task_num; i++)
    {
        threadpool.exec(workUsePool, (void *)pDBPool, i); //把所有的任务都给线程池
    }
    cout << "need time0: " <<  get_tick_count()  - start_time << "ms\n";
    threadpool.waitForAllDone(); // 等待所有执行万再退出
    cout << "need time1: " <<  get_tick_count()  - start_time << "ms\n";
    threadpool.stop();
    cout << "need time2: " <<  get_tick_count()  - start_time << "ms\n\n";
    delete pDBPool;
    return 0;
}

Mysql与redis自动重连

Mysql

1
2
3
4
5
// 位于Init函数
int reconnect = 1;
mysql_options(m_mysql, MYSQL_OPT_RECONNECT, &reconnect); // 开启自动尝试重连
// 位于sql执行前
mysql_ping(m_mysql); // 尝试重新连接服务器

Redis

1
2
3
4
5
6
7
// 位于Init
if(m_pContext) return 0; // 连接正常
/// ...
m_pContext = redisConnectWithTimeout(m_server_ip.c_str(), m_server_port, timeout);
// 位于执行前
if(Init() != 0){//...
}

连接池连接数量设置

  1. 经验公式：（连接数 = (核心数*2) + 有效磁盘数）
  2. IO密集型：一个cpu设置（1/cpu利用率）个线程，假设cpu利用率为5/(100+5),即一个cpu设置21个线程，4个cpu设置84个线程，那么连接数也为84个
  3. 一般情况下：连接数 = 线程数
  4. 具体项目需要测试并具体分析

同步与异步连接

  1. 同步：编程方便、性能比不上异步
  2. 异步：性能高、编程麻烦
    1. 拆分任务
      1. 发起查询
      2. 获取查询结果（等待回调）

其他参考

  1. kafka C++客户端 异步处理编程
  2. druid连接池设计理念
  3. HikariCP wiki jdb连接池组件及连接数设置
  4. 思考
    1. 长时间不归还连接如何检测
    2. 连接数量如果经常进入连接等待状态如何处理
2024-03-01 该篇文章被 Cleofwine 归为分类: 服务端

上一篇：后端技术_内存池的实现与场景分析 下一篇：后端技术_手写线程池与性能分析
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题