1、 bool disconnect（）; bool publish（const std:string &channel_name, const std:message）; private: / 下面三个回调函数供redis服务调用 / 连接回调 static void connect_callback（const redisAsyncContext *redis_context, int status）; / 断开连接的回调 static void disconnect_callback（const redisAsyncContext *redis_context, int status
2、）; / 执行命令回调 static void command_callback（redisAsyncContext *redis_context, void *reply, void *privdata）; / 事件分发线程函数 static void *event_thread（void *data）; void *event_proc（）; / libevent事件对象 event_base *_event_base; / 事件线程ID pthread_t _event_thread; / 事件线程的信号量 sem_t _event_sem; / hiredis异步对象 redisAsy
3、ncContext *_redis_context; ; #endif redis_publisher.cpp redis_publisher.cpp & */ #include #include #include #include redis_publisher.h CRedisPublisher:CRedisPublisher（）:_event_base（0）, _event_thread（0）, _redis_context（0） CRedisPublisher:CRedisPublisher（） bool CRedisPublisher:init（） / initialize the
4、event _event_base = event_base_new（）; / 创建libevent对象 if （NULL = _event_base） printf（ Create redis event failed.n）; return false; memset（&_event_sem, 0, sizeof（_event_sem）; int ret = sem_init（&_event_sem, 0, 0）; if （ret != 0） printf（ Init sem failed.n return true; bool CRedisPublisher:uninit（） _event
5、_base = NULL; sem_destroy（&_event_sem）; return true;connect（） / connect redis _redis_context = redisAsyncConnect（127.0.0.1, 6379）; / 异步连接到redis服务器上，使用默认端口 if （NULL = _redis_context） printf（ Connect redis failed.n if （_redis_context-&err） printf（ Connect redis error: %d, %sn, _redis_context-&err, _re
6、dis_context-&errstr）; / 输出错误信息 return false; / attach the event redisLibeventAttach（_redis_context, _event_base）; / 将事件绑定到redis context上，使设置给redis的回调跟事件关联 / 创建事件处理线程 int ret = pthread_create（&_event_thread, 0, &CRedisPublisher:event_thread, this）; create event thread failed.n disconnect（）; / 设置连接回调，
7、当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态 redisAsyncSetConnectCallback（_redis_context, &connect_callback）; / 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连 redisAsyncSetDisconnectCallback（_redis_context, &disconnect_callback）; / 启动事件线程 sem_post（&disconnect（） if （_redis_context） redisAsyncDisconnect（_redis_co
8、ntext）; redisAsyncFree（_redis_context）; _redis_context = NULL;publish（const std:message） int ret = redisAsyncCommand（_redis_context, &command_callback, this, PUBLISH %s %s, channel_name.c_str（）, message.c_str（）; if （REDIS_ERR = ret） printf（Publish command failed: %dn, ret）; void CRedisPublisher:conn
9、ect_callback（const redisAsyncContext *redis_context, int status） if （status != REDIS_OK） printf（ Error: %sn, redis_context-& else printf（ Redis connected!n void CRedisPublisher:disconnect_callback（ const redisAsyncContext *redis_context, int status） if （status != REDIS_OK） / 这里异常退出，可以尝试重连 printf（ /
10、消息接收回调函数 void CRedisPublisher:command_callback（redisAsyncContext *redis_context, void *reply, void *privdata） printf（command callback.n / 这里不执行任何操作 void *CRedisPublisher:event_thread（void *data） if （NULL = data） printf（ Error! assert（false）; return NULL; CRedisPublisher *self_this = reinterpret_cast
11、（data）; return self_this-&event_proc（）; void *CRedisPublisher:event_proc（） sem_wait（& / 开启事件分发，event_base_dispatch会阻塞 event_base_dispatch（_event_base）; redis_subscriber.h redis_subscriber.h & 封装hiredis，实现消息订阅redis功能 */ #ifndef REDIS_SUBSCRIBER_H #define REDIS_SUBSCRIBER_H #include #include #include
12、#include #include #include #include #include #include class CRedisSubscriber public: typedef std:tr1:function NotifyMessageFn; / 回调函数对象类型，当接收到消息后调用回调把消息发送出去 CRedisSubscriber（）; CRedisSubscriber（）; bool init（const NotifyMessageFn &fn）; / 传入回调对象 bool uninit（）; / 可以多次调用，订阅多个频道 bool subscribe（const std:
13、channel_name）; / 通知外层的回调函数对象 NotifyMessageFn _notify_message_fn; #endif redis_subscriber.cpp: /* & redis_subscriber.cpp &redis_subscriber.h CRedisSubscriber:CRedisSubscriber（）:_event_base（0）, _event_thread（0）, _redis_context（0） CRedisSubscriber:CRedisSubscriber（） bool CRedisSubscriber:init（const Not
14、ifyMessageFn &fn） / initialize the event _notify_message_fn = fn; _event_base = event_base_new（）; bool CRedisSubscriber:CRedisSubscriber: bool CRedisSubscriber:subscribe（const std:channel_name） int ret = redisAsyncCommand（_redis_context, &SUBSCRIBE %s, channel_name.c_str（）;Subscribe command failed:
15、printf（ Subscribe success: void CRedisSubscriber: void CRedisSubscriber: / 消息接收回调函数 void CRedisSubscriber:command_callback（redisAsyncContext *redis_context, void *reply, void *privdata） if （NULL = reply | NULL = privdata） return ; / 静态函数中，要使用类的成员变量，把当前的this指针传进来，用this指针间接访问 CRedisSubscriber *self_th
16、is = reinterpret_cast（privdata）; redisReply *redis_reply = reinterpret_cast（reply）; / 订阅接收到的消息是一个带三元素的数组 if （redis_reply-&type = REDIS_REPLY_ARRAY & redis_reply-&elements = 3） printf（ Recieve message:%s:%d:%dn, redis_reply-&element0-&str, redis_reply-&len, redis_reply-&element1-&element2-&len）; / 调用函数对象把消息通知给外层 self_this-&_notify_message_fn（redis_reply-& void *CRedisSubscriber:event_thread
