2025/11/21：
这段时间写http框架项目，先对项目中需要用到的数据库连接池做构建
具体的原理部分可以参考 数据库连接池原理

系列文章：

数据库连接池

代码详解http请求报文解析流程

http框架路由匹配算法实现

session实现和http server类最终组装

参考：kama httpserver
复现：TinyHttpServer

一、连接池架构思考

在进行数据库原理学习之后，思考一下数据库需要什么组件。

连接池主要需要包括以下对象：

• 连接队列，与之相关的连接池大小以及初始化连接数量
• 动态生产连接与删除连接的线程（非必要）
• 与数据库进行连接的各种必要参数
• 对外暴露getConnection连接提供接口

连接队列中的每个连接所涉及到的对象就比较简单了：

• 查询、update相关操作具体函数
• 管理一条具体的连接MYSQL*

简单分析一下同步互斥的关系：

• 对连接队列的操作需要互斥，涉及到的对象有动态操作线程、getConnection方法以及最后的连接池析构操作
• 每个对连接队列的操作结束之后（除了最后的析构），都需要对条件变量做一个notify

二、模块具体分析

单条connection设计如下，基本上和上面讲到的一样，可以思考一下连接内部的计时器是如何被动态删减连接线程所利用的。

• 新建连接以及查询操作结束后计时器刷新

class sqlConnection
{
public:
    sqlConnection(const std::string& host,
                const std::string& user,
                const std::string& password,
                const std::string& database,
                const unsigned int& port);
    ~sqlConnection();

    // 禁止拷贝
    sqlConnection(const sqlConnection&) = delete;
    sqlConnection& operator=(const sqlConnection&) = delete;

    // 执行查询操作，返回结果集指针。调用方需使用 mysql_free_result 释放。
    MYSQL_RES* executeQuery(const std::string& sql) const;

    // 执行更新操作，返回受影响行数，失败返回 -1。
    int executeUpdate(const std::string& sql) const;

    // 返回空闲时间(ms) —— 距离上次 refreshTime 的时间差
    int getConnIdleTime() const;

    // 重置内部计时器
    void refreshTime();

    // 判断连接是否有效
    bool isValid() const { return conn_ != nullptr; }

private:
    MYSQL* conn_;
    std::chrono::system_clock::time_point executeTime_;
};

具体代码挺简单的，其实没什么好说的：

#include "../../../include/utils/db/sqlConnection.h"


sqlConnection::sqlConnection(
    const std::string& host,
    const std::string& user,
    const std::string& password,
    const std::string& database,
    const unsigned int& port)
{
    // 初始化 MYSQL 句柄
    MYSQL* handle = mysql_init(nullptr);
    if (handle == nullptr)
    {
        LOG_ERROR << "mysql_init failed";
        conn_ = nullptr;
        return;
    }
    // 建立数据库连接
    conn_ = mysql_real_connect(handle, host.c_str(), user.c_str(),
                                password.c_str(), database.c_str(), port, nullptr, 0);
    if (conn_ == nullptr)
    {
        LOG_ERROR << "Failed to connect to database: " << mysql_error(handle);
        mysql_close(handle);
        return;
    }
    // 设置当前时间
    refreshTime();
}

sqlConnection::~sqlConnection()
{
    if (conn_ != nullptr)
    {
        mysql_close(conn_);
        conn_ = nullptr;
    }
}

MYSQL_RES* sqlConnection::executeQuery(const std::string& sql) const
{
    if (conn_ == nullptr)
    {
        LOG_ERROR << "executeQuery called on invalid connection";
        return nullptr;
    }
    if (mysql_query(conn_, sql.c_str()) != 0)
    {
        LOG_ERROR << "Query failed: " << mysql_error(conn_);
        return nullptr;
    }
    MYSQL_RES* result = mysql_store_result(conn_);
    return result; // 由调用方释放 mysql_free_result(result)
}

int sqlConnection::executeUpdate(const std::string& sql) const
{
    if (conn_ == nullptr)
    {
        LOG_ERROR << "executeUpdate called on invalid connection";
        return -1;
    }
    if (mysql_query(conn_, sql.c_str()) != 0)
    {
        LOG_ERROR << "Update failed: " << mysql_error(conn_);
        return -1;
    }
    return mysql_affected_rows(conn_);
}

int sqlConnection::getConnIdleTime() const
{
    // 返回距上次刷新时间的毫秒数
    auto now = std::chrono::system_clock::now();
    auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(now - executeTime_).count();
    return static_cast<int>(diff);
}

void sqlConnection::refreshTime()
{
    executeTime_ = std::chrono::system_clock::now();
}

同时，连接池的内部对象如下：

#pragma once
#include <atomic>

#include "sqlConnection.h"
#include <condition_variable>
#include <nlohmann/json.hpp>
#include <fstream>
#include <queue>
#include <thread>

class sqlConnectionPool
{
public:
    static sqlConnectionPool& getInstance();

    sqlConnectionPool(const sqlConnectionPool&) = delete;
    sqlConnectionPool& operator=(const sqlConnectionPool&) = delete;

    ~sqlConnectionPool();

    std::shared_ptr<sqlConnection> getConnection();

    // 删除闲置连接
    void connDeletor();
    // 生产新连接
    void connProducer();

private:
    sqlConnectionPool();

    // 数据库连接参数
    std::string host_;
    std::string user_;
    std::string password_;
    std::string database_;
    unsigned int port_;

    // 连接池
    std::queue<std::shared_ptr<sqlConnection>> connQueue_;
    std::mutex mtx_;
    std::size_t poolSize_;
    std::size_t initPoolSize_;
    std::condition_variable cv_;

    int connTimeOut_;
    int maxIdleTime_;

    std::atomic<int> qSize_{0};

    std::atomic<bool> run_{true};

    std::thread producer_;
    std::thread deletor_;
};

可以看到，因为连接池需要被多个线程所操作，因此这里使用了懒加载的方式来构造单例对象。此外，为什么另外使用一个qSize_原子类型来进行连接的记录，这是因为该变量记录的连接数量是连接的总量，与队列的size基本无关（队列中的连接会被取出，但是总连接数不会因此变化）。该原子类型主要用于动态增删连接线程中的连接数量控制。

下面这部分代码涉及到了单例对象的构造，具体的对象构造通过一个json配置文件来实现——有现成的json字符串解析方法，为什么还要自己搓正则表达式呢？

#include "utils/db/sqlConnectionPool.h"

sqlConnectionPool& sqlConnectionPool::getInstance()
{
    static sqlConnectionPool instance;
    return instance;
}

sqlConnectionPool::sqlConnectionPool()
{
    LOG_INFO << "create sqlConnectionPool";
    // 读取配置json文件 在项目目录下
    std::fstream cfgFile("../dbconfig.json");

    if (!cfgFile.is_open())
    {
        LOG_ERROR << "database config file not found";
    }

    try
    {
        nlohmann::json configJson;
        cfgFile >> configJson;
        std::cout << configJson << std::endl;

        host_ = configJson["host"].get<std::string>();
        port_ = configJson["port"].get<unsigned int>();
        user_ = configJson["user"].get<std::string>();
        password_ = configJson["password"].get<std::string>();
        initPoolSize_ = configJson["initPoolSize"].get<int>();
        poolSize_ = configJson["poolSize"].get<int>();
        connTimeOut_ = configJson["connTimeout"].get<int>();
        maxIdleTime_ = configJson["maxIdleTime"].get<int>();
        database_ = configJson["database"].get<std::string>();
    }
    catch (std::exception& e)
    {
        LOG_ERROR << e.what();
    }

    for (int i = 0; i < initPoolSize_; i++)
    {
        auto conn = std::make_shared<sqlConnection>(host_, user_, password_, database_, port_);
        connQueue_.push(conn);
        qSize_.fetch_add(1);
    }

    producer_ = std::thread(&sqlConnectionPool::connProducer, this);
    deletor_ = std::thread(&sqlConnectionPool::connDeletor, this);
}

具体的json文件内容就不必写了，就是对应字段设置对应值罢了。

下面是连接池最核心的函数。可以看到这里对连接池的退出状态做了一下判断。除此之外，最核心的逻辑在于返回值的构建。

std::shared_ptr<sqlConnection> sqlConnectionPool::getConnection()
{
    if (!run_.load()) return nullptr;
    std::unique_lock<std::mutex> lock(mtx_);

    // 等待队列非空或线程要退出
    cv_.wait(lock, [this] { return !connQueue_.empty() || !run_.load(); });

    if (!run_.load() && connQueue_.empty()) return nullptr;

    auto conn = connQueue_.front();
    connQueue_.pop();

    // 返回连接，将其删除器改为向队列中归还连接
    std::shared_ptr<sqlConnection> sp(conn.get(),
           [this, conn](sqlConnection*) {
               std::lock_guard<std::mutex> lock(mtx_);
               connQueue_.push(conn);
               cv_.notify_one();
           });
    // 通知其他线程，不然会造成死锁
    cv_.notify_all();
    return sp;
}

返回值的构建核心在于定义了一个智能指针的删除器，删除器中捕获（复制）了一个指向连接对象的指针，在出该函数作用域的时候，原conn指针被析构，但是其指向的连接对象不被析构，因为捕获的conn指针指向的控制块与原conn指向的是同一个，此时控制块的count计数不为0。

一定需要传入conn的复制，也就是share指针吗？不可以直接取指针，然后删除器定义为将该指针归还吗？也就是如下的逻辑是否可以：

std::shared_ptr<sqlConnection> sp(conn.get(),
           [this](sqlConnection*) {
               std::lock_guard<std::mutex> lock(mtx_);
               connQueue_.push(用this的get新构造的share指针对象);
               cv_.notify_one();
           });

由于自定义删除器的存在，返回的智能指针sp指向的控制块与conn及其复制不同。shareptr的删除逻辑在于，当前控制块的count为0时删除指针所指向的对象。向lambda表达式传入conn使得该控制块的count始终不为0。如果构造方式如上，这就意味着在出getconnection作用域的同时，conn析构导致其控制块count为0，那么它所指向的对象就被删除了。

生产者和删除者线程其实没什么好说的，主要是需要注意一下其间的同步互斥关系：

void sqlConnectionPool::connProducer()
{
    while (run_.load())
    {
        std::unique_lock<std::mutex> lock(mtx_);
        // 等待队列为空或线程要退出
        cv_.wait(lock, [this] { return connQueue_.empty() || !run_.load(); });

        if (!run_.load()) break;

        if (qSize_.load() < poolSize_)
        {
            auto conn = std::make_shared<sqlConnection>(host_, user_, password_, database_, port_);
            connQueue_.push(conn);
            qSize_.fetch_add(1);
        }

        // 通知消费者线程，可以消费连接了
        cv_.notify_all();
    }
}

sqlConnectionPool::~sqlConnectionPool()
{
    LOG_INFO << "destroy sqlConnections";
    run_.store(false);

    // 唤醒所有等待的线程/消费者
    cv_.notify_all();

    // 等待后台线程结束
    if (producer_.joinable()) producer_.join();
    if (deletor_.joinable()) deletor_.join();

    // 清理连接队列
    std::unique_lock<std::mutex> lock(mtx_);
    while (!connQueue_.empty())
    {
        connQueue_.pop();
    }

    // 通知线程结束
    cv_.notify_all();
}

三、连接池测试

测试文件内容如下：

#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <atomic>
#include <fstream>
#include <nlohmann/json.hpp>
#include <mysql/mysql.h>
#include "../include/utils/db/sqlConnectionPool.h"
#include "../include/utils/db/sqlConnection.h"

struct DBConfig {
    std::string host;
    std::string user;
    std::string password;
    std::string database;
    unsigned int port{0};
};

// 读取配置文件（与连接池保持一致路径）
DBConfig loadConfig(const std::string& path = "../dbconfig.json") {
    DBConfig cfg;
    std::fstream cfgFile(path);
    if (!cfgFile.is_open()) {
        std::cerr << "[ERROR] Cannot open config file: " << path << std::endl;
        return cfg;
    }
    try {
        nlohmann::json j; cfgFile >> j;
        cfg.host = j["host"].get<std::string>();
        cfg.user = j["user"].get<std::string>();
        cfg.password = j["password"].get<std::string>();
        cfg.database = j["database"].get<std::string>();
        cfg.port = j["port"].get<unsigned int>();
    } catch (std::exception& e) {
        std::cerr << "[ERROR] Parse config failed: " << e.what() << std::endl;
    }
    return cfg;
}

// 使用连接池进行多线程查询
long long benchmarkWithPool(int threads, int queriesPerThread) {
    auto& pool = sqlConnectionPool::getInstance();
    std::atomic<int> okCount{0};
    auto start = std::chrono::steady_clock::now();
    std::vector<std::thread> ths;
    ths.reserve(threads);
    for (int t = 0; t < threads; ++t) {
        ths.emplace_back([queriesPerThread, &okCount]() {
            for (int i = 0; i < queriesPerThread; ++i) {
                auto conn = sqlConnectionPool::getInstance().getConnection();
                if (!conn || !conn->isValid()) continue;
                MYSQL_RES* res = conn->executeQuery("SELECT 1");
                if (res) {
                    mysql_free_result(res);
                    okCount.fetch_add(1, std::memory_order_relaxed);
                }
                conn->refreshTime();
            }
        });
    }
    for (auto& th : ths) th.join();
    auto end = std::chrono::steady_clock::now();
    long long ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "[Pool] Executed " << okCount.load() << " successful queries in " << ms << " ms" << std::endl;
    return ms;
}

// 不使用连接池：每次查询都新建+销毁一个连接（最差情况）
long long benchmarkWithoutPool(const DBConfig& cfg, int threads, int queriesPerThread) {
    std::atomic<int> okCount{0};
    auto start = std::chrono::steady_clock::now();
    std::vector<std::thread> ths;
    ths.reserve(threads);
    for (int t = 0; t < threads; ++t) {
        ths.emplace_back([&cfg, queriesPerThread, &okCount]() {
            for (int i = 0; i < queriesPerThread; ++i) {
                sqlConnection conn(cfg.host, cfg.user, cfg.password, cfg.database, cfg.port);
                if (!conn.isValid()) continue;
                MYSQL_RES* res = conn.executeQuery("SELECT 1");
                if (res) {
                    mysql_free_result(res);
                    okCount.fetch_add(1, std::memory_order_relaxed);
                }
            }
        });
    }
    for (auto& th : ths) th.join();
    auto end = std::chrono::steady_clock::now();
    long long ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "[NoPool] Executed " << okCount.load() << " successful queries in " << ms << " ms" << std::endl;
    return ms;
}

int main() {
    int threads = 8;
    int queriesPerThread = 50;

    std::cout << "=== Connection Pool Benchmark ===" << std::endl;

    DBConfig cfg = loadConfig();
    if (cfg.host.empty()) {
        std::cerr << "[ERROR] Config invalid, abort benchmark" << std::endl;
        return 1;
    }

    auto warmConn = sqlConnectionPool::getInstance().getConnection();
    if (!warmConn || !warmConn->isValid()) {
        std::cerr << "[ERROR] Failed to get initial pooled connection" << std::endl;
        return 2;
    }
    MYSQL_RES* warmRes = warmConn->executeQuery("SELECT 1");
    if (warmRes) mysql_free_result(warmRes);

    // 归还连接（暂时解决死锁）
    // warmConn.reset();

    long long poolMs = benchmarkWithPool(threads, queriesPerThread);
    long long noPoolMs = benchmarkWithoutPool(cfg, threads, queriesPerThread);

    std::cout << "-------------------------------------" << std::endl;
    std::cout << "Threads: " << threads << ", Queries/Thread: " << queriesPerThread << std::endl;
    std::cout << "With Pool Time:    " << poolMs << " ms" << std::endl;
    std::cout << "Without Pool Time: " << noPoolMs << " ms" << std::endl;
    if (poolMs > 0) {
        std::cout << "Speedup (NoPool/Pool): " << std::fixed << (double)noPoolMs / (double)poolMs << "x" << std::endl;
    }

    return 0;
}

结果如下：

四、测试死锁问题分析

在测试过程中，出现了一个死锁问题。具体来说，当"initPoolSize"配置为1时，在执行benchmarkwithpool函数的时候，进程阻塞在了

auto conn = sqlConnectionPool::getInstance().getConnection();

这个位置，打点调试之后，发现在这个地方阻塞时连接池中的连接为0，但是qSize为1。这就代表之前有一个连接被取出之后不再创建新连接了。在之前取出的连接有啥呢，就是在main函数中用到的

auto warmConn = sqlConnectionPool::getInstance().getConnection();

因为使用了单例模式，因此这里用到的对象和benchmark中应该是同一个。试着在调用benchmark之前将该连接reset了，结果显示正常连接。但是显然逻辑还是不太正常，前面的问题说明多线程状态下这个连接池不太靠谱。

把reset去掉之后继续调试问题，发现生产者线程根本没启动，一直堵在等待条件变量这个位置（此时benchmark中的getconnection因为队列中没有连接也堵塞中了）。

为什么reset之后就可以了，不reset就启动不了呢？

思考一下就发现问题所在了。reset之后，归还连接会导致一个条件信号通知，这时在benchmark中获取连接正常调用（生产者不动，因为此时连接队列不为空），然后就回归了正常的多线程调用流程。如果没有reset，此时因为第一条连接未归还，全局中没有条件变量被触发且队列为空，benchmark中的getconnection以及生产者都不会被触发（两者不被触发的原因不同），但是这时候我们是希望生产者被触发的，重点关于条件变量的通知与否，因此可以定位到原因在getconnectrion在取出连接之后并没有进行一次notify_all。下面展示函数修正前后区别：

    // 返回连接，将其删除器改为向队列中归还连接
    std::shared_ptr<sqlConnection> sp(conn.get(),
           [this, conn](sqlConnection*) {
               std::lock_guard<std::mutex> lock(mtx_);
               connQueue_.push(conn);
               cv_.notify_one();
           });
    // 通知其他线程，不然会造成死锁，这句是解决死锁的关键代码
    cv_.notify_all();
    return sp;