快捷導(dǎo)航

使用kendynet構(gòu)建異步redis訪問(wèn)服務(wù)

更新時(shí)間：2014年01月24日 15:46:17 作者：

這篇文章主要介紹了在kendynet上寫(xiě)的一個(gè)簡(jiǎn)單的redis異步訪問(wèn)接口,大家參考使用吧

最近開(kāi)始在kendynet上開(kāi)發(fā)手游服務(wù)端，游戲類型是生存挑戰(zhàn)類的，要存儲(chǔ)的數(shù)據(jù)結(jié)構(gòu)和類型都比較簡(jiǎn)單，于是選擇了用redis做存儲(chǔ)，數(shù)據(jù)類型使用string基本就足夠了。于是在kendynet上寫(xiě)了一個(gè)簡(jiǎn)單的redis異步訪問(wèn)接口

設(shè)計(jì)理念
1.項(xiàng)目時(shí)間緊迫，不打算提供一個(gè)大而全的訪問(wèn)接口，只提供一個(gè)request接口用以發(fā)出redis請(qǐng)求.

2.數(shù)據(jù)在redis中key和value都存儲(chǔ)為string,由使用者負(fù)責(zé)將數(shù)據(jù)序列化成string,從string反序列化回?cái)?shù)據(jù).

3.服務(wù)支持本地訪問(wèn)和遠(yuǎn)程訪問(wèn)，服務(wù)自動(dòng)根據(jù)請(qǐng)求發(fā)起的位置將結(jié)果返回給請(qǐng)求者.

4.數(shù)據(jù)庫(kù)操作結(jié)果通過(guò)異步消息返回給調(diào)用者

下面先看一個(gè)使用示例：

復(fù)制代碼代碼如下:

#include <stdio.h>
#include <stdlib.h>
#include "core/msgdisp.h"
#include <stdint.h>
#include "testcommon.h"
#include "core/db/asyndb.h"

asyndb_t asydb;

int g = 0;
int count = 0;

void db_setcallback(struct db_result *result);

void db_getcallback(struct db_result *result)
{
    //printf("%s\n",result->result_str);
    count++;
    char req[256];
    snprintf(req,256,"set key%d %d",g,g);
    if(0 != asydb->request(asydb,new_dbrequest(db_set,req,db_setcallback,result->ud,make_by_msgdisp((msgdisp_t)result->ud))))
        printf("request error\n");
}

void db_setcallback(struct db_result *result)
{
    if(result->ud == NULL) printf("error\n");
    char req[256];
    snprintf(req,256,"get key%d",g);
    g = (g+1)%102400;
    asydb->request(asydb,new_dbrequest(db_get,req,db_getcallback,result->ud,make_by_msgdisp((msgdisp_t)result->ud)));
}

int32_t asynprocesspacket(msgdisp_t disp,msgsender sender,rpacket_t rpk)
{
    uint16_t cmd = rpk_read_uint16(rpk);
    if(cmd == CMD_DB_RESULT)
    {
        struct db_result *result = rpk_read_dbresult(rpk);
        result->callback(result);
        free_dbresult(result);
    }
    return 1;
}

static void *service_main(void *ud){
    msgdisp_t disp = (msgdisp_t)ud;
    while(!stop){
        msg_loop(disp,50);
    }
    return NULL;
}

int main(int argc,char **argv)
{
    setup_signal_handler();
    msgdisp_t disp1 = new_msgdisp(NULL,
                                  NULL,
                                  NULL,
                                  NULL,
                                  asynprocesspacket,
                                  NULL);

thread_t service1 = create_thread(THREAD_JOINABLE);

    msgdisp_t disp2 = new_msgdisp(NULL,
                                  NULL,
                                  NULL,
                                  NULL,
                                  asynprocesspacket,
                                  NULL);

    thread_t service2 = create_thread(THREAD_JOINABLE);
    asydb = new_asyndb();
    asydb->connectdb(asydb,"127.0.0.1",6379);
    asydb->connectdb(asydb,"127.0.0.1",6379);
    //發(fā)出第一個(gè)請(qǐng)求uu
    char req[256];
    snprintf(req,256,"set key%d %d",g,g);

asydb->request(asydb,new_dbrequest(db_set,req,db_setcallback,disp1,make_by_msgdisp(disp1)));
thread_start_run(service1,service_main,(void*)disp1);

asydb->request(asydb,new_dbrequest(db_set,req,db_setcallback,disp2,make_by_msgdisp(disp2)));
thread_start_run(service2,service_main,(void*)disp2);

    uint32_t tick,now;
    tick = now = GetSystemMs();
    while(!stop){
        sleepms(100);
        now = GetSystemMs();
        if(now - tick > 1000)
        {
            printf("count:%d\n",count);
            tick = now;
            count = 0;
        }
    }
    thread_join(service1);
    thread_join(service2);
    return 0;
}

上面的示例程序創(chuàng)建了一個(gè)redis異步處理器,然后建立了兩個(gè)到同一個(gè)redis服務(wù)器的連接,在實(shí)現(xiàn)中，每個(gè)連接都會(huì)創(chuàng)建一個(gè)工作線程，用以完成數(shù)據(jù)庫(kù)請(qǐng)求.這些工作線程會(huì)共享一個(gè)任務(wù)隊(duì)列，使用者發(fā)出的請(qǐng)求被投遞到任務(wù)隊(duì)列中，由工作線程提取并執(zhí)行.

之后創(chuàng)建兩個(gè)消息分離器和兩個(gè)線程，然后發(fā)起兩個(gè)set請(qǐng)求和啟動(dòng)消息分離器線程.

當(dāng)set返回后，由消息分離器回調(diào)dbsetcallback，在dbsetcallback中繼續(xù)發(fā)起一個(gè)get請(qǐng)求, 在get的回調(diào)db_getcallback繼續(xù)發(fā)起新的set請(qǐng)求，如此反復(fù).