linux下使用hiredis非同步API實現sub/pub訊息訂閱和釋出的功能

阿新 • • 發佈：2019-01-14

本文轉載自連結：最近使用redis的c介面——hiredis，使客戶端與redis伺服器通訊，實現訊息訂閱和釋出（PUB/SUB）的功能，我把遇到的一些問題和解決方法列出來供大家學習。廢話不多說，先貼程式碼。 redis_publisher.h

/*************************************************************************
    > File Name: redis_publisher.h
    > Author: chenzengba
    > Mail: [email protected] 
    > 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;
};

#endif

redis_publisher.cpp

/*************************************************************************
    > File Name: redis_publisher.cpp
    > Author: chenzengba
    > Mail: [email protected] 
    > 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: [email protected] 
    > 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;
};

#endif

redis_subscriber.cpp:

/*************************************************************************
    > File Name: redis_subscriber.cpp
    > Author: chenzengba
    > Mail: [email protected] 
    > 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;
}

問題1：hiredis官網沒有非同步介面的實現例子。

hiredis提供了幾個非同步通訊的API，一開始根據API名字的理解，我們實現了跟redis伺服器建立連線、訂閱和釋出的功能，可在實際使用的時候，程式並沒有像我們預想的那樣，除了能夠建立連線外，任何事情都沒發生。網上查了很多資料，原來hiredis的非同步實現是通過事件來分發redis傳送過來的訊息的，hiredis可以使用libae、libev、libuv和libevent中的任何一個實現事件的分發，網上的資料提示使用libae、libev和libuv可能發生其他問題，這裡為了方便就選用libevent。hireds官網並沒有對libevent做任何介紹，也沒用說明使用非同步機制需要引入事件的介面，所以一開始走了很多彎路。關於libevent的使用這裡就不再贅述，詳情可以見libevent官網。 libevent官網：http://libevent.org/ libevent api文件：https://www.monkey.org/~provos/libevent/doxygen-2.0.1/include_2event2_2event_8h.html#6e9827de8c3014417b11b48f2fe688ae CRedisPublisher和CRedisSubscriber的初始化過程：初始化事件處理，並獲得事件處理的例項：

_event_base = event_base_new();

在獲得redisAsyncContext *之後，呼叫

redisLibeventAttach(_redis_context, _event_base);

這樣就將事件處理和redis關聯起來，最後在另一個執行緒呼叫

event_base_dispatch(_event_base);

啟動事件的分發，這是一個阻塞函式，因此，建立了一個新的執行緒處理事件分發，值得注意的是，這裡用訊號燈_event_sem控制執行緒的啟動，意在程式呼叫

    redisAsyncSetConnectCallback(_redis_context, 
        &CRedisSubscriber::connect_callback);
    redisAsyncSetDisconnectCallback(_redis_context,
        &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://libevent.org/下載最新的穩定版解壓後進入解壓目錄，執行命令 ./configure -prefix=/usr sudo make && make install boost庫：直接執行安裝：sudo apt-get install libboost-dev 如果你不是用std::tr1::function的函式物件來給外層通知訊息，就不需要boost庫。你可以用介面的形式實現回撥，把介面傳給CRedisSubscribe類，讓它在接收到訊息後呼叫介面回撥，通知外層。

問題4 如何使用

最後貼出例子程式碼。 publisher.cpp，實現釋出訊息：

/*************************************************************************
    > File Name: publisher.cpp
    > Author: chenzengba
    > Mail: [email protected] 
    > Created Time: Sat 23 Apr 2016 12:13:24 PM CST
 ************************************************************************/

#include "redis_publisher.h"

int main(int argc, char *argv[])
{
    CRedisPublisher publisher;

    bool ret = publisher.init();
    if (!ret) 
    {
        printf("Init failed.\n");
        return 0;
    }

    ret = publisher.connect();
    if (!ret)
    {
        printf("connect failed.");
        return 0;
    }

    while (true)
    {
        publisher.publish("test-channel", "Test message");
        sleep(1);
    }

    publisher.disconnect();
    publisher.uninit();
    return 0;
}

subscriber.cpp實現訂閱訊息：

/*************************************************************************
    > File Name: subscriber.cpp
    > Author: chenzengba
    > Mail: [email protected] 
    > Created Time: Sat 23 Apr 2016 12:26:42 PM CST
 ************************************************************************/

#include "redis_subscriber.h"

void recieve_message(const char *channel_name,
    const char *message, int len)
{
    printf("Recieve message:\n    channel name: %s\n    message: %s\n",
        channel_name, message);
}

int main(int argc, char *argv[])
{
    CRedisSubscriber subscriber;
    CRedisSubscriber::NotifyMessageFn fn = 
        bind(recieve_message, std::tr1::placeholders::_1,
        std::tr1::placeholders::_2, std::tr1::placeholders::_3);

    bool ret = subscriber.init(fn);
    if (!ret)
    {
        printf("Init failed.\n");
        return 0;
    }

    ret = subscriber.connect();
    if (!ret)
    {
        printf("Connect failed.\n");
        return 0;
    }

    subscriber.subscribe("test-channel");

    while (true)
    {
        sleep(1);
    }

    subscriber.disconnect();
    subscriber.uninit();

    return 0;
}

關於編譯的問題：在g++中編譯，注意要加上-lhiredis -levent引數，下面是一個簡單的Makefile：

EXE=server_main client_main
CC=g++
FLAG=-lhiredis -levent
OBJ=redis_publisher.o publisher.o redis_subscriber.o subscriber.o

all:$(EXE)

$(EXE):$(OBJ)
	$(CC) -o publisher redis_publisher.o publisher.o $(FLAG)
	$(CC) -o subscriber redis_subscriber.o subscriber.o $(FLAG)

redis_publisher.o:redis_publisher.h
redis_subscriber.o:redis_subscriber.h

publisher.o:publisher.cpp
	$(CC) -c publisher.cpp

subscriber.o:subscriber.cpp
	$(CC) -c subscriber.cpp
clean:
	rm publisher subscriber *.o

致謝： redis非同步API使用libevent：http://www.tuicool.com/articles/N73uuu

linux下使用hiredis非同步API實現sub/pub訊息訂閱和釋出的功能

問題1：hiredis官網沒有非同步介面的實現例子。

問題2 奇特的‘ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context’錯誤

問題3 相關依賴的庫

問題4 如何使用

linux下使用hiredis非同步API實現sub/pub訊息訂閱和釋出的功能

Linux下mysql基於MyCat實現主從復制和讀寫分離

Linux下Wi-Fi的實現：wireless_tools和wpa_supplicant

Linux下的簡易shell實現

Linux下網路socket程式設計——實現伺服器（select）與多個客戶端通訊

linux下操作svn，實現根據時間段檢視某個指定使用者提交的記錄

Linux下進度條的實現（加彩色版本）

Linux下用c語言實現發送http請求方式可以Get或者Post例程參考

Linux下用c語言實現傳送http請求方式可以Get或者Post例程參考

Linux下併發伺服器的實現

Linux下安裝Redis並實現遠端連線，Redis Desktop Manager視覺化連線

Linux下用c語言實現傳送http請求

linux下使用crontab如何實現mysql資料庫每天自動備份定時備份

live555在Linux下最簡單地實現實時流媒體點播

Linux下如何用C實現MD5加密

linux下利用openssl來實現證書的頒發（詳細步驟）

Linux下通用掃描器API——SANE（ Scanner Access Now Easy）

Linux下簡易web伺服器實現

linux下socket程式設計 select實現非阻塞模式多臺客戶端與伺服器通訊

Linux下使用libevent庫實現伺服器端程式設計

linux下使用hiredis非同步API實現sub/pub訊息訂閱和釋出的功能

問題1：hiredis官網沒有非同步介面的實現例子。

問題2 奇特的‘ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context’錯誤

問題3 相關依賴的庫

問題4 如何使用

相關推薦