搜索
首页数据库mysql教程基于Hiredis异步API的聊天系统实现

基于Hiredis异步API的聊天系统实现

WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB
WBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWBOYWB
Jun 07, 2016 pm 02:51 PM
api基于实现异步系统聊天

基于Hiredis异步API的聊天系统实现 上一篇文章http://blog.csdn.net/qq_34788352/article/details/51313027使用Hiredis的同步API实现了发送消息的客户端，当我使用同步API实现订阅频道客户端时，一旦订阅频道，就会出现无法操作的情况，这是就是同步和异步的

基于Hiredis异步API的聊天系统实现

上一篇文章http://blog.csdn.net/qq_34788352/article/details/51313027使用Hiredis的同步API实现了发送消息的客户端，当我使用同步API实现订阅频道客户端时，一旦订阅频道，就会出现无法操作的情况，这是就是同步和异步的问题。使用同步API，订阅频道后，客户端会进入阻塞状态，等待订阅频道发布的消息，不能实现既订阅频道，又能发布消息的功能。为了实现一个客户端既能订阅频道，又能发布消息的功能，就需要使用Hiredis的异步API。

首先特别感谢黄天霸、tickTick、vah101，当我遇到各种各样奇怪问题的时候，你们的帖子给与了我解答，谢谢。

开始正题，hiredis异步的实现主要是依靠redis自带的ae事件库或者libev事件库或者libevent的事件库或者libuv事件库。网上一些人是通过libevent事件库来实现，本系统则使用Redis自带ae事件库来实现，Redis不用libevent事件库而选择重写一个ae事件库，必定有其道理。

首先介绍使用到的异步API，位于async.h中： 

<code class=" hljs perl">redisAsyncContext <span class="hljs-variable">*redisAsyncConnect</span>(const char <span class="hljs-variable">*ip</span>, <span class="hljs-keyword">int</span> port); <span class="hljs-regexp">//</span>用于建立异步连接

<span class="hljs-keyword">int</span> redisAsyncSetConnectCallback(redisAsyncContext <span class="hljs-variable">*ac</span>, redisConnectCallback <span class="hljs-variable">*fn</span>);  <span class="hljs-regexp">//</span>设置连接回调函数，回调函数形式：void callback(const redisAsyncContext <span class="hljs-variable">*c</span>, <span class="hljs-keyword">int</span> status);

<span class="hljs-keyword">int</span> redisAsyncSetDisconnectCallback(redisAsyncContext <span class="hljs-variable">*ac</span>, redisDisconnectCallback <span class="hljs-variable">*fn</span>);  <span class="hljs-regexp">//</span>设置断开连接回调函数,回调函数形式：void callback(const redisAsyncContext <span class="hljs-variable">*c</span>, <span class="hljs-keyword">int</span> status);

void redisAsyncDisconnect(redisAsyncContext <span class="hljs-variable">*ac</span>); <span class="hljs-regexp">//</span>断开异步连接

void redisAsyncFree(redisAsyncContext <span class="hljs-variable">*ac</span>); <span class="hljs-regexp">//</span>释放建立连接时，创建的redisAsyncContext结构

<span class="hljs-keyword">int</span> redisAsyncCommand(redisAsyncContext <span class="hljs-variable">*ac</span>, redisCallbackFn <span class="hljs-variable">*fn</span>, void <span class="hljs-variable">*privdata</span>, const char <span class="hljs-variable">*format</span>, ...); <span class="hljs-regexp">//</span>发送Redis命令，需要实现一个回调函数来出来命令的返回，fn是回调函数的地址,回调函数形式：void callback(redisAsyncContext <span class="hljs-variable">*c</span>, void <span class="hljs-variable">*reply</span>, void <span class="hljs-variable">*pridata</span>);
</code>

有了上面的异步API，就可以开始客户端的撰写。
首先封装订阅端

<code class=" hljs cpp"><span class="hljs-comment">//sub_client.h</span>

<span class="hljs-preprocessor">#ifndef REDIS_SUB_CLIENT_H  </span>
<span class="hljs-preprocessor">#define REDIS_SUB_CLIENT_H  </span>

<span class="hljs-keyword">extern</span> <span class="hljs-string">"C"</span>
{  
<span class="hljs-preprocessor">#include <stdlib.h>  </span>
<span class="hljs-preprocessor">#include <async.h>  </span>
<span class="hljs-preprocessor">#include <adapters/ae.h> </span>
<span class="hljs-preprocessor">#include <unistd.h>  </span>
<span class="hljs-preprocessor">#include <pthread.h>  </span>
<span class="hljs-preprocessor">#include <semaphore.h> </span>
} 

<span class="hljs-preprocessor">#include <string>  </span>
<span class="hljs-preprocessor">#include <vector>  </span>

<span class="hljs-keyword">using</span> <span class="hljs-keyword">namespace</span> <span class="hljs-built_in">std</span>;

<span class="hljs-keyword">class</span> CRedisSubClient  
{  
<span class="hljs-keyword">public</span>:   

    CRedisSubClient();  
    ~CRedisSubClient();  

    <span class="hljs-keyword">bool</span> init();   <span class="hljs-comment">//初始化，事件对象，信号量  </span>
    <span class="hljs-keyword">bool</span> uninit();  <span class="hljs-comment">//释放对象</span>
    <span class="hljs-keyword">bool</span> connect();  <span class="hljs-comment">//连接服务器</span>
    <span class="hljs-keyword">bool</span> disconnect(); <span class="hljs-comment">//断开服务器</span>

    <span class="hljs-comment">//订阅频道 </span>
    <span class="hljs-keyword">bool</span> subscribe(<span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &channel_name);  

<span class="hljs-keyword">private</span>:  
    <span class="hljs-comment">// 下面三个回调函数供redis服务调用  </span>
    <span class="hljs-comment">// 连接回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> connect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
        <span class="hljs-keyword">int</span> status);  

    <span class="hljs-comment">// 断开连接的回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> disconnect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
        <span class="hljs-keyword">int</span> status);  

    <span class="hljs-comment">// 执行命令回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> command_callback(redisAsyncContext *redis_context,  
        <span class="hljs-keyword">void</span> *reply, <span class="hljs-keyword">void</span> *privdata);  

    <span class="hljs-comment">// 事件分发线程函数  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> *event_thread(<span class="hljs-keyword">void</span> *data);  
    <span class="hljs-keyword">void</span> *event_proc();  

<span class="hljs-keyword">private</span>:  
    <span class="hljs-comment">// ae事件对象  </span>
    aeEventLoop *loop;  
    <span class="hljs-comment">// 事件线程ID  </span>
    pthread_t _event_thread;  
    <span class="hljs-comment">// 事件线程的信号量  </span>
    sem_t _event_sem;  
    <span class="hljs-comment">// hiredis异步对象  </span>
    redisAsyncContext *_redis_context;  

};  

<span class="hljs-preprocessor">#endif  </span>
</code>

使用extern “C”是因为redis和hiredis都是c写的，当使用c++链接c代码生成的库中的函数时，会出现undefined reference的问题。

订阅端的实现

<code class=" hljs cpp"><span class="hljs-comment">//sub_client.cpp</span>

<span class="hljs-preprocessor">#include <stddef.h>  </span>
<span class="hljs-preprocessor">#include <assert.h>  </span>
<span class="hljs-preprocessor">#include <string.h>  </span>
<span class="hljs-preprocessor">#include "sub_client.h" </span>

<span class="hljs-keyword">using</span> <span class="hljs-keyword">namespace</span> <span class="hljs-built_in">std</span>; 

CRedisSubClient::CRedisSubClient():loop(<span class="hljs-number">0</span>), _event_thread(<span class="hljs-number">0</span>),  
_redis_context(<span class="hljs-number">0</span>)  
{  
}  

CRedisSubClient::~CRedisSubClient()  
{  
}  

<span class="hljs-keyword">bool</span> CRedisSubClient::init()  
{  

    loop = aeCreateEventLoop(<span class="hljs-number">64</span>);    <span class="hljs-comment">// 创建ae对象</span>

    <span class="hljs-keyword">if</span> (NULL == loop)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Create redis event failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-built_in">memset</span>(&_event_sem, <span class="hljs-number">0</span>, <span class="hljs-keyword">sizeof</span>(_event_sem)); 

    <span class="hljs-keyword">int</span> ret = sem_init(&_event_sem, <span class="hljs-number">0</span>, <span class="hljs-number">0</span>);  <span class="hljs-comment">//初始化线程信号量</span>

    <span class="hljs-keyword">if</span> (ret != <span class="hljs-number">0</span>)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Init sem failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisSubClient::uninit()  
{  
    loop = NULL;  

    sem_destroy(&_event_sem);  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisSubClient::connect()  
{    
    _redis_context = redisAsyncConnect(<span class="hljs-string">"127.0.0.1"</span>, <span class="hljs-number">6379</span>);    <span class="hljs-comment">// 异步连接到redis服务器上，使用6380端口</span>

    <span class="hljs-keyword">if</span> (NULL == _redis_context)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Connect redis failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-keyword">if</span> (_redis_context->err)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Connect redis error: %d, %s\n"</span>,   
            _redis_context->err, _redis_context->errstr);    <span class="hljs-comment">// 输出错误信息  </span>
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    redisAeAttach(loop,_redis_context);   <span class="hljs-comment">// 将事件绑定到redis context上，使redis的回调跟事件关联  </span>

    <span class="hljs-comment">// 创建事件处理线程  </span>
    <span class="hljs-keyword">int</span> ret = pthread_create(&_event_thread, <span class="hljs-number">0</span>, &CRedisSubscriber::event_thread, <span class="hljs-keyword">this</span>);  

    <span class="hljs-keyword">if</span> (ret != <span class="hljs-number">0</span>)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Create event thread failed.\n"</span>);  
        disconnect();  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-comment">// 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态  </span>
    redisAsyncSetConnectCallback(_redis_context,   
        &CRedisSubClient::connect_callback);  

    <span class="hljs-comment">// 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连  </span>
    redisAsyncSetDisconnectCallback(_redis_context,  
        &CRedisSubClient::disconnect_callback);  

    <span class="hljs-comment">// 启动事件线程  </span>
    sem_post(&_event_sem);  
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisSubClient::disconnect()  
{  
    <span class="hljs-keyword">if</span> (_redis_context)  
    {  
        redisAsyncDisconnect(_redis_context);  
        redisAsyncFree(_redis_context);  
        _redis_context = NULL;  
    }  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisSubClient::subscribe(<span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &channel_name)  
{  
    <span class="hljs-keyword">int</span> ret = redisAsyncCommand(_redis_context,   
        &CRedisSubscriber::command_callback, <span class="hljs-keyword">this</span>, <span class="hljs-string">"SUBSCRIBE %s"</span>,   
        channel_name.c_str()); <span class="hljs-comment">//订阅一个频道</span>

    <span class="hljs-keyword">if</span> (REDIS_ERR == ret)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Subscribe command failed: %d\n"</span>, ret);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Subscribe success: %s\n"</span>, channel_name.c_str());  
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">void</span> CRedisSubClient::connect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
    <span class="hljs-keyword">int</span> status)  
{  
    <span class="hljs-keyword">if</span> (status != REDIS_OK)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Error: %s\n"</span>, redis_context->errstr);  
    }  
    <span class="hljs-keyword">else</span>  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Redis connected!"</span>);  
    }  
}  

<span class="hljs-keyword">void</span> CRedisSubClient::disconnect_callback(  
    <span class="hljs-keyword">const</span> redisAsyncContext *redis_context, <span class="hljs-keyword">int</span> status)  
{  
    <span class="hljs-keyword">if</span> (status != REDIS_OK)  
    {    
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Error: %s\n"</span>, redis_context->errstr);  
    }  
}  

<span class="hljs-comment">// 消息接收回调函数  </span>
<span class="hljs-keyword">void</span> CRedisSubClient::command_callback(redisAsyncContext *redis_context,  
    <span class="hljs-keyword">void</span> *reply, <span class="hljs-keyword">void</span> *privdata)  
{  
    <span class="hljs-keyword">if</span> (NULL == reply || NULL == privdata) {  
        <span class="hljs-keyword">return</span> ;  
    }  

    redisReply *redis_reply = <span class="hljs-keyword">reinterpret_cast</span><redisReply *>(reply);  

    <span class="hljs-comment">// 订阅接收到的消息是一个带三元素的数组  </span>
    <span class="hljs-keyword">if</span> (redis_reply->type == REDIS_REPLY_ARRAY &&  
    redis_reply->elements == <span class="hljs-number">3</span>)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Recieve message:%s %s %s\n"</span>,  
        redis_reply->element[<span class="hljs-number">0</span>]->str  
        redis_reply->element[<span class="hljs-number">1</span>]->str  
        redis_reply->element[<span class="hljs-number">2</span>]->str);    
    }  
}  

<span class="hljs-keyword">void</span> *CRedisSubClient::event_thread(<span class="hljs-keyword">void</span> *data)  
{  
    <span class="hljs-keyword">if</span> (NULL == data)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Error!\n"</span>);  
        assert(<span class="hljs-keyword">false</span>);  
        <span class="hljs-keyword">return</span> NULL;  
    }  

    CRedisSubClient *self_this = <span class="hljs-keyword">reinterpret_cast</span><CRedisSubClient *>(data);  
    <span class="hljs-keyword">return</span> self_this->event_proc();  
}  

<span class="hljs-keyword">void</span> *CRedisSubClient::event_proc()  
{  
    sem_wait(&_event_sem);  

    <span class="hljs-comment">//进行事件处理循环  </span>
    aeMain(loop);  

    <span class="hljs-keyword">return</span> NULL;  
}  </code>

发布端的封装：

<code class=" hljs cpp"><span class="hljs-comment">//pub_client.h</span>

<span class="hljs-preprocessor">#ifndef REDIS_PUB_CLIENT_H  </span>
<span class="hljs-preprocessor">#define REDIS_PUB_CLIENT_H  </span>

<span class="hljs-keyword">extern</span> <span class="hljs-string">"C"</span>
{  
<span class="hljs-preprocessor">#include <stdlib.h>  </span>
<span class="hljs-preprocessor">#include <async.h>  </span>
<span class="hljs-preprocessor">#include <adapters/ae.h> </span>
<span class="hljs-preprocessor">#include <unistd.h>  </span>
<span class="hljs-preprocessor">#include <pthread.h>  </span>
<span class="hljs-preprocessor">#include <semaphore.h> </span>
} 

<span class="hljs-preprocessor">#include <string>  </span>
<span class="hljs-preprocessor">#include <vector>  </span>

<span class="hljs-keyword">using</span> <span class="hljs-keyword">namespace</span> <span class="hljs-built_in">std</span>;

<span class="hljs-keyword">class</span> CRedisPubClient  
{  
<span class="hljs-keyword">public</span>:   

    CRedisPubClient();  
    ~CRedisPubClient();  

    <span class="hljs-keyword">bool</span> init();   <span class="hljs-comment">//初始化，事件对象，信号量  </span>
    <span class="hljs-keyword">bool</span> uninit();  <span class="hljs-comment">//释放对象</span>
    <span class="hljs-keyword">bool</span> connect();  <span class="hljs-comment">//连接服务器</span>
    <span class="hljs-keyword">bool</span> disconnect();  <span class="hljs-comment">//断开服务器</span>

    <span class="hljs-comment">//订阅频道 </span>
    <span class="hljs-keyword">bool</span> publish(<span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &channel_name, <span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &message);  

<span class="hljs-keyword">private</span>:  
    <span class="hljs-comment">// 下面三个回调函数供redis服务调用  </span>
    <span class="hljs-comment">// 连接回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> connect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
        <span class="hljs-keyword">int</span> status);  

    <span class="hljs-comment">// 断开连接的回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> disconnect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
        <span class="hljs-keyword">int</span> status);  

    <span class="hljs-comment">// 执行命令回调  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> command_callback(redisAsyncContext *redis_context,  
        <span class="hljs-keyword">void</span> *reply, <span class="hljs-keyword">void</span> *privdata);  

    <span class="hljs-comment">// 事件分发线程函数  </span>
    <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> *event_thread(<span class="hljs-keyword">void</span> *data);  
    <span class="hljs-keyword">void</span> *event_proc();  

<span class="hljs-keyword">private</span>:  
    <span class="hljs-comment">// ae事件对象  </span>
    aeEventLoop *loop;  
    <span class="hljs-comment">// 事件线程ID  </span>
    pthread_t _event_thread;  
    <span class="hljs-comment">// 事件线程的信号量  </span>
    sem_t _event_sem;  
    <span class="hljs-comment">// hiredis异步对象  </span>
    redisAsyncContext *_redis_context;  

};  

<span class="hljs-preprocessor">#endif  </span></code>

发布端的实现：

<code class=" hljs cpp"><span class="hljs-comment">//pub_client.cpp</span>

<span class="hljs-preprocessor">#include <stddef.h>  </span>
<span class="hljs-preprocessor">#include <assert.h>  </span>
<span class="hljs-preprocessor">#include <string.h>  </span>
<span class="hljs-preprocessor">#include "pub_client.h" </span>

<span class="hljs-keyword">using</span> <span class="hljs-keyword">namespace</span> <span class="hljs-built_in">std</span>; 

CRedisPubClient::CRedisPubClient():loop(<span class="hljs-number">0</span>), _event_thread(<span class="hljs-number">0</span>),  
_redis_context(<span class="hljs-number">0</span>)  
{  
}  

CRedisPubClient::~CRedisPubClient()  
{  
}  

<span class="hljs-keyword">bool</span> CRedisPubClient::init()  
{  

    loop = aeCreateEventLoop(<span class="hljs-number">64</span>);    <span class="hljs-comment">// 创建ae对象</span>

    <span class="hljs-keyword">if</span> (NULL == loop)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Create redis event failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-built_in">memset</span>(&_event_sem, <span class="hljs-number">0</span>, <span class="hljs-keyword">sizeof</span>(_event_sem)); 

    <span class="hljs-keyword">int</span> ret = sem_init(&_event_sem, <span class="hljs-number">0</span>, <span class="hljs-number">0</span>);  <span class="hljs-comment">//初始化线程信号量</span>

    <span class="hljs-keyword">if</span> (ret != <span class="hljs-number">0</span>)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Init sem failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisPubClient::uninit()  
{  
    loop = NULL;  

    sem_destroy(&_event_sem);  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisPubClient::connect()  
{    
    _redis_context = redisAsyncConnect(<span class="hljs-string">"127.0.0.1"</span>, <span class="hljs-number">6379</span>);    <span class="hljs-comment">// 异步连接到redis服务器上，使用6380端口</span>

    <span class="hljs-keyword">if</span> (NULL == _redis_context)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Connect redis failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-keyword">if</span> (_redis_context->err)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Connect redis error: %d, %s\n"</span>,   
            _redis_context->err, _redis_context->errstr);    <span class="hljs-comment">// 输出错误信息  </span>
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    redisAeAttach(loop,_redis_context);   <span class="hljs-comment">// 将事件绑定到redis context上，使redis的回调跟事件关联  </span>

    <span class="hljs-comment">// 创建事件处理线程  </span>
    <span class="hljs-keyword">int</span> ret = pthread_create(&_event_thread, <span class="hljs-number">0</span>, &CRedisSubscriber::event_thread, <span class="hljs-keyword">this</span>);  

    <span class="hljs-keyword">if</span> (ret != <span class="hljs-number">0</span>)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Create event thread failed.\n"</span>);  
        disconnect();  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-comment">// 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态  </span>
    redisAsyncSetConnectCallback(_redis_context,   
        &CRedisSubClient::connect_callback);  

    <span class="hljs-comment">// 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连  </span>
    redisAsyncSetDisconnectCallback(_redis_context,  
        &CRedisSubClient::disconnect_callback);  

    <span class="hljs-comment">// 启动事件线程  </span>
    sem_post(&_event_sem);  
    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisPubClient::disconnect()  
{  
    <span class="hljs-keyword">if</span> (_redis_context)  
    {  
        redisAsyncDisconnect(_redis_context);  
        redisAsyncFree(_redis_context);  
        _redis_context = NULL;  
    }  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;  
}  

<span class="hljs-keyword">bool</span> CRedisPubClient::publish(<span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &channel_name, <span class="hljs-keyword">const</span> <span class="hljs-built_in">string</span> &message)  
{  
    <span class="hljs-keyword">int</span> ret = redisAsyncCommand(_redis_context,   
        &CRedisPublisher::command_callback, <span class="hljs-keyword">this</span>, <span class="hljs-string">"PUBLISH %s %s"</span>,   
        channel_name.c_str(), message.c_str());  <span class="hljs-comment">//发布消息  </span>

    <span class="hljs-keyword">if</span> (REDIS_ERR == ret)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Publish command failed: %d\n"</span>, ret);  
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;  
    }  

    <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;    
}  

<span class="hljs-keyword">void</span> CRedisPubClient::connect_callback(<span class="hljs-keyword">const</span> redisAsyncContext *redis_context,  
    <span class="hljs-keyword">int</span> status)  
{  
    <span class="hljs-keyword">if</span> (status != REDIS_OK)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Error: %s\n"</span>, redis_context->errstr);  
    }  
    <span class="hljs-keyword">else</span>  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Redis connected!"</span>);  
    }  
}  

<span class="hljs-keyword">void</span> CRedisPubClient::disconnect_callback(  
    <span class="hljs-keyword">const</span> redisAsyncContext *redis_context, <span class="hljs-keyword">int</span> status)  
{  
    <span class="hljs-keyword">if</span> (status != REDIS_OK)  
    {    
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Error: %s\n"</span>, redis_context->errstr);  
    }  
}  

<span class="hljs-comment">// 消息接收回调函数  </span>
<span class="hljs-keyword">void</span> CRedisPubClient::command_callback(redisAsyncContext *redis_context,  
    <span class="hljs-keyword">void</span> *reply, <span class="hljs-keyword">void</span> *privdata)  
{  
    <span class="hljs-keyword">if</span> (NULL == reply || NULL == privdata) 
    {  
        <span class="hljs-keyword">return</span> ;  
    }  

    redisReply *redis_reply = <span class="hljs-keyword">reinterpret_cast</span><redisReply *>(reply); 

    <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Publish: %s"</span>,redis_reply.str);
}  

<span class="hljs-keyword">void</span> *CRedisPubClient::event_thread(<span class="hljs-keyword">void</span> *data)  
{  
    <span class="hljs-keyword">if</span> (NULL == data)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">": Error!\n"</span>);  
        assert(<span class="hljs-keyword">false</span>);  
        <span class="hljs-keyword">return</span> NULL;  
    }  

    CRedisPubClient *self_this = <span class="hljs-keyword">reinterpret_cast</span><CRedisPubClient *>(data);  
    <span class="hljs-keyword">return</span> self_this->event_proc();  
}  

<span class="hljs-keyword">void</span> *CRedisPubClient::event_proc()  
{  
    sem_wait(&_event_sem);  

    <span class="hljs-comment">//进行事件处理循环  </span>
    aeMain(loop);  

    <span class="hljs-keyword">return</span> NULL;  
}  </code>

测试封装的sub_client和pub_client类:

<code class=" hljs cpp"><span class="hljs-comment">//test_subpub.cpp</span>

<span class="hljs-preprocessor">#include "pub_client.h"</span>
<span class="hljs-preprocessor">#include "sub_client.h"  </span>

<span class="hljs-keyword">int</span> main(<span class="hljs-keyword">int</span> argc, <span class="hljs-keyword">char</span> *argv[])  
{  
    CRedisPubClient publisher;  
    CRedisSubClient subcriber;

    <span class="hljs-keyword">bool</span> ret_pub = publisher.init();  
    <span class="hljs-keyword">bool</span> ret_sub = subcriber.init();

    <span class="hljs-keyword">if</span> (!ret_sub&&!ret_pub)   
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"Init failed.\n"</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-number">0</span>;  
    }  

    ret_pub = publisher.connect(); 
    ret_sub = subcriber.connect(); 

    <span class="hljs-keyword">if</span> (!ret_sub&&!ret_pub)  
    {  
        <span class="hljs-built_in">printf</span>(<span class="hljs-string">"connect failed."</span>);  
        <span class="hljs-keyword">return</span> <span class="hljs-number">0</span>;  
    }  

    subscriber.subcribe(<span class="hljs-string">"sports"</span>);

    <span class="hljs-keyword">while</span> (<span class="hljs-keyword">true</span>)  
    {  
        publisher.publish(<span class="hljs-string">"sports"</span>, <span class="hljs-string">"ball"</span>);  
        sleep(<span class="hljs-number">1</span>);  
    }  

    publisher.disconnect();  
    publisher.uninit();  
    subscriber.disconnect();
    subscriber.disconnect();
    <span class="hljs-keyword">return</span> <span class="hljs-number">0</span>;  

} </code>

终于到了编译链接运行的阶段，我在这个地方卡了快一天，期间各种编译、链接的错误。直接编译链接会出现找到不aeCreateFileEvent，aeDeleteFileEvent，aeMain等等错误。

  1. 先将redis/src文件夹中的ae.c，ae.h，ae_epoll.c，config.h，zmalloc.c，zmalloc.h拷贝至hiredis目录下
  2. 用gcc -c ae.c gcc -c zmalloc.c生成ae.o和zmalloc.o，利用这两个文件生成静态库 ar -r libar.a ae.o zmalloc.o
  3. 然后编译g++ -o test_subpub test_subpub.cpp pub_client.cpp sub_client.cpp -pthread -I ../ -I./ -I ../adapters ../libhiredis.a ../libae.a
  4. 运行./test_subpub

可以看到，客户端中消息在不断地滚动，即同时实现了订阅频道和发送消息的功能。

声明
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn
相关文章
MySQL：初学者的基本技能MySQL：初学者的基本技能Apr 18, 2025 am 12:24 AM

MySQL适合初学者学习数据库技能。1.安装MySQL服务器和客户端工具。2.理解基本SQL查询，如SELECT。3.掌握数据操作：创建表、插入、更新、删除数据。4.学习高级技巧：子查询和窗口函数。5.调试和优化：检查语法、使用索引、避免SELECT*，并使用LIMIT。

MySQL：结构化数据和关系数据库MySQL：结构化数据和关系数据库Apr 18, 2025 am 12:22 AM

MySQL通过表结构和SQL查询高效管理结构化数据，并通过外键实现表间关系。1.创建表时定义数据格式和类型。2.使用外键建立表间关系。3.通过索引和查询优化提高性能。4.定期备份和监控数据库确保数据安全和性能优化。

MySQL：解释的关键功能和功能MySQL：解释的关键功能和功能Apr 18, 2025 am 12:17 AM

MySQL是一个开源的关系型数据库管理系统，广泛应用于Web开发。它的关键特性包括：1.支持多种存储引擎，如InnoDB和MyISAM，适用于不同场景；2.提供主从复制功能，利于负载均衡和数据备份；3.通过查询优化和索引使用提高查询效率。

SQL的目的：与MySQL数据库进行交互SQL的目的：与MySQL数据库进行交互Apr 18, 2025 am 12:12 AM

SQL用于与MySQL数据库交互，实现数据的增、删、改、查及数据库设计。1）SQL通过SELECT、INSERT、UPDATE、DELETE语句进行数据操作；2）使用CREATE、ALTER、DROP语句进行数据库设计和管理；3）复杂查询和数据分析通过SQL实现，提升业务决策效率。

初学者的MySQL：开始数据库管理初学者的MySQL：开始数据库管理Apr 18, 2025 am 12:10 AM

MySQL的基本操作包括创建数据库、表格，及使用SQL进行数据的CRUD操作。1.创建数据库：CREATEDATABASEmy_first_db;2.创建表格：CREATETABLEbooks(idINTAUTO_INCREMENTPRIMARYKEY,titleVARCHAR(100)NOTNULL,authorVARCHAR(100)NOTNULL,published_yearINT);3.插入数据：INSERTINTObooks(title,author,published_year)VA

MySQL的角色：Web应用程序中的数据库MySQL的角色：Web应用程序中的数据库Apr 17, 2025 am 12:23 AM

MySQL在Web应用中的主要作用是存储和管理数据。1.MySQL高效处理用户信息、产品目录和交易记录等数据。2.通过SQL查询，开发者能从数据库提取信息生成动态内容。3.MySQL基于客户端-服务器模型工作，确保查询速度可接受。

mysql：构建您的第一个数据库mysql：构建您的第一个数据库Apr 17, 2025 am 12:22 AM

构建MySQL数据库的步骤包括：1.创建数据库和表，2.插入数据，3.进行查询。首先，使用CREATEDATABASE和CREATETABLE语句创建数据库和表，然后用INSERTINTO语句插入数据，最后用SELECT语句查询数据。

MySQL：一种对数据存储的初学者友好方法MySQL：一种对数据存储的初学者友好方法Apr 17, 2025 am 12:21 AM

MySQL适合初学者，因为它易用且功能强大。1.MySQL是关系型数据库，使用SQL进行CRUD操作。2.安装简单，需配置root用户密码。3.使用INSERT、UPDATE、DELETE、SELECT进行数据操作。4.复杂查询可使用ORDERBY、WHERE和JOIN。5.调试需检查语法，使用EXPLAIN分析查询。6.优化建议包括使用索引、选择合适数据类型和良好编程习惯。

See all articles

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序，用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

AI Hentai Generator

AI Hentai Generator

免费生成ai无尽的。

显示更多

热门文章

R.E.P.O.能量晶体解释及其做什么（黄色晶体）
1 个月前By尊渡假赌尊渡假赌尊渡假赌
R.E.P.O.最佳图形设置
1 个月前By尊渡假赌尊渡假赌尊渡假赌
刺客信条阴影：贝壳谜语解决方案
3 周前ByDDD
Windows 11 KB5054979中的新功能以及如何解决更新问题
2 周前ByDDD
威尔R.E.P.O.有交叉游戏吗？
1 个月前By尊渡假赌尊渡假赌尊渡假赌
显示更多

热工具

Dreamweaver Mac版

Dreamweaver Mac版

视觉化网页开发工具

PhpStorm Mac 版本

PhpStorm Mac 版本

最新（2018.2.1 ）专业的PHP集成开发工具

螳螂BT

螳螂BT

Mantis是一个易于部署的基于Web的缺陷跟踪工具，用于帮助产品缺陷跟踪。它需要PHP、MySQL和一个Web服务器。请查看我们的演示和托管服务。

适用于 Eclipse 的 SAP NetWeaver 服务器适配器

适用于 Eclipse 的 SAP NetWeaver 服务器适配器

将Eclipse与SAP NetWeaver应用服务器集成。

WebStorm Mac版

WebStorm Mac版

好用的JavaScript开发工具

显示更多

热门话题

gmail邮箱登陆入口在哪里
7552
15
CakePHP 教程
1382
52
steam的账户名称是什么格式
83
11
win11激活密钥永久
59
19
NYT连接提示和答案
22
95
显示更多