运用
http://blog.csdn.net/xumaojun/article/details/51558237 中的redis_publisher.hredis_publisher.cpp
redis_subscriber.h redis_subscriber.cpp四个文件,做一个操作类进行测试.
头文件 Policy.h
#pragma once
#include "redis_publisher.h"
#include "redis_subscriber.h"
#include <iostream> using namespace std; class Policy
{
public:
Policy();
~Policy();
void publish();
static void recieve_message(const char *channel_name,const char *message, int len); CRedisPublisher publisher;
CRedisSubscriber subscriber;
};
代码文件 Policy.cpp
#include "Policy.h" Policy::Policy()
{
bool ret = publisher.init();
if (!ret) {
printf("Init failed.\n"); } ret = publisher.connect();
if (!ret) {
printf("connect failed."); } //***********************************
CRedisSubscriber::NotifyMessageFn fn =
bind(recieve_message, std::tr1::placeholders::_1,
std::tr1::placeholders::_2, std::tr1::placeholders::_3); bool ret1 = subscriber.init(fn);
if (!ret1) {
printf("Init failed.\n");
} ret1 = subscriber.connect();
if (!ret1) {
printf("Connect failed.\n");
}
subscriber.subscribe("test-channel");
} Policy::~Policy()
{
publisher.disconnect();
publisher.uninit(); subscriber.disconnect();
subscriber.uninit();
} void Policy::publish()
{
cout<<"Policy::publish()...\n"<<endl;
publisher.publish("test-channel", "Test message");
} void Policy::recieve_message(const char *channel_name,
const char *message, int len)
{
printf("Recieve message: channel name: %s message: %s\n",
channel_name, message);
}
测试文件 testPolicy.cpp
#include "Policy.h"
int main(int argc, char *argv[])
{
Policy m_policy;
int i=0;
while(i<8){
m_policy.publish();
sleep(2);
cout<<"main sleep...\n";
i++;
}
return 0;
}
Makefile
EXE=server_main client_main
CC=g++
FLAG=-lhiredis -levent -pthread
OBJ=redis_publisher.o redis_subscriber.o Policy.o testPolicy.o all:$(EXE) $(EXE):$(OBJ)
$(CC) -o testPolicy redis_publisher.o redis_subscriber.o Policy.o testPolicy.o $(FLAG) redis_publisher.o:redis_publisher.h
redis_subscriber.o:redis_subscriber.h Policy.o:Policy.h
这样编译之后,可以自发自收.
本例的用途在于可以简单地在项目中实现异构的数据收发.
附上原作代码:
redis_publisher.h
/*************************************************************************
> File Name: redis_publisher.h
> Author: chenzengba
> Mail: chenzengba@gmail.com
> Created Time: Sat 23 Apr 2016 10:15:09 PM CST
> Description: 封装hiredis，实现消息发布给redis功能
************************************************************************/ #ifndef REDIS_PUBLISHER_H
#define REDIS_PUBLISHER_H #include <stdlib.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>
#include <string>
#include <vector>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <boost/tr1/functional.hpp> class CRedisPublisher
{
public:
CRedisPublisher();
~CRedisPublisher(); bool init();
bool uninit();
bool connect();
bool disconnect(); bool publish(const std::string &channel_name,
const std::string &message); private:
// 下面三个回调函数供redis服务调用
// 连接回调
static void connect_callback(const redisAsyncContext *redis_context,
int status); // 断开连接的回调
static void disconnect_callback(const redisAsyncContext *redis_context,
int status); // 执行命令回调
static void command_callback(redisAsyncContext *redis_context,
void *reply, void *privdata); // 事件分发线程函数
static void *event_thread(void *data);
void *event_proc(); private:
// libevent事件对象
event_base *_event_base;
// 事件线程ID
pthread_t _event_thread;
// 事件线程的信号量
sem_t _event_sem;
// hiredis异步对象
redisAsyncContext *_redis_context;
};
redis_publisher.cpp
/*************************************************************************
> File Name: redis_publisher.cpp
> Author: chenzengba
> Mail: chenzengba@gmail.com
> Created Time: Sat 23 Apr 2016 10:15:09 PM CST
> Description:
************************************************************************/ #include <stddef.h>
#include <assert.h>
#include <string.h>
#include "redis_publisher.h" CRedisPublisher::CRedisPublisher():_event_base(0), _event_thread(0),
_redis_context(0)
{
} CRedisPublisher::~CRedisPublisher()
{
} bool CRedisPublisher::init()
{
// initialize the 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 false;
} return true;
} bool CRedisPublisher::uninit()
{
_event_base = NULL; sem_destroy(&_event_sem);
return true;
} bool CRedisPublisher::connect()
{
// connect redis
_redis_context = redisAsyncConnect("127.0.0.1", 6379); // 异步连接到redis服务器上，使用默认端口
if (NULL == _redis_context)
{
printf(": Connect redis failed.\n");
return false;
} if (_redis_context->err)
{
printf(": Connect redis error: %d, %s\n",
_redis_context->err, _redis_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);
if (ret != 0)
{
printf(": create event thread failed.\n");
disconnect();
return false;
} // 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态
redisAsyncSetConnectCallback(_redis_context,
&CRedisPublisher::connect_callback); // 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连
redisAsyncSetDisconnectCallback(_redis_context,
&CRedisPublisher::disconnect_callback); // 启动事件线程
sem_post(&_event_sem);
return true;
} bool CRedisPublisher::disconnect()
{
if (_redis_context)
{
redisAsyncDisconnect(_redis_context);
redisAsyncFree(_redis_context);
_redis_context = NULL;
} return true;
} bool CRedisPublisher::publish(const std::string &channel_name,
const std::string &message)
{
int ret = redisAsyncCommand(_redis_context,
&CRedisPublisher::command_callback, this, "PUBLISH %s %s",
channel_name.c_str(), message.c_str());
if (REDIS_ERR == ret)
{
printf("Publish command failed: %d\n", ret);
return false;
} return true;
} void CRedisPublisher::connect_callback(const redisAsyncContext *redis_context,
int status)
{
if (status != REDIS_OK)
{
printf(": Error: %s\n", redis_context->errstr);
}
else
{
printf(": Redis connected!\n");
}
} void CRedisPublisher::disconnect_callback(
const redisAsyncContext *redis_context, int status)
{
if (status != REDIS_OK)
{
// 这里异常退出，可以尝试重连
printf(": Error: %s\n", redis_context->errstr);
}
} // 消息接收回调函数
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!\n");
assert(false);
return NULL;
} CRedisPublisher *self_this = reinterpret_cast<CRedisPublisher *>(data);
return self_this->event_proc();
} void *CRedisPublisher::event_proc()
{
sem_wait(&_event_sem); // 开启事件分发，event_base_dispatch会阻塞
event_base_dispatch(_event_base); return NULL;
}
redis_subscriber.h
/*************************************************************************
> File Name: redis_subscriber.h
> Author: chenzengba
> Mail: chenzengba@gmail.com
> Created Time: Sat 23 Apr 2016 10:15:09 PM CST
> Description: 封装hiredis，实现消息订阅redis功能
************************************************************************/ #ifndef REDIS_SUBSCRIBER_H
#define REDIS_SUBSCRIBER_H #include <stdlib.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>
#include <string>
#include <vector>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <boost/tr1/functional.hpp> class CRedisSubscriber
{
public:
typedef std::tr1::function<void (const char *, const char *, int)> NotifyMessageFn; // 回调函数对象类型，当接收到消息后调用回调把消息发送出去 CRedisSubscriber();
~CRedisSubscriber(); bool init(const NotifyMessageFn &fn); // 传入回调对象
bool uninit();
bool connect();
bool disconnect(); // 可以多次调用，订阅多个频道
bool subscribe(const std::string &channel_name); private:
// 下面三个回调函数供redis服务调用
// 连接回调
static void connect_callback(const redisAsyncContext *redis_context,
int status); // 断开连接的回调
static void disconnect_callback(const redisAsyncContext *redis_context,
int status); // 执行命令回调
static void command_callback(redisAsyncContext *redis_context,
void *reply, void *privdata); // 事件分发线程函数
static void *event_thread(void *data);
void *event_proc(); private:
// libevent事件对象
event_base *_event_base;
// 事件线程ID
pthread_t _event_thread;
// 事件线程的信号量
sem_t _event_sem;
// hiredis异步对象
redisAsyncContext *_redis_context; // 通知外层的回调函数对象
NotifyMessageFn _notify_message_fn;
};
redis_subscriber.cpp
/*************************************************************************
> File Name: redis_subscriber.cpp
> Author: chenzengba
> Mail: chenzengba@gmail.com
> Created Time: Sat 23 Apr 2016 10:15:09 PM CST
> Description:
************************************************************************/ #include <stddef.h>
#include <assert.h>
#include <string.h>
#include "redis_subscriber.h" CRedisSubscriber::CRedisSubscriber():_event_base(0), _event_thread(0),
_redis_context(0)
{
} CRedisSubscriber::~CRedisSubscriber()
{
} bool CRedisSubscriber::init(const NotifyMessageFn &fn)
{
// initialize the event
_notify_message_fn = fn;
_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 false;
} return true;
} bool CRedisSubscriber::uninit()
{
_event_base = NULL; sem_destroy(&_event_sem);
return true;
} bool CRedisSubscriber::connect()
{
// connect redis
_redis_context = redisAsyncConnect("127.0.0.1", 6379); // 异步连接到redis服务器上，使用默认端口
if (NULL == _redis_context)
{
printf(": Connect redis failed.\n");
return false;
} if (_redis_context->err)
{
printf(": Connect redis error: %d, %s\n",
_redis_context->err, _redis_context->errstr); // 输出错误信息
return false;
} // attach the event
redisLibeventAttach(_redis_context, _event_base); // 将事件绑定到redis context上，使设置给redis的回调跟事件关联 // 创建事件处理线程
int ret = pthread_create(&_event_thread, 0, &CRedisSubscriber::event_thread, this);
if (ret != 0)
{
printf(": create event thread failed.\n");
disconnect();
return false;
} // 设置连接回调，当异步调用连接后，服务器处理连接请求结束后调用，通知调用者连接的状态
redisAsyncSetConnectCallback(_redis_context,
&CRedisSubscriber::connect_callback); // 设置断开连接回调，当服务器断开连接后，通知调用者连接断开，调用者可以利用这个函数实现重连
redisAsyncSetDisconnectCallback(_redis_context,
&CRedisSubscriber::disconnect_callback); // 启动事件线程
sem_post(&_event_sem);
return true;
} bool CRedisSubscriber::disconnect()
{
if (_redis_context)
{
redisAsyncDisconnect(_redis_context);
redisAsyncFree(_redis_context);
_redis_context = NULL;
} return true;
} bool CRedisSubscriber::subscribe(const std::string &channel_name)
{
int ret = redisAsyncCommand(_redis_context,
&CRedisSubscriber::command_callback, this, "SUBSCRIBE %s",
channel_name.c_str());
if (REDIS_ERR == ret)
{
printf("Subscribe command failed: %d\n", ret);
return false;
} printf(": Subscribe success: %s\n", channel_name.c_str());
return true;
} void CRedisSubscriber::connect_callback(const redisAsyncContext *redis_context,
int status)
{
if (status != REDIS_OK)
{
printf(": Error: %s\n", redis_context->errstr);
}
else
{
printf(": Redis connected!");
}
} void CRedisSubscriber::disconnect_callback(
const redisAsyncContext *redis_context, int status)
{
if (status != REDIS_OK)
{
// 这里异常退出，可以尝试重连
printf(": Error: %s\n", redis_context->errstr);
}
} // 消息接收回调函数
void CRedisSubscriber::command_callback(redisAsyncContext *redis_context,
void *reply, void *privdata)
{
if (NULL == reply || NULL == privdata) {
return ;
} // 静态函数中，要使用类的成员变量，把当前的this指针传进来，用this指针间接访问
CRedisSubscriber *self_this = reinterpret_cast<CRedisSubscriber *>(privdata);
redisReply *redis_reply = reinterpret_cast<redisReply *>(reply); // 订阅接收到的消息是一个带三元素的数组
if (redis_reply->type == REDIS_REPLY_ARRAY &&
redis_reply->elements == 3)
{
printf(": Recieve message:%s:%d:%s:%d:%s:%d\n",
redis_reply->element[0]->str, redis_reply->element[0]->len,
redis_reply->element[1]->str, redis_reply->element[1]->len,
redis_reply->element[2]->str, redis_reply->element[2]->len); // 调用函数对象把消息通知给外层
self_this->_notify_message_fn(redis_reply->element[1]->str,
redis_reply->element[2]->str, redis_reply->element[2]->len);
}
} void *CRedisSubscriber::event_thread(void *data)
{
if (NULL == data)
{
printf(": Error!\n");
assert(false);
return NULL;
} CRedisSubscriber *self_this = reinterpret_cast<CRedisSubscriber *>(data);
return self_this->event_proc();
} void *CRedisSubscriber::event_proc()
{
sem_wait(&_event_sem); // 开启事件分发，event_base_dispatch会阻塞
event_base_dispatch(_event_base); return NULL;
}
