linux下使用hiredis异步API实现sub/pub消息订阅和发布的功能

 

       最近使用redis的c接口——hiredis，使客户端与redis服务器通信，实现消息订阅和发布（PUB/SUB）的功能，我把遇到的一些问题和解决方法列出来供大家学习。

redis_publisher.h

[cpp] view plain copy

1. /************************************************************************* 
2.     > File Name: redis_publisher.h 
3.     > Created Time: Sat 23 Apr 2016 10:15:09 PM CST 
4.     > Description: 封装hiredis，实现消息发布给redis功能 
5.  ************************************************************************/  
6.   
7. #ifndef REDIS_PUBLISHER_H  
8. #define REDIS_PUBLISHER_H  
9.   
10. #include   
11. #include   
12. #include   
13. #include   
14. #include   
15. #include   
16. #include   
17. #include   
18. #include   
19.   
20. class CRedisPublisher  
21. {  
22. public:      
23.     CRedisPublisher();  
24.     ~CRedisPublisher();  
25.   
26.     bool init();  
27.     bool uninit();  
28.     bool connect();  
29.     bool disconnect();  
30.       
31.     bool publish(const std::string &channel_name,   
32.         const std::string &message);  
33.   
34. private:  
35.      // 下面三个回调函数供redis服务调用  
36.     // 连接回调  
37.     static void connect_callback(const redisAsyncContext *redis_context,  
38.         int status);  
39.       
40.     // 断开连接的回调  
41.     static void disconnect_callback(const redisAsyncContext *redis_context,  
42.         int status);  
43.   
44.     // 执行命令回调  
45.     static void command_callback(redisAsyncContext *redis_context,  
46.         void *reply, void *privdata);  
47.   
48.     // 事件分发线程函数  
49.     static void *event_thread(void *data);  
50.     void *event_proc();  
51.   
52. private:  
53.      // libevent事件对象  
54.     event_base *_event_base;  
55.     // 事件线程ID  
56.     pthread_t _event_thread;  
57.     // 事件线程的信号量  
58.     sem_t _event_sem;  
59.     // hiredis异步对象  
60.     redisAsyncContext *_redis_context;  
61. };  
62.   
63. #endif  

redis_publisher.cpp

[cpp] view plain copy

1. /************************************************************************* 
2.     > File Name: redis_publisher.cpp 
3.     > Created Time: Sat 23 Apr 2016 10:15:09 PM CST 
4.     > Description:  
5.  ************************************************************************/  
6.    
7. #include   
8. #include   
9. #include   
10. #include "redis_publisher.h"  
11.   
12. CRedisPublisher::CRedisPublisher():_event_base(0), _event_thread(0),  
13. _redis_context(0)  
14. {  
15. }  
16.   
17. CRedisPublisher::~CRedisPublisher()  
18. {  
19. }  
20.   
21. bool CRedisPublisher::init()  
22. {  
23.     // initialize the event  
24.     _event_base = event_base_new();    // 创建libevent对象  
25.     if (NULL == _event_base)  
26.     {  
27.         printf(": Create redis event failed.\n");  
28.         return false;  
29.     }  
30.   
31.     memset(&_event_sem, 0, sizeof(_event_sem));  
32.     int ret = sem_init(&_event_sem, 0, 0);  
33.     if (ret != 0)  
34.     {  
35.         printf(": Init sem failed.\n");  
36.         return false;  
37.     }  
38.   
39.     return true;  
40. }  
41.   
42. bool CRedisPublisher::uninit()  
43. {  
44.     _event_base = NULL;  
45.       
46.     sem_destroy(&_event_sem);     
47.     return true;  
48. }  
49.   
50. bool CRedisPublisher::connect()  
51. {  
52.     // connect redis  
53.     _redis_context = redisAsyncConnect("127.0.0.1", 6379);    // 异步连接到redis服务器上，使用默认端口  
54.     if (NULL == _redis_context)  
55.     {  
56.         printf(": Connect redis failed.\n");  
57.         return false;  
58.     }  
59.   
60.     if (_redis_context->err)  
61.     {  
62.         printf(": Connect redis error: %d, %s\n",   
63.             _redis_context->err, _redis_context->errstr);    // 输出错误信息  
64.         return false;  
65.     }  
66.   
67.     // attach the event  
68.     redisLibeventAttach(_redis_context, _event_base);    // 将事件绑定到redis context上，使设置给redis的回调跟事件关联  
69.       
70.     // 创建事件处理线程  
71.     int ret = pthread_create(&_event_thread, 0, &CRedisPublisher::event_thread, this);  
72.     if (ret != 0)  
73.     {  
74.         printf(": create event thread failed.\n");  
75.         disconnect();  
76.         return false;  
77.     }  
78.   
79.     // 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态  
80.     redisAsyncSetConnectCallback(_redis_context,   
81.         &CRedisPublisher::connect_callback);  
82.   
83.     // 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连  
84.     redisAsyncSetDisconnectCallback(_redis_context,  
85.         &CRedisPublisher::disconnect_callback);  
86.   
87.     // 启动事件线程  
88.     sem_post(&_event_sem);  
89.     return true;  
90. }  
91.   
92. bool CRedisPublisher::disconnect()  
93. {  
94.     if (_redis_context)  
95.     {  
96.         redisAsyncDisconnect(_redis_context);  
97.         redisAsyncFree(_redis_context);  
98.         _redis_context = NULL;  
99.     }  
100.   
101.     return true;  
102. }  
103.   
104. bool CRedisPublisher::publish(const std::string &channel_name,  
105.     const std::string &message)  
106. {  
107.     int ret = redisAsyncCommand(_redis_context,   
108.         &CRedisPublisher::command_callback, this, "PUBLISH %s %s",   
109.         channel_name.c_str(), message.c_str());  
110.     if (REDIS_ERR == ret)  
111.     {  
112.         printf("Publish command failed: %d\n", ret);  
113.         return false;  
114.     }  
115.   
116.     return true;  
117. }  
118.   
119. void CRedisPublisher::connect_callback(const redisAsyncContext *redis_context,  
120.     int status)  
121. {  
122.     if (status != REDIS_OK)  
123.     {  
124.         printf(": Error: %s\n", redis_context->errstr);  
125.     }  
126.     else  
127.     {  
128.         printf(": Redis connected!\n");  
129.     }  
130. }  
131.   
132. void CRedisPublisher::disconnect_callback(  
133.     const redisAsyncContext *redis_context, int status)  
134. {  
135.     if (status != REDIS_OK)  
136.     {  
137.         // 这里异常退出，可以尝试重连  
138.         printf(": Error: %s\n", redis_context->errstr);  
139.     }  
140. }  
141.   
142. // 消息接收回调函数  
143. void CRedisPublisher::command_callback(redisAsyncContext *redis_context,  
144.     void *reply, void *privdata)  
145. {  
146.     printf("command callback.\n");  
147.     // 这里不执行任何操作  
148. }  
149.   
150. void *CRedisPublisher::event_thread(void *data)  
151. {  
152.     if (NULL == data)  
153.     {  
154.         printf(": Error!\n");  
155.         assert(false);  
156.         return NULL;  
157.     }  
158.   
159.     CRedisPublisher *self_this = reinterpret_cast(data);  
160.     return self_this->event_proc();  
161. }  
162.   
163. void *CRedisPublisher::event_proc()  
164. {  
165.     sem_wait(&_event_sem);  
166.       
167.     // 开启事件分发，event_base_dispatch会阻塞  
168.     event_base_dispatch(_event_base);  
169.   
170.     return NULL;  
171. }  

redis_subscriber.h

[cpp] view plain copy

1. /************************************************************************* 
2.     > File Name: redis_subscriber.h 
3.     > Created Time: Sat 23 Apr 2016 10:15:09 PM CST 
4.     > Description: 封装hiredis，实现消息订阅redis功能 
5.  ************************************************************************/  
6.   
7. #ifndef REDIS_SUBSCRIBER_H  
8. #define REDIS_SUBSCRIBER_H  
9.   
10. #include   
11. #include   
12. #include   
13. #include   
14. #include   
15. #include   
16. #include   
17. #include   
18. #include   
19.   
20. class CRedisSubscriber  
21. {  
22. public:  
23.     typedef std::tr1::function \  
24.         NotifyMessageFn;    // 回调函数对象类型，当接收到消息后调用回调把消息发送出去  
25.           
26.     CRedisSubscriber();  
27.     ~CRedisSubscriber();  
28.       
29.     bool init(const NotifyMessageFn &fn);   // 传入回调对象  
30.     bool uninit();  
31.     bool connect();  
32.     bool disconnect();  
33.       
34.     // 可以多次调用，订阅多个频道  
35.     bool subscribe(const std::string &channel_name);  
36.       
37. private:  
38.     // 下面三个回调函数供redis服务调用  
39.     // 连接回调  
40.     static void connect_callback(const redisAsyncContext *redis_context,  
41.         int status);  
42.       
43.     // 断开连接的回调  
44.     static void disconnect_callback(const redisAsyncContext *redis_context,  
45.         int status);  
46.   
47.     // 执行命令回调  
48.     static void command_callback(redisAsyncContext *redis_context,  
49.         void *reply, void *privdata);  
50.   
51.     // 事件分发线程函数  
52.     static void *event_thread(void *data);  
53.     void *event_proc();  
54.       
55. private:  
56.     // libevent事件对象  
57.     event_base *_event_base;  
58.     // 事件线程ID  
59.     pthread_t _event_thread;  
60.     // 事件线程的信号量  
61.     sem_t _event_sem;  
62.     // hiredis异步对象  
63.     redisAsyncContext *_redis_context;  
64.       
65.     // 通知外层的回调函数对象  
66.     NotifyMessageFn _notify_message_fn;  
67. };  
68.   
69. #endif  

redis_subscriber.cpp:

1. /************************************************************************* 
2.     > File Name: redis_subscriber.cpp 
3.     > Created Time: Sat 23 Apr 2016 10:15:09 PM CST 
4.     > Description:  
5.  ************************************************************************/  
6.    
7. #include   
8. #include   
9. #include   
10. #include "redis_subscriber.h"  
11.   
12. CRedisSubscriber::CRedisSubscriber():_event_base(0), _event_thread(0),  
13. _redis_context(0)  
14. {  
15. }  
16.   
17. CRedisSubscriber::~CRedisSubscriber()  
18. {  
19. }  
20.   
21. bool CRedisSubscriber::init(const NotifyMessageFn &fn)  
22. {  
23.     // initialize the event  
24.     _notify_message_fn = fn;  
25.     _event_base = event_base_new();    // 创建libevent对象  
26.     if (NULL == _event_base)  
27.     {  
28.         printf(": Create redis event failed.\n");  
29.         return false;  
30.     }  
31.   
32.     memset(&_event_sem, 0, sizeof(_event_sem));  
33.     int ret = sem_init(&_event_sem, 0, 0);  
34.     if (ret != 0)  
35.     {  
36.         printf(": Init sem failed.\n");  
37.         return false;  
38.     }  
39.   
40.     return true;  
41. }  
42.   
43. bool CRedisSubscriber::uninit()  
44. {  
45.     _event_base = NULL;  
46.       
47.     sem_destroy(&_event_sem);     
48.     return true;  
49. }  
50.   
51. bool CRedisSubscriber::connect()  
52. {  
53.     // connect redis  
54.     _redis_context = redisAsyncConnect("127.0.0.1", 6379);    // 异步连接到redis服务器上，使用默认端口  
55.     if (NULL == _redis_context)  
56.     {  
57.         printf(": Connect redis failed.\n");  
58.         return false;  
59.     }  
60.   
61.     if (_redis_context->err)  
62.     {  
63.         printf(": Connect redis error: %d, %s\n",   
64.             _redis_context->err, _redis_context->errstr);    // 输出错误信息  
65.         return false;  
66.     }  
67.   
68.     // attach the event  
69.     redisLibeventAttach(_redis_context, _event_base);    // 将事件绑定到redis context上，使设置给redis的回调跟事件关联  
70.       
71.     // 创建事件处理线程  
72.     int ret = pthread_create(&_event_thread, 0, &CRedisSubscriber::event_thread, this);  
73.     if (ret != 0)  
74.     {  
75.         printf(": create event thread failed.\n");  
76.         disconnect();  
77.         return false;  
78.     }  
79.   
80.     // 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态  
81.     redisAsyncSetConnectCallback(_redis_context,   
82.         &CRedisSubscriber::connect_callback);  
83.   
84.     // 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连  
85.     redisAsyncSetDisconnectCallback(_redis_context,  
86.         &CRedisSubscriber::disconnect_callback);  
87.   
88.     // 启动事件线程  
89.     sem_post(&_event_sem);  
90.     return true;  
91. }  
92.   
93. bool CRedisSubscriber::disconnect()  
94. {  
95.     if (_redis_context)  
96.     {  
97.         redisAsyncDisconnect(_redis_context);  
98.         redisAsyncFree(_redis_context);  
99.         _redis_context = NULL;  
100.     }  
101.   
102.     return true;  
103. }  
104. 
     
105. bool CRedisSubscriber::subscribe(const std::string &channel_name)  
106. {  
107.     int ret = redisAsyncCommand(_redis_context,   
108.         &CRedisSubscriber::command_callback, this, "SUBSCRIBE %s",   
109.         channel_name.c_str());  
110.     if (REDIS_ERR == ret)  
111.     {  
112.         printf("Subscribe command failed: %d\n", ret);  
113.         return false;  
114.     }  
115.       
116.     printf(": Subscribe success: %s\n", channel_name.c_str());  
117.     return true;  
118. }  
119.   
120. void CRedisSubscriber::connect_callback(const redisAsyncContext *redis_context,  
121.     int status)  
122. {  
123.     if (status != REDIS_OK)  
124.     {  
125.         printf(": Error: %s\n", redis_context->errstr);  
126.     }  
127.     else  
128.     {  
129.         printf(": Redis connected!");  
130.     }  
131. }  
132.   
133. void CRedisSubscriber::disconnect_callback(  
134.     const redisAsyncContext *redis_context, int status)  
135. {  
136.     if (status != REDIS_OK)  
137.     {  
138.         // 这里异常退出，可以尝试重连  
139.         printf(": Error: %s\n", redis_context->errstr);  
140.     }  
141. }  
142.   
143. // 消息接收回调函数  
144. void CRedisSubscriber::command_callback(redisAsyncContext *redis_context,  
145.     void *reply, void *privdata)  
146. {  
147.     if (NULL == reply || NULL == privdata) {  
148.         return ;  
149.     }  
150.   
151.     // 静态函数中，要使用类的成员变量，把当前的this指针传进来，用this指针间接访问  
152.     CRedisSubscriber *self_this = reinterpret_cast(privdata);  
153.     redisReply *redis_reply = reinterpret_cast(reply);  
154.       
155.     // 订阅接收到的消息是一个带三元素的数组  
156.     if (redis_reply->type == REDIS_REPLY_ARRAY &&  
157.     redis_reply->elements == 3)  
158.     {  
159.         printf(": Recieve message:%s:%d:%s:%d:%s:%d\n",  
160.         redis_reply->element[0]->str, redis_reply->element[0]->len,  
161.         redis_reply->element[1]->str, redis_reply->element[1]->len,  
162.         redis_reply->element[2]->str, redis_reply->element[2]->len);  
163.           
164.         // 调用函数对象把消息通知给外层  
165.         self_this->_notify_message_fn(redis_reply->element[1]->str,  
166.             redis_reply->element[2]->str, redis_reply->element[2]->len);  
167.     }  
168. }  
169.   
170. void *CRedisSubscriber::event_thread(void *data)  
171. {  
172.     if (NULL == data)  
173.     {  
174.         printf(": Error!\n");  
175.         assert(false);  
176.         return NULL;  
177.     }  
178.   
179.     CRedisSubscriber *self_this = reinterpret_cast(data);  
180.     return self_this->event_proc();  
181. }  
182.   
183. void *CRedisSubscriber::event_proc()  
184. {  
185.     sem_wait(&_event_sem);  
186.       
187.     // 开启事件分发，event_base_dispatch会阻塞  
188.     event_base_dispatch(_event_base);  
189.   
190.     return NULL;  
191. }  

问题1：hiredis官网没有异步接口的实现例子。

        hiredis提供了几个异步通信的API，一开始根据API名字的理解，我们实现了跟redis服务器建立连接、订阅和发布的功能，可在实际使用的时候，程序并没有像我们预想的那样，除了能够建立连接外，任何事情都没发生。

        网上查了很多资料，原来hiredis的异步实现是通过事件来分发redis发送过来的消息的，hiredis可以使用libae、libev、libuv和libevent中的任何一个实现事件的分发，网上的资料提示使用libae、libev和libuv可能发生其他问题，这里为了方便就选用libevent。hireds官网并没有对libevent做任何介绍，也没用说明使用异步机制需要引入事件的接口，所以一开始走了很多弯路。

        关于libevent的使用这里就不再赘述，详情可以见libevent官网。

libevent官网：http:///

libevent api文档：https://www./~provos/libevent/doxygen-2.0.1/include_2event2_2event_8h.html#6e9827de8c3014417b11b48f2fe688ae

CRedisPublisher和CRedisSubscriber的初始化过程：

初始化事件处理，并获得事件处理的实例：

[cpp] view plain copy

1. _event_base = event_base_new();  

在获得redisAsyncContext *之后，调用

[cpp] view plain copy

1. redisLibeventAttach(_redis_context, _event_base);  

这样就将事件处理和redis关联起来，最后在另一个线程调用

[cpp] view plain copy

1. event_base_dispatch(_event_base);  

启动事件的分发，这是一个阻塞函数，因此，创建了一个新的线程处理事件分发，值得注意的是，这里用信号灯_event_sem控制线程的启动，意在程序调用

[cpp] view plain copy

1. redisAsyncSetConnectCallback(_redis_context,   
2.     &CRedisSubscriber::connect_callback);  
3. redisAsyncSetDisconnectCallback(_redis_context,  
4.     &CRedisSubscriber::disconnect_callback);  

之后，能够完全捕捉到这两个回调。

问题2 奇特的‘ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context’错误

        有些人会觉得这两个类设计有点冗余，我们发现CRedisPublisher和CRedisSubscriber很多逻辑是一样的，为什么不把他们整合到一起成一个类，既能够发布消息也能够订阅消息。其实一开始我就是这么干的，在使用的时候发现，用同个redisAsynContex *对象进行消息订阅和发布，与redis服务连接会自动断开，disconnect_callback回调会被调用，并且返回奇怪的错误：ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context，因此，不能使用同个redisAsyncContext *对象实现发布和订阅。这里为了减少设计的复杂性，就将两个类的逻辑分开了。

        当然，你也可以将相同的逻辑抽象到一个基类里，并实现publish和subscribe接口。

问题3 相关依赖的库

        编译之前，需要安装hiredis、libevent和boost库，我是用的是Ubuntu x64系统。

hiredis官网：https://github.com/redis/hiredis

下载源码解压，进入解压目录，执行make && make install命令。

libevent官网：http:///下载最新的稳定版

解压后进入解压目录，执行命令

./configure -prefix=/usr

sudo make && make install

boost库：直接执行安装：sudo apt-get install libboost-dev

如果你不是用std::tr1::function的函数对象来给外层通知消息，就不需要boost库。你可以用接口的形式实现回调，把接口传给CRedisSubscribe类，让它在接收到消息后调用接口回调，通知外层。

问题4 如何使用

        最后贴出例子代码。

publisher.cpp，实现发布消息：

[cpp] view plain copy

1. /************************************************************************* 
2.     > File Name: publisher.cpp 
3.     > Author: chenzengba 
4.     > Mail: [e-mail]  
5.     > Created Time: Sat 23 Apr 2016 12:13:24 PM CST 
6.  ************************************************************************/  
7.   
8. #include "redis_publisher.h"  
9.   
10. int main(int argc, char *argv[])  
11. {  
12.     CRedisPublisher publisher;  
13.   
14.     bool ret = publisher.init();  
15.     if (!ret)   
16.     {  
17.         printf("Init failed.\n");  
18.         return 0;  
19.     }  
20.   
21.     ret = publisher.connect();  
22.     if (!ret)  
23.     {  
24.         printf("connect failed.");  
25.         return 0;  
26.     }  
27.   
28.     while (true)  
29.     {  
30.         publisher.publish("test-channel", "Test message");  
31.         sleep(1);  
32.     }  
33.   
34.     publisher.disconnect();  
35.     publisher.uninit();  
36.     return 0;  
37. }  

subscriber.cpp实现订阅消息：

[cpp] view plain copy

1. /************************************************************************* 
2.     > File Name: subscriber.cpp 
3.     > Author: chenzengba 
4.     > Mail: [e-mail]  
5.     > Created Time: Sat 23 Apr 2016 12:26:42 PM CST 
6.  ************************************************************************/  
7.   
8. #include "redis_subscriber.h"  
9.   
10. void recieve_message(const char *channel_name,  
11.     const char *message, int len)  
12. {  
13.     printf("Recieve message:\n    channel name: %s\n    message: %s\n",  
14.         channel_name, message);  
15. }  
16.   
17. int main(int argc, char *argv[])  
18. {  
19.     CRedisSubscriber subscriber;  
20.     CRedisSubscriber::NotifyMessageFn fn =   
21.         bind(recieve_message, std::tr1::placeholders::_1,  
22.         std::tr1::placeholders::_2, std::tr1::placeholders::_3);  
23.   
24.     bool ret = subscriber.init(fn);  
25.     if (!ret)  
26.     {  
27.         printf("Init failed.\n");  
28.         return 0;  
29.     }  
30.   
31.     ret = subscriber.connect();  
32.     if (!ret)  
33.     {  
34.         printf("Connect failed.\n");  
35.         return 0;  
36.     }  
37.   
38.     subscriber.subscribe("test-channel");  
39.   
40.     while (true)  
41.     {  
42.         sleep(1);  
43.     }  
44.   
45.     subscriber.disconnect();  
46.     subscriber.uninit();  
47.   
48.     return 0;  
49. }  

关于编译的问题：在g++中编译，注意要加上-lhiredis -levent参数，下面是一个简单的Makefile：

[cpp] view plain copy

1. EXE=server_main client_main  
2. CC=g++  
3. FLAG=-lhiredis -levent  
4. OBJ=redis_publisher.o publisher.o redis_subscriber.o subscriber.o  
5.   
6. all:$(EXE)  
7.   
8. $(EXE):$(OBJ)  
9.     $(CC) -o publisher redis_publisher.o publisher.o $(FLAG)  
10.     $(CC) -o subscriber redis_subscriber.o subscriber.o $(FLAG)  
11.   
12. redis_publisher.o:redis_publisher.h  
13. redis_subscriber.o:redis_subscriber.h  
14.   
15. publisher.o:publisher.cpp  
16.     $(CC) -c publisher.cpp  
17.   
18. subscriber.o:subscriber.cpp  
19.     $(CC) -c subscriber.cpp  
20. clean:  
21.     rm publisher subscriber *.o