利用redis制作消息队列

redis在游戏服务器中的使用初探(一) 环境搭建
redis在游戏服务器中的使用初探(二) 客户端开源库选择
redis在游戏服务器中的使用初探(三) 信息存储
redis在游戏服务器中的使用初探(四) redis应用

在学习分布式对象存储的期间，有这么一个需求

"多个接口服务(本文当作客户端Clinet)需要以固定间隔向所有的多个服务器发送心跳，保证服务器确认客户端状态。

服务器在接收到文件读取请求时候，会广播询问所有数据服务器(本文也当作服务器)存储的数据情况"

以上一对多  的询问，是需要消息队列来进行通讯的

但是其实 redis也可以作为轻量级的消息队列来完成这个需求。

结构图

服务器开启一个线程进行redis订阅模式，当有人在指定频道发布消息时，所有订阅该频道的节点都可以接收到消息。

但是订阅操作如果我们不想采取固定时间间隔去获取频道是否有消息这么LOW的方案，其实是需要做成异步模式的。

而windows下 hredis异步模式是需要libevent支持的。 两者都是linux下运行良好的开源库，在windows下却是问题多多。

经过多次尝试，我决定放弃使用这两个开源库而选择cpp-redis。(linux下使用hredis和libevent ，有时间会试试)

流程如下：

一个服务节点 需要开启一个线程 进行客户端消息队列的订阅，每当收到消息就会调用收到消息的回调函数

而最初开启的服务节点的运行线程会定时的在服务器消息队列发布询问数据存储的信息。

客户端节点则相反 开启一个线程 定时向客户端消息队列发布心跳信息。

最初开启的客户端节点进行服务器消息队列的订阅，若收到服务器的数据存储询问，则进行本身是否存储该数据的判断

由于资源有限，最开始我们开启了5个线程 来模拟 2个服务器和 3个客户端

代码如下

#include <iostream>
#include <Winsock2.h>
#include <thread>
#include <mutex>

#include "cpp_redis/cpp_redis"
#include "tacopie/tacopie"

using namespace std;

const int serverThreadNum = 2;
const int clientThreadNum = 3;
const int heartBeatTime = 1;
const int ServerQueryTime = 1;
const std::string clientChanName = "ClientChan";
const std::string serverChanName = "ServerChan";
std::mutex g_mutex;

class WinsockGuard {
public:
    WinsockGuard() {
        WORD version = MAKEWORD(2, 2);
        if (WSAStartup(version, &data) != 0) {
            std::cerr << "WSAStartup() failure!" << std::endl;
            return;
        }
    }

    ~WinsockGuard() {
        WSACleanup();
    }
private:
    WSADATA data;
};

bool SubcribCommFunc(int threadNum,bool isServer) {
    cpp_redis::subscriber sub;

    try {
        sub.connect("127.0.0.1", 6379, [](const std::string& host, std::size_t port, cpp_redis::subscriber::connect_state status) {
            if (status == cpp_redis::subscriber::connect_state::dropped) {
                {std::lock_guard<std::mutex> l(g_mutex); std::cout << "client disconnected from " << host << ":" << port << std::endl; }
                //should_exit.notify_all();
            }
        });

    }
    catch (std::exception& e) {
        {std::lock_guard<std::mutex> l(g_mutex); std::cerr << "in " << __FUNCTION__ << ".err = " << e.what() << std::endl; }
        return false;
    }
    std::string chanName;
    if (isServer) {chanName = clientChanName;}
    else {chanName = serverChanName;}

    sub.subscribe(chanName.c_str(), [threadNum, isServer](const std::string& chan, const std::string& msg) {
        string s;
        if (isServer)s = "server ";
        else s = "client ";
        s += to_string(threadNum);s += " recv ";
        {std::lock_guard<std::mutex> l(g_mutex); std::cout << s.c_str() << chan << ": " << msg << std::endl; }
        //todo Check heatbeat or response
    });
    sub.commit();

    while (1) {
        std::this_thread::sleep_for(std::chrono::seconds(50000));
    }

    return true;
}

bool RecvClientInfo(int i) {
    return SubcribCommFunc(i,true);
}

bool PublishCommFunc(int threadNum, bool isServer, string publishStr) {
    cpp_redis::client client;
    try {
        client.connect("127.0.0.1", 6379, [threadNum, isServer,&publishStr](const std::string& host, std::size_t port, cpp_redis::client::connect_state status) {
            if (status == cpp_redis::client::connect_state::dropped) {
                {std::lock_guard<std::mutex> l(g_mutex);  std::cout << "disconnected from " << host << ":" << port << std::endl; }
            }
        });
        while (1) {
            std::string chanName;
            if (isServer) {chanName = serverChanName;}
            else {    chanName = clientChanName;}

            client.publish(chanName.c_str(), publishStr.c_str());
            client.commit();

            int PubliLoopTime = 9;
            if (isServer) {PubliLoopTime = ServerQueryTime;}
            else {PubliLoopTime = heartBeatTime;}

            std::this_thread::sleep_for(std::chrono::seconds(PubliLoopTime));
        }
    }
    catch (std::exception& e) {
        {std::lock_guard<std::mutex> l(g_mutex); std::cerr << "in " << __FUNCTION__ << ".err = " << e.what() << std::endl; }
        return false;
    }

    return true;
}

void QueryWhoSaveDataLoop(int i) {
    string s = "Server thread ";s += to_string(i);s += " query Who save data? ";
    PublishCommFunc(i, true, s);
    return;
}

void ServerFunc(int i) {
    {std::lock_guard<std::mutex> l(g_mutex);std::cout << "Enter ServerFunc threadNo = " << i << std::endl;}
    //开启一个订阅客户端消息队列的线程 接受客户端的心跳包
    thread t = thread(RecvClientInfo, i);
    t.detach();

    //开启一个定时检测心跳超时的客户端 todo

    //本线程不定时随机 发送一个询问各个客户端是否保存有数据
    QueryWhoSaveDataLoop(i);

    std::this_thread::sleep_for(std::chrono::seconds(500));
}

void SendHeatBeatOnTime(int threadNum, int sendTime) {
    string s = "client thread ";s += to_string(threadNum);s += " send heartbeat";
    PublishCommFunc(threadNum, false, s);
}

void ClientFunc(int i) {
    {std::lock_guard<std::mutex> l(g_mutex);std::cout << "Enter ClientFunc threadNo = " << i << std::endl;}

    //开启一个线程 定时发送心跳包
    int s = heartBeatTime;
    std::thread t = thread(SendHeatBeatOnTime, i, s);
    t.detach();

    SubcribCommFunc(i, false);
}

void Start() {
    thread serverThread[serverThreadNum];
    thread clientThread[clientThreadNum];

    for (int i = 0; i < serverThreadNum; i++) {
        serverThread[i] = thread(ServerFunc, i);
    }
    for (int i = 0; i < clientThreadNum; i++) {
        clientThread[i] = thread(ClientFunc, i);
    }
    //==================================================
    for (int i = 0; i < serverThreadNum; i++) {
        serverThread[i].join();
    }
    for (int i = 0; i < clientThreadNum; i++) {
        clientThread[i].join();
    }
}

int main()
{
    WinsockGuard g;
    Start();
    std::cout << "Finish!
";
}

开启redis  运行代码如图

番外： 补上我在ubuntu下进行的libevent + hiredis的异步测试

首先是安装源头更新 更新 gcc  g++  make 等工具

sudo apt-get update

sudo apt-get install g++ gcc

安装 redis server

sudo apt-get install redis-server 

现在可以通过下面的命令查看到该进程：
ps -ef|grep redis

然后安装 hiredis 和 libevent

sudo apt-get install libhiredis-dev

sudo apt-get install libevent-dev

安装完成验证下是否正确安装

编写libevent 示例代码

#include <event.h>
#include <stdio.h>

struct event ev;
struct timeval tv;


void time_cb(int fd, short event, void *argc)
{
    printf(  "timer wakeup
"); 
    event_add(&ev, &tv);
}

int main()
{
    struct event_base *base = event_init();

    tv.tv_sec = 1;
    tv.tv_usec = 0;
    evtimer_set(&ev, time_cb, NULL);
    event_add(&ev, &tv);
    event_base_dispatch(base);

    return 0;
}

执行编译命令并运行 gcc -o eventexe libeventTest.c  -levent

./eventexe  执行无错误则验证通过

编写hiredis示例代码

#include <stdio.h> 
#include <hiredis/hiredis.h> 
int main() 
{ 
     redisContext *conn  = redisConnect("127.0.0.1",6379); 
     if(conn != NULL && conn->err) 
     {   
         printf("connection error: %s
",conn->errstr); 
         return 0; 
     }   
     redisReply *reply = (redisReply*)redisCommand(conn,"set foo 1234"); 
     freeReplyObject(reply); 
             
     reply = redisCommand(conn,"get foo"); 
     printf("%s
",reply->str); 
     freeReplyObject(reply); 
             
     redisFree(conn); 
     return 0; 
}

执行编译命令并运行 gcc -o hiredisCli hiredisTest.c  -lhiredis

./hiredisCli  执行无错误则验证通过

libevent和hiredis都确认无误后 开始测试异步代码

编写异步示例代码

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

#include <hiredis/hiredis.h> 
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>

void getCallback(redisAsyncContext *c, void *r, void *privdata) {
    redisReply *reply = r;
    if (reply == NULL) return;
    printf("argv[%s]: %s
", (char*)privdata, reply->str);

    /* Disconnect after receiving the reply to GET */
    redisAsyncDisconnect(c);
}

void connectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        printf("Error: %s
", c->errstr);
        return;
    }
    printf("Connected...
");
}

void disconnectCallback(const redisAsyncContext *c, int status) {
    if (status != REDIS_OK) {
        printf("Error: %s
", c->errstr);
        return;
    }
    printf("Disconnected...
");
}

int main (int argc, char **argv) {
    signal(SIGPIPE, SIG_IGN);
    struct event_base *base = event_base_new();

    redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
    if (c->err) {
        /* Let *c leak for now... */
        printf("Error: %s
", c->errstr);
        return 1;
    }

    redisLibeventAttach(c,base);
    redisAsyncSetConnectCallback(c,connectCallback);
    redisAsyncSetDisconnectCallback(c,disconnectCallback);
    redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
    redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
    event_base_dispatch(base);
    return 0;
}

执行编译命令并运行  

gcc -o async async.c -lhiredis -levent

./async

测试成功