Redis客户端连接方式Hiredis简单封装使用,连接池、屏蔽连接细节

工作需要对Hiredis进行了简单封装，实现功能：

1、API进行统一，对外只提供一个接口；

2、屏蔽上层应用对连接的细节处理；

3、底层采用队列的方式保持连接池，保存连接会话；

4、重连时采用时间戳进行控制，每隔一定时间(3s)重连一次，防止频繁重试造成的不必要浪费。

先看一下Hiredis的常用数据结构与API：

//hiredis/hiredis.h
/* Context for a connection to Redis */
typedef struct redisContext {
int err; /* Error flags, 0 when there is no error */
char errstr[128]; /* String representation of error when applicable */
int fd;
int flags;
char *obuf; /* Write buffer */
redisReader *reader; /* Protocol reader */
} redisContext;

/* This is the reply object returned by redisCommand() */
#define REDIS_REPLY_STRING 1
#define REDIS_REPLY_ARRAY 2
#define REDIS_REPLY_INTEGER 3
#define REDIS_REPLY_NIL 4
#define REDIS_REPLY_STATUS 5
#define REDIS_REPLY_ERROR 6
typedef struct redisReply {
int type; /* REDIS_REPLY_* */
long long integer; /* The integer when type is REDIS_REPLY_INTEGER */
int len; /* Length of string */
char *str; /* Used for both REDIS_REPLY_ERROR and REDIS_REPLY_STRING */
size_t elements; /* number of elements, for REDIS_REPLY_ARRAY */
struct redisReply **element; /* elements vector for REDIS_REPLY_ARRAY */
} redisReply;

redisContext *redisConnectWithTimeout(const char *ip, int port, struct timeval tv);
void redisFree(redisContext *c);

下面直接上封装后的代码：

class KGRedisClient
{
public:
KGRedisClient(string ip, int port, int timeout = 2000);
virtual ~KGRedisClient();

bool ExecuteCmd(const char *cmd, size_t len, string &response);
redisReply* ExecuteCmd(const char *cmd, size_t len);

private:
int m_timeout;
int m_serverPort;
string m_setverIp;
CCriticalSection m_lock;
std::queue<redisContext *> m_clients;

time_t m_beginInvalidTime;
static const int m_maxReconnectInterval = 3;

redisContext* CreateContext();
void ReleaseContext(redisContext *ctx, bool active);
bool CheckStatus(redisContext *ctx);
};

KGRedisClient::KGRedisClient(string ip, int port, int timeout)
{
m_timeout = timeout;
m_serverPort = port;
m_setverIp = ip;

m_beginInvalidTime = 0;
}

KGRedisClient::~KGRedisClient()
{
CAutoLock autolock(m_lock);
while(!m_clients.empty())
{
redisContext *ctx = m_clients.front();
redisFree(ctx);
m_clients.pop();
}
}

bool KGRedisClient::ExecuteCmd(const char *cmd, size_t len,string &response)
{
redisReply *reply = ExecuteCmd(cmd, len);
if(reply == NULL) return false;

boost::shared_ptr<redisReply> autoFree(reply, freeReplyObject);
if(reply->type == REDIS_REPLY_INTEGER)
{
response = _IntToStrA(reply->integer);
return true;
}
else if(reply->type == REDIS_REPLY_STRING)
{
response.assign(reply->str, reply->len);
return true;
}
else if(reply->type == REDIS_REPLY_STATUS)
{
response.assign(reply->str, reply->len);
return true;
}
else if(reply->type == REDIS_REPLY_NIL)
{
response = "";
return true;
}
else if(reply->type == REDIS_REPLY_ERROR)
{
response.assign(reply->str, reply->len);
return false;
}
else if(reply->type == REDIS_REPLY_ARRAY)
{
response = "Not Support Array Result!!!";
return false;
}
else
{
response = "Undefine Reply Type";
return false;
}
}

redisReply* KGRedisClient::ExecuteCmd(const char *cmd, size_t len)
{
redisContext *ctx = CreateContext();
if(ctx == NULL) return NULL;

redisReply *reply = (redisReply*)redisCommand(ctx, "%b", cmd, len);

ReleaseContext(ctx, reply != NULL);

return reply;
}

redisContext* KGRedisClient::CreateContext()
{
{
CAutoLock autolock(m_lock);
if(!m_clients.empty())
{
redisContext *ctx = m_clients.front();
m_clients.pop();

return ctx;
}
}

time_t now = time(NULL);
if(now < m_beginInvalidTime + m_maxReconnectInterval) return NULL;

struct timeval tv;
tv.tv_sec = m_timeout / 1000;
tv.tv_usec = (m_timeout % 1000) * 1000;;
redisContext *ctx = redisConnectWithTimeout(m_setverIp.c_str(), m_serverPort, tv);
if(ctx == NULL || ctx->err != 0)
{
if(ctx != NULL) redisFree(ctx);

m_beginInvalidTime = time(NULL);

return NULL;
}

return ctx;
}

void KGRedisClient::ReleaseContext(redisContext *ctx, bool active)
{
if(ctx == NULL) return;
if(!active) {redisFree(ctx); return;}

CAutoLock autolock(m_lock);
m_clients.push(ctx);
}

bool KGRedisClient::CheckStatus(redisContext *ctx)
{
redisReply *reply = (redisReply*)redisCommand(ctx, "ping");
if(reply == NULL) return false;

boost::shared_ptr<redisReply> autoFree(reply, freeReplyObject);

if(reply->type != REDIS_REPLY_STATUS) return false;
if(strcasecmp(reply->str,"PONG") != 0) return false;

return true;
}

稍加解释：

成员变量：m_clients用于保存连接池。

成员变量：m_beginInvalidTime、m_maxReconnectInterval 用于控制断掉时的频繁连接。

对外API：ExecuteCmd(const char *cmd, string &response);