C++ redis connection pool

学习teamtalk服务端源码，把redis连接池的实现记录下,用到了hiredis三方库和头文件。

整个redis缓存池是三个类实现的，redis-manager,redis-pool,redis-conn。

main函数获取redis-manager对象实例

CacheManager* CacheManager::getInstance()
{
    if (!s_cache_manager) {
        s_cache_manager = new CacheManager();
        if (s_cache_manager->Init()) {
            delete s_cache_manager;
            s_cache_manager = NULL;
        }
    }
 
    return s_cache_manager;
}

main() {
    //初始化redis
    CacheManager* pCacheManager = CacheManager::getInstance();
    if (!pCacheManager)
    {
        log("CacheManager init failed");
        return -1;
    }
}

通过redis-manager类init函数读取配置文件，创建对应的缓存池对象，并初始化缓存池，每个缓存池有一个缓存池名对应一个redis-pool指针加入到map中管理。

int CacheManager::Init()
{
    CConfigFileReader config_file("dbproxyserver.conf");
 
    //CacheInstances=unread,group_set,token,sync,group_member
    char* cache_instances = config_file.GetConfigName("CacheInstances");
    if (!cache_instances) {
        log("not configure CacheIntance");
        return 1;
    }
 
    char host[64];
    char port[64];
    char db[64];
    char maxconncnt[64];
    CStrExplode instances_name(cache_instances, ',');
    for (uint32_t i = 0; i < instances_name.GetItemCnt(); i++) {
        char* pool_name = instances_name.GetItem(i);
        //printf("%s", pool_name);
        snprintf(host, 64, "%s_host", pool_name);
        snprintf(port, 64, "%s_port", pool_name);
        snprintf(db, 64, "%s_db", pool_name);
        snprintf(maxconncnt, 64, "%s_maxconncnt", pool_name);
 
        char* cache_host = config_file.GetConfigName(host);
        char* str_cache_port = config_file.GetConfigName(port);
        char* str_cache_db = config_file.GetConfigName(db);
        char* str_max_conn_cnt = config_file.GetConfigName(maxconncnt);
        if (!cache_host || !str_cache_port || !str_cache_db || !str_max_conn_cnt) {
            log("not configure cache instance: %s", pool_name);
            return 2;
        }
 
        CachePool* pCachePool = new CachePool(pool_name, cache_host, atoi(str_cache_port),
                atoi(str_cache_db), atoi(str_max_conn_cnt));
        if (pCachePool->Init()) {
            log("Init cache pool failed");
            return 3;
        }
 
        m_cache_pool_map.insert(make_pair(pool_name, pCachePool));
    }
 
    return 0;
}

缓存池init的时候会创建redis连接对象，默认数量为2，并将创建的redis连接对象加入到缓存池的成员变量空闲redis连接数组中，等待被取用。

int CachePool::Init()
{
    for (int i = 0; i < m_cur_conn_cnt; i++) {
        CacheConn* pConn = new CacheConn(this);
        if (pConn->Init()) {
            delete pConn;
            return 1;
        }
 
        m_free_list.push_back(pConn);
    }
 
    log("cache pool: %s, list size: %lu", m_pool_name.c_str(), m_free_list.size());
    return 0;
}

redis连接对象init的时候主要用于连接的初始化和重连操作，每次对redis增删改查的之前都会调用init这个函数，每4秒尝试重连一次。

/*
 * redis初始化连接和重连操作，类似mysql_ping()
 */
int CacheConn::Init()
{
    if (m_pContext) {
        return 0;
    }
 
    // 4s 尝试重连一次
    uint64_t cur_time = (uint64_t)time(NULL);
    if (cur_time < m_last_connect_time + 4) {
        return 1;
    }
 
    m_last_connect_time = cur_time;
 
    // 200ms超时
    struct timeval timeout = {0, 200000};
    m_pContext = redisConnectWithTimeout(m_pCachePool->GetServerIP(), m_pCachePool->GetServerPort(), timeout);
    if (!m_pContext || m_pContext->err) {
        if (m_pContext) {
            log("redisConnect failed: %s", m_pContext->errstr);
            redisFree(m_pContext);
            m_pContext = NULL;
        } else {
            log("redisConnect failed");
        }
 
        return 1;
    }
 
    redisReply* reply = (redisReply *)redisCommand(m_pContext, "SELECT %d", m_pCachePool->GetDBNum());
    if (reply && (reply->type == REDIS_REPLY_STATUS) && (strncmp(reply->str, "OK", 2) == 0)) {
        freeReplyObject(reply);
        return 0;
    } else {
        log("select cache db failed");
        return 2;
    }
}

整个初始化的过程大概结束，下面就是通过redis-manager从缓存池中获取释放redis连接的过程。

缓存池名是通过配置文件中读入的，获取缓存池中的redis连接需要传入缓存池名，通过map获取到缓存池名对应的缓存池对象指针，并调用其成员函数getCacheConn获取redis连接。

CacheConn* CacheManager::GetCacheConn(const char* pool_name)
{
    map<string, CachePool*>::iterator it = m_cache_pool_map.find(pool_name);
    if (it != m_cache_pool_map.end()) {
        return it->second->GetCacheConn();
    } else {
        return NULL;
    }
}

调用缓存池CachePool成员函数getCacheConn获取redis连接这块代码有可学之处，他用了pthread库里面的信号量来优雅地等待空闲连接数组中的连接对象，为了线程安全需要在访问空闲连接数组的时候加锁，不断判断数组中是否存在可用的redis连接对象，如果没有可用的连接，则将判断此时正在使用的redis连接对象个数是否超过指定最大值，超过则将用信号量阻塞，直到有空闲的连接被唤醒。如果此时连接数小于指定最大值，则将继续创建redis连接对象并初始化一个可用的redis连接对象加入到空闲redis连接数组中。如果数组中有可用的redis连接则将从数组头部取出一个元素并返回。

CacheConn* CachePool::GetCacheConn()
{
    m_free_notify.Lock();
 
    while (m_free_list.empty()) {
        if (m_cur_conn_cnt >= m_max_conn_cnt) {
            m_free_notify.Wait();
        } else {
            CacheConn* pCacheConn = new CacheConn(this);
            int ret = pCacheConn->Init();
            if (ret) {
                log("Init CacheConn failed");
                delete pCacheConn;
                m_free_notify.Unlock();
                return NULL;
            } else {
                m_free_list.push_back(pCacheConn);
                m_cur_conn_cnt++;
                log("new cache connection: %s, conn_cnt: %d", m_pool_name.c_str(), m_cur_conn_cnt);
            }
        }
    }
 
    CacheConn* pConn = m_free_list.front();
    m_free_list.pop_front();
 
    m_free_notify.Unlock();
 
    return pConn;
}

释放redis连接对象，传入需要释放的redis连接对象指针，加锁，遍历redis连接对象数组中是否存在该连接对象，不存在则将连接对象加入到该空闲redis连接数组中，信号量通知唤醒被阻塞的线程，redis连接数组中有空闲连接可用。

void CachePool::RelCacheConn(CacheConn* pCacheConn)
{
    m_free_notify.Lock();
 
    list<CacheConn*>::iterator it = m_free_list.begin();
    for (; it != m_free_list.end(); it++) {
        if (*it == pCacheConn) {
            break;
        }
    }
 
    if (it == m_free_list.end()) {
        m_free_list.push_back(pCacheConn);
    }
 
    m_free_notify.Signal();
    m_free_notify.Unlock();
}