【C++】手写 MySQL 连接池（一）：整体结构与基本使用流程

在 C 程序中可以直接使用 MySQL C API 建立数据库连接并执行 SQL。但在服务器程序或多线程程序中如果每执行一次查询都重新连接数据库不仅代码重复还会产生较大的连接开销。连接池的核心思想是程序启动时提前创建一定数量的数据库连接执行 SQL 时直接使用已有连接使用完成后再交还给连接池。本文先从整体角度认识一个同时支持同步查询和异步任务的 MySQL 连接池不深入分析任务队列、工作线程和预处理语句的内部实现。一、为什么需要 MySQL 连接池最基础的 MySQL 使用流程通常是// 初始化MySQL连接对象 MYSQL* mysql mysql_init(nullptr); // 连接数据库 mysql_real_connect( mysql, 127.0.0.1, root, password, sakila, 3306, nullptr, 0 ); // 执行SQL mysql_query(mysql, SELECT * FROM actor); // 关闭数据库连接 mysql_close(mysql);如果程序只执行一次查询这种方式没有明显问题。但在服务器程序中可能会不断处理用户请求收到请求 ↓ 连接MySQL ↓ 执行SQL ↓ 关闭连接 ↓ 收到下一个请求 ↓ 再次连接MySQL频繁创建和关闭数据库连接会带来以下问题重复进行网络连接和身份验证增加数据库服务器压力多线程环境下连接管理比较混乱查询较多时响应速度不够稳定容易出现连接忘记关闭的问题。连接池会提前创建多个连接MySQL连接池 ├── connection1 ├── connection2 ├── connection3 └── connection4当线程需要执行查询时从连接池中寻找一个空闲连接业务线程 ↓ 获取空闲连接 ↓ 执行SQL ↓ 释放连接锁 ↓ 连接继续留在池中这里所谓的“释放连接”并不是调用mysql_close()将连接真正关闭而是解除连接的占用状态让其他线程继续使用。二、连接池由哪些核心部分组成一个完整的 MySQL 连接池不仅要保存数据库连接还要负责连接创建、任务提交、线程执行和资源关闭。整体结构可以概括为DatabaseWorkerPool ├── 同步连接集合 ├── 异步连接集合 ├── SQL任务队列 ├── 数据库连接信息 └── 预处理语句信息连接池类的核心成员如下template class T class DatabaseWorkerPool { private: enum InternalIndex { IDX_ASYNC, // 异步连接下标 IDX_SYNCH, // 同步连接下标 IDX_SIZE }; // 异步SQL任务使用的生产者消费者队列 std::unique_ptr ProducerConsumerQueueSQLOperation* _queue; /* * 保存两组连接 * * _connections[IDX_ASYNC]异步连接 * _connections[IDX_SYNCH]同步连接 */ std::array std::vectorstd::unique_ptrT, IDX_SIZE _connections; // 保存主机、端口、用户名、密码和数据库名 std::unique_ptrMySQLConnectionInfo _connectionInfo; // 保存预处理语句的参数数量 std::vectoruint8 _preparedStatementSize; // 异步连接数量 uint8 _async_threads; // 同步连接数量 uint8 _synch_threads; };这里使用模板template class T class DatabaseWorkerPool;表示连接池可以管理不同类型的数据库连接。例如针对sakila数据库定义一个连接类型class SakilaDatabaseConnection : public MySQLConnection { public: // 同步连接构造函数 SakilaDatabaseConnection( MySQLConnectionInfo connInfo ); // 异步连接构造函数 SakilaDatabaseConnection( ProducerConsumerQueueSQLOperation** queue, MySQLConnectionInfo connInfo ); // 注册当前数据库需要使用的预处理语句 void DoPrepareStatements() override; };然后创建对应的连接池对象// 头文件中声明 extern DatabaseWorkerPool SakilaDatabaseConnection SakilaDatabase;// 源文件中定义 DatabaseWorkerPool SakilaDatabaseConnection SakilaDatabase;这样其他代码可以统一通过SakilaDatabase.Query(...); SakilaDatabase.Execute(...); SakilaDatabase.Close();操作数据库不需要直接管理底层的MYSQL*连接对象。三、同步连接和异步连接有什么区别这个连接池会同时创建同步连接和异步连接。配置连接池时可以指定两种连接的数量loader.AddDatabase( SakilaDatabase, 127.0.0.1;3306;root;password;sakila, 8, // 创建8个异步连接 2 // 创建2个同步连接 );连接字符串的格式为主机地址;端口;用户名;密码;数据库名例如127.0.0.1;3306;root;password;sakila解析后的连接信息保存在struct MySQLConnectionInfo { std::string user; // 用户名 std::string password; // 密码 std::string database; // 数据库名 std::string host; // 主机地址 std::string port_or_socket; // 端口或Socket std::string ssl; // SSL配置 };连接池通过下面的接口保存这些配置template class T void DatabaseWorkerPoolT::SetConnectionInfo( const std::string infoString, uint8 asyncThreads, uint8 synchThreads) { // 解析数据库连接字符串 _connectionInfo std::make_uniqueMySQLConnectionInfo( infoString ); // 保存异步连接数量 _async_threads asyncThreads; // 保存同步连接数量 _synch_threads synchThreads; }调用Open()时先创建异步连接再创建同步连接template class T uint32 DatabaseWorkerPoolT::Open() { // 创建异步连接 uint32 error OpenConnections( IDX_ASYNC, _async_threads ); if (error ! 0) { return error; } // 创建同步连接 error OpenConnections( IDX_SYNCH, _synch_threads ); return error; }创建连接时根据连接类型调用不同的构造函数template class T uint32 DatabaseWorkerPoolT::OpenConnections( InternalIndex type, uint8 numConnections) { for (uint8 i 0; i numConnections; i) { std::unique_ptrT connection; if (type IDX_ASYNC) { /* * 异步连接需要绑定SQL任务队列。 * 后续SQL任务会由后台工作线程执行。 */ connection std::make_uniqueT( _queue.get(), *_connectionInfo ); } else { /* * 同步连接不需要任务队列。 * 调用Query时由当前业务线程直接执行。 */ connection std::make_uniqueT( *_connectionInfo ); } // 真正调用MySQL API连接数据库 uint32 error connection-Open(); if (error ! 0) { // 当前类型的连接创建失败清理已创建连接 _connections[type].clear(); return error; } // 将成功建立的连接加入连接池 _connections[type].push_back( std::move(connection) ); } return 0; }两种连接可以先简单理解为同步连接 调用Query后当前线程等待SQL执行完成并返回结果。 异步连接 SQL先进入任务队列由后台工作线程负责执行。例如同步查询auto result SakilaDatabase.Query( SELECT actor_id, first_name FROM actor WHERE actor_id 1 );当前线程会等待数据库返回查询结果。异步查询则是auto callback SakilaDatabase.AsyncQuery( SELECT actor_id, first_name FROM actor WHERE actor_id 1 );SQL 会先封装成任务并放入队列调用线程不需要在当前位置一直等待。第一篇先重点理解同步查询的使用流程异步任务会在后续单独分析。四、MySQL 连接池的基本使用流程使用连接池时完整流程通常包括初始化MySQL客户端库 ↓ 注册数据库连接池 ↓ 创建同步和异步连接 ↓ 准备预处理语句 ↓ 执行SQL查询 ↓ 关闭连接池 ↓ 释放MySQL客户端库下面给出一个完整示例#include DatabaseEnv.h #include DatabaseLoader.h #include MySQLThreading.h #include Log.h int main() { /* * 第一步初始化MySQL客户端库。 * * 内部调用mysql_library_init() * 应当在创建数据库连接前执行。 */ MySQL::Library_Init(); /* * 第二步创建数据库加载器。 * * DatabaseLoader负责统一打开连接池 * 并准备所有预处理语句。 */ DatabaseLoader loader; /* * 第三步注册数据库。 * * 连接字符串格式 * host;port;user;password;database * * 最后的8和2分别表示 * 8个异步连接 * 2个同步连接。 */ loader.AddDatabase( SakilaDatabase, 127.0.0.1;3306;root;password;sakila, 8, 2 ); /* * 第四步打开连接池并准备SQL语句。 */ if (!loader.Load()) { TC_LOG_ERROR( , SakilaDatabase connect error ); // 释放MySQL客户端库资源 MySQL::Library_End(); return 1; } TC_LOG_INFO( , SakilaDatabase connect success ); /* * 第五步执行同步查询。 * * Query会从同步连接池中寻找一个空闲连接 * 当前线程会等待SQL执行完成。 */ auto result SakilaDatabase.Query( SELECT actor_id, first_name, last_name, last_update FROM actor WHERE actor_id 1 ); // 查询失败或没有结果 if (!result) { TC_LOG_ERROR(, select empty); SakilaDatabase.Close(); MySQL::Library_End(); return 1; } /* * QueryResult已经定位到第一行数据 * 可以通过下标获取不同字段。 */ TC_LOG_INFO( , actor_id%u, first_name%s, last_name%s, last_update%s, (*result)[0].GetUInt8(), (*result)[1].GetString(), (*result)[2].GetString(), (*result)[3].GetString() ); /* * 第六步关闭连接池。 * * 清理异步连接和同步连接 * 每个MySQLConnection析构时会关闭底层MYSQL连接。 */ SakilaDatabase.Close(); /* * 第七步释放MySQL客户端库。 * * 内部调用mysql_library_end() * 应当在所有数据库连接关闭后执行。 */ MySQL::Library_End(); return 0; }DatabaseLoader::Load()内部会依次执行两步bool DatabaseLoader::Load() { // 打开所有已注册的数据库连接池 if (!OpenDatabases()) { return false; } // 为所有连接准备预处理语句 if (!PrepareStatements()) { return false; } return true; }如果其中某个数据库连接池创建失败加载器还会关闭之前已经成功打开的数据库避免只创建一部分连接后直接退出。五、完整运行流程与注意事项连接池从启动到执行查询的完整流程如下调用MySQL::Library_Init() ↓ 创建DatabaseLoader ↓ 调用AddDatabase注册连接信息 ↓ SetConnectionInfo解析连接字符串 ↓ 创建异步连接 ↓ 每个异步连接绑定任务队列和工作线程 ↓ 创建同步连接 ↓ PrepareStatements准备预处理语句 ↓ 调用Query执行同步查询 ↓ 从同步连接池获取空闲连接 ↓ 执行SQL并返回结果 ↓ 连接解除占用继续留在连接池 ↓ 调用Close关闭全部连接 ↓ 调用MySQL::Library_End()使用时还需要注意以下问题。第一连接字符串中的字段顺序不能写错host;port;user;password;database第二示例中的密码应该替换为实际配置正式程序中不建议直接将数据库密码写死在源代码里可以放在配置文件或环境变量中。第三同步连接数量不能设置为0。同步调用Query()时需要从同步连接集合中获取连接如果没有同步连接就无法正常执行查询。第四执行完同步查询后底层代码必须及时解除连接锁。否则该连接会一直处于占用状态后续线程可能无法获得空闲连接。第五程序退出前应先关闭连接池再调用MySQL::Library_End();不能在数据库连接仍然存在时提前结束 MySQL 客户端库。这套连接池的核心可以总结为提前创建多个MySQL连接 将连接分成同步和异步两组 同步查询直接获取空闲连接执行 异步SQL通过任务队列交给后台线程 程序退出时统一关闭并释放所有连接。0voice · GitHub