工作需要对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); //Issue a command to Redis, NULL if error, otherwise reply void *redisCommand(redisContext *c, const char *format, ...); /* Function to free the reply objects hiredis returns by default. */ void freeReplyObject(void *reply);
下面直接上封装后的代码:
class KGRedisClient { public: KGRedisClient(string ip, int port, int timeout = 2000); virtual ~KGRedisClient(); bool ExecuteCmd(const char *cmd, string &response); redisReply* ExecuteCmd(const char *cmd); 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, string &response) { redisReply *reply = ExecuteCmd(cmd); 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) { redisContext *ctx = CreateContext(); if(ctx == NULL) return NULL; redisReply *reply = (redisReply*)redisCommand(ctx, cmd); 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);
时间: 2024-10-10 14:13:28