Redis 内存管理与事件处理

《Redis 内存管理与事件处理》要点：
本文介绍了Redis 内存管理与事件处理，希望对您有用。如果有疑问，可以联系我们。

1 Redis内存治理

Redis内存管理相关文件为zmalloc.c/zmalloc.h,其只是对C中内存管理函数做了简单的封装,屏蔽了底层平台的差别,并增加了内存使用情况统计的功能.

void *zmalloc(size_t size) {

// 多申请的一部分内存用于存储当前分配了多少本身的内存

void *ptr = malloc(size+PREFIX_SIZE);

if (!ptr) zmalloc_oom_handler(size);

#ifdef HAVE_MALLOC_SIZE

update_zmalloc_stat_alloc(zmalloc_size(ptr));

return ptr;

#else

*((size_t*)ptr) = size;

// 内存分派统计

update_zmalloc_stat_alloc(size+PREFIX_SIZE);

return (char*)ptr+PREFIX_SIZE;

#endif

}

内存结构图示：

2 变乱处理

Redis的事件类型分为时间事件和文件事件,文件事件也便是网络连接事件.时间事件的处理是在epoll_wait返回处理文件事件后处理的,每次epoll_wait的超时时间都是Redis最近的一个定时器时间.

Redis在进行变乱处理前,首先会进行初始化,初始化的主要逻辑在main/initServer函数中.初始化流程主要做的工作如下：

• 设置旌旗灯号回调函数.

• 创建变乱循环机制,即调用epoll_create.

• 创建服务监听端口,创建定时变乱,并将这些变乱添加到变乱机制中.

void initServer(void) {

int j;

// 设置信号对应的处置函数

signal(SIGHUP,SIG_IGN);

signal(SIGPIPE,SIG_IGN);

setupSignalHandlers();

...

createSharedObjects();

adjustOpenFilesLimit();

// 创建变乱循环机制,及调用epoll_create创建epollfd用于变乱监听

server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);

server.db = zmalloc(sizeof(redisDb)*server.dbnum);

/* Open the TCP listening socket for the user commands. */

// 创立监听服务端口,socket/bind/listen

if (server.port != 0 &&

listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)

exit(1);

...

/* Create the Redis databases,and initialize other internal state. */

for (j = 0; j < server.dbnum; j++) {

server.db[j].dict = dictCreate(&dbDictType,NULL);

server.db[j].expires = dictCreate(&keyptrDictType,NULL);

server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);

server.db[j].ready_keys = dictCreate(&setDictType,NULL);

server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);

server.db[j].eviction_pool = evictionPoolAlloc();

server.db[j].id = j;

server.db[j].avg_ttl = 0;

}

...

/* Create the serverCron() time event,that's our main way to process

* background operations. 创建定时变乱 */

if(aeCreateTimeEvent(server.el,1,serverCron,NULL,NULL) == AE_ERR) {

serverPanic("Can't create the serverCron time event.");

exit(1);

}

/* Create an event handler for accepting new connections in TCP and Unix

* domain sockets. */

for (j = 0; j < server.ipfd_count; j++) {

if (aeCreateFileEvent(server.el,server.ipfd[j],AE_READABLE,

acceptTcpHandler,NULL) == AE_ERR)

{

serverPanic(

"Unrecoverable error creating server.ipfd file event.");

}

}

// 将变乱加入到变乱机制中,调用链为 aeCreateFileEvent/aeApiAddEvent/epoll_ctl

if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,

acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");

/* Open the AOF file if needed. */

if (server.aof_state == AOF_ON) {

server.aof_fd = open(server.aof_filename,

O_WRONLY|O_APPEND|O_CREAT,0644);

if (server.aof_fd == -1) {

serverLog(LL_WARNING,"Can't open the append-only file: %s",

strerror(errno));

exit(1);

}

}

...

}

变乱处理流程

变乱处理函数链：aeMain / aeProcessEvents / aeApiPoll / epoll_wait.

常见的事件机制处理流程是：调用epoll_wait等待事件来临,然后遍历每一个epoll_event,提取epoll_event中的events和data域,data域常用来存储fd或者指针,不过一般的做法是提取出events和data.fd,然后根据fd找到对应的回调函数,fd与对应回调函数之间的映射关系可以存储在特定的数据结构中,好比数组或者哈希表,然后调用事件回调函数来处理.

Redis中用了一个数组来保留fd与回调函数的映射关系,使用数组的优点就是简单高效,但是数组一般使用在建立的连接不太多情况,而Redis正好符合这个情况,一般Redis的文件事件大都是客户端建立的连接,而客户端的连接个数是一定的,该数量通过配置项maxclients来指定.

static int aeApiPoll(aeEventLoop *eventLoop,struct timeval *tvp) {

aeApiState *state = eventLoop->apidata;

int retval,numevents = 0;

retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,

tvp 必修 (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);

if (retval > 0) {

int j;

numevents = retval;

for (j = 0; j < numevents; j++) {

int mask = 0;

struct epoll_event *e = state->events+j;

if (e->events & EPOLLIN) mask |= AE_READABLE;

if (e->events & EPOLLOUT) mask |= AE_WRITABLE;

if (e->events & EPOLLERR) mask |= AE_WRITABLE;

if (e->events & EPOLLHUP) mask |= AE_WRITABLE;

eventLoop->fired[j].fd = e->data.fd;

eventLoop->fired[j].mask = mask;

}

}

return numevents;

}

int aeProcessEvents(aeEventLoop *eventLoop,int flags)

{

numevents = aeApiPoll(eventLoop,tvp);

for (j = 0; j < numevents; j++) {

// 从eventLoop->events数组中查找对应的回调函数

aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];

int mask = eventLoop->fired[j].mask;

int fd = eventLoop->fired[j].fd;

int rfired = 0;

/* note the fe->mask & mask & ... code: maybe an already processed

* event removed an element that fired and we still didn't

* processed,so we check if the event is still valid. */

if (fe->mask & mask & AE_READABLE) {

rfired = 1;

fe->rfileProc(eventLoop,fd,fe->clientData,mask);

}

if (fe->mask & mask & AE_WRITABLE) {

if (!rfired || fe->wfileProc != fe->rfileProc)

fe->wfileProc(eventLoop,mask);

}

processed++;

}

...

}

文件变乱的监听

Redis监听端口的变乱回调函数链是：acceptTcpHandler / acceptCommonHandler / createClient / aeCreateFileEvent / aeApiAddEvent / epoll_ctl.

在Reids监听事件处理流程中,会将客户端的连接fd添加到事件机制中,并设置其回调函数为readQueryFromClient,该函数负责处理客户端的命令哀求.

命令处置流程

命令处理流程链是：readQueryFromClient / processInputBuffer / processCommand / call / 对应命令的回调函数(c->cmd->proc),好比get key命令的处理回调函数为getCommand.getCommand的执行流程是先到client对应的数据库字典中根据key来查找数据,然后根据响应消息格式将查询结果填充到响应消息中.

IT进修就选优就业

3 如何添加自界说命令

如安在Redis中添加自定的命令呢?其中只需要改动以下几个地方就行了,比如自定义命令random xxx,然后返回redis: xxx,因为hello xxx和get key类似,所以就依葫芦画瓢.random命令用来返回一个小于xxx的随机值.

首先在redisCommandTable数组中添加自界说的命令,redisCommandTable数组界说在server.c中.然后在getCommand界说处后面添加randomCommand的界说,getCommand界说在t_string.c中.最后在server.h中添加helloCommand的声明.整个修改patch文件如下,代码基于redis-2.8.9版本.

From 5304020683078273c1bc6cc9666dab95efa18607 Mon Sep 17 00:00:00 2001

From: luoxn28 <[e-mail]>

Date: Fri,30 Jun 2017 04:43:47 -0700

Subject: [PATCH] add own command: random num

---

src/server.c | 3 ++-

src/server.h | 1 +

src/t_string.c | 44 ++++++++++++++++++++++++++++++++++++++++++++

3 files changed,47 insertions(+),1 deletion(-)

diff --git a/src/server.c b/src/server.c

index 609f396..e040104 100644

--- a/src/server.c

+++ b/src/server.c

@@ -296,7 +296,8 @@ struct redisCommand redisCommandTable[] = {

{"pfdebug",pfdebugCommand,-3,"w",0},

{"post",securityWarningCommand,-1,"lt",

{"host:",

- {"latency",latencyCommand,-2,"aslt",0}

+ {"latency",

+ {"random",randomCommand,2,"rF",0}

};

struct evictionPoolEntry *evictionPoolAlloc(void);

diff --git a/src/server.h b/src/server.h

index 3fa7c3a..427ac92 100644

--- a/src/server.h

+++ b/src/server.h

@@ -1485,6 +1485,7 @@ void setnxCommand(client *c);

void setexCommand(client *c);

void psetexCommand(client *c);

void getCommand(client *c);

+void randomCommand(client *c);

void delCommand(client *c);

void existsCommand(client *c);

void setbitCommand(client *c);

diff --git a/src/t_string.c b/src/t_string.c

index 8c737c4..df4022d 100644

--- a/src/t_string.c

+++ b/src/t_string.c

@@ -173,6 +173,50 @@ void getCommand(client *c) {

getGenericCommand(c);

}

+static bool checkRandomNum(char *num)

+{

+ char *c = num;

+

+ while (*c != '\0') {

+ if (!(('0' <= *c) && (*c <= '9'))) {

+ return false;

+ }

+ c++;

+ }

+

+ return true;

+}

+

+/**

+ * command: random n

+ * return a random num < n,if n <= 0,return 0

+ * @author: luoxiangnan

+ */

+void randomCommand(client *c)

+{

+ char buff[64] = {0};

+ int num = 0;

+ robj *o = NULL;

+

+ if (!checkRandomNum(c->argv[1]->ptr)) {

+ o = createObject(OBJ_STRING,sdsnewlen("sorry,it's not a num :(",

+ strlen("sorry,it's not a num :(")));

+ addReplyBulk(c,o);

+ return;

+ }

+

+ sscanf(c->argv[1]->ptr,"%d",&num);

+ if (num > 0) {

+ num = random() % num;

+ } else {

+ num = 0;

+ }

+

+ sprintf(buff,"%s %d","redis: ",num);

+ o = createObject(OBJ_STRING,sdsnewlen(buff,strlen(buff)));

+ addReplyBulk(c,o);

+}

+

void getsetCommand(client *c) {

if (getGenericCommand(c) == C_ERR) return;

c->argv[2] = tryObjectEncoding(c->argv[2]);

--

1.8.3.1

成果如下所示：

更多优质内容保举：

体验别样旅游感触感染 中公教育推出全新互联网+北京游,零利润体验10天9晚帝都风情：

http://www.ujiuye.com/zt/qgsqxly/必修wt.bd=zdy35845tt

有钱任性,某公司豪掷500万赞助20左右年轻人找工作,起因是做善良的人：

http://www.ujiuye.com/zt/jyfc/必修wt.bd=zdy35845tt

学安卓,免学费!50天兴致课程等你来抢!

http://www.ujiuye.com/xydt/2017/13042.html必修wt.bd=zdy35845tt

欢迎参与《Redis 内存管理与事件处理》讨论，分享您的想法，编程之家PHP学院为您提供专业教程。