初始化服務(wù)器代碼如下：

void initServer() {int j;signal(SIGHUP, SIG_IGN); signal(SIGPIPE, SIG_IGN);setupSignalHandlers();if (server.syslog_enabled) {openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT,server.syslog_facility);}server.current_client = NULL;server.clients = listCreate(); //創(chuàng)建客戶隊(duì)列server.clients_to_close = listCreate(); //創(chuàng)建將關(guān)閉的客戶隊(duì)列server.slaves = listCreate(); //創(chuàng)建從機(jī)隊(duì)列server.monitors = listCreate(); //創(chuàng)建監(jiān)控隊(duì)列server.slaveseldb = -1; /* Force to emit the first SELECT command. */server.unblocked_clients = listCreate(); //創(chuàng)建非堵塞客戶隊(duì)列server.ready_keys = listCreate(); //創(chuàng)建可讀key隊(duì)列createSharedObjects(); // 創(chuàng)建共享對(duì)象adjustOpenFilesLimit(); //改變可打開(kāi)文件的最大數(shù)量server.el = aeCreateEventLoop(server.maxclients+REDIS_EVENTLOOP_FDSET_INCR); //創(chuàng)建事件處理server.db = zmalloc(sizeof(redisDb)*server.dbnum); //分別db內(nèi)存/* Open the TCP listening socket for the user commands. */if (listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR) //監(jiān)聽(tīng)端口exit(1);/* Open the listening Unix domain socket. */if (server.unixsocket != NULL) {unlink(server.unixsocket); /* don't care if this fails */server.sofd = anetUnixServer(server.neterr,server.unixsocket,server.unixsocketperm);if (server.sofd == ANET_ERR) {redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr);exit(1);}}/* Abort if there are no listening sockets at all. */if (server.ipfd_count == 0 && server.sofd < 0) {redisLog(REDIS_WARNING, "Configured to not listen anywhere, exiting.");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].id = j;server.db[j].avg_ttl = 0;}server.pubsub_channels = dictCreate(&keylistDictType,NULL);server.pubsub_patterns = listCreate();listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);server.cronloops = 0;server.rdb_child_pid = -1;server.aof_child_pid = -1;aofRewriteBufferReset();server.aof_buf = sdsempty();server.lastsave = time(NULL); /* At startup we consider the DB saved. */server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */server.rdb_save_time_last = -1;server.rdb_save_time_start = -1;server.dirty = 0;server.stat_numcommands = 0;server.stat_numconnections = 0;server.stat_expiredkeys = 0;server.stat_evictedkeys = 0;server.stat_starttime = time(NULL);server.stat_keyspace_misses = 0;server.stat_keyspace_hits = 0;server.stat_peak_memory = 0;server.stat_fork_time = 0;server.stat_rejected_conn = 0;server.stat_sync_full = 0;server.stat_sync_partial_ok = 0;server.stat_sync_partial_err = 0;memset(server.ops_sec_samples,0,sizeof(server.ops_sec_samples));server.ops_sec_idx = 0;server.ops_sec_last_sample_time = mstime();server.ops_sec_last_sample_ops = 0;server.unixtime = time(NULL);server.mstime = mstime();server.lastbgsave_status = REDIS_OK;server.repl_good_slaves_count = 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) {redisPanic("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){redisPanic("Unrecoverable error creating server.ipfd file event.");}}if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");/* Open the AOF file if needed. */if (server.aof_state == REDIS_AOF_ON) {server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);if (server.aof_fd == -1) {redisLog(REDIS_WARNING, "Can't open the append-only file: %s",strerror(errno));exit(1);}}/* 32 bit instances are limited to 4GB of address space, so if there is* no explicit limit in the user provided configuration we set a limit* at 3 GB using maxmemory with 'noeviction' policy'. This avoids* useless crashes of the Redis instance for out of memory. */if (server.arch_bits == 32 && server.maxmemory == 0) {redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");server.maxmemory = 3072LL*(1024*1024); /* 3 GB */server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;}replicationScriptCacheInit();scriptingInit();slowlogInit();bioInit(); }

1.1 信號(hào)處理

signal(SIGHUP, SIG_IGN); signal(SIGPIPE, SIG_IGN);setupSignalHandlers();

signal語(yǔ)法：

#include <signal.h>void (*signal(int sig, void (*func)(int)))(int); int sighold(int sig); int sigignore(int sig); int sigpause(int sig); int sigrelse(int sig); void (*sigset(int sig, void (*disp)(int)))(int);

signal變量定義在signal.h文件中，其中：

1.信號(hào)

Signal	Description
SIGABRT	由調(diào)用abort函數(shù)產(chǎn)生，進(jìn)程非正常退出
SIGALRM	用alarm函數(shù)設(shè)置的timer超時(shí)或setitimer函數(shù)設(shè)置的interval timer超時(shí)
SIGBUS	某種特定的硬件異常，通常由內(nèi)存訪問(wèn)引起
SIGCANCEL	由Solaris Thread Library內(nèi)部使用，通常不會(huì)使用
SIGCHLD	進(jìn)程Terminate或Stop的時(shí)候，SIGCHLD會(huì)發(fā)送給它的父進(jìn)程。缺省情況下該Signal會(huì)被忽略
SIGCONT	當(dāng)被stop的進(jìn)程恢復(fù)運(yùn)行的時(shí)候，自動(dòng)發(fā)送
SIGEMT	和實(shí)現(xiàn)相關(guān)的硬件異常
SIGFPE	數(shù)學(xué)相關(guān)的異常，如被0除，浮點(diǎn)溢出，等等
SIGFREEZE	Solaris專(zhuān)用，Hiberate或者Suspended時(shí)候發(fā)送
SIGHUP	發(fā)送給具有Terminal的Controlling Process，當(dāng)terminal被disconnect時(shí)候發(fā)送
SIGILL	非法指令異常
SIGINFO	BSD signal。由Status Key產(chǎn)生，通常是CTRL+T。發(fā)送給所有Foreground Group的進(jìn)程
SIGINT	由Interrupt Key產(chǎn)生，通常是CTRL+C或者DELETE。發(fā)送給所有ForeGround Group的進(jìn)程
SIGIO	異步IO事件
SIGIOT	實(shí)現(xiàn)相關(guān)的硬件異常，一般對(duì)應(yīng)SIGABRT
SIGKILL	無(wú)法處理和忽略。中止某個(gè)進(jìn)程
SIGLWP	由Solaris Thread Libray內(nèi)部使用
SIGPIPE	在reader中止之后寫(xiě)Pipe的時(shí)候發(fā)送
SIGPOLL	當(dāng)某個(gè)事件發(fā)送給Pollable Device的時(shí)候發(fā)送
SIGPROF	Setitimer指定的Profiling Interval Timer所產(chǎn)生
SIGPWR	和系統(tǒng)相關(guān)。和UPS相關(guān)。
SIGQUIT	輸入Quit Key的時(shí)候（CTRL+/）發(fā)送給所有Foreground Group的進(jìn)程
SIGSEGV	非法內(nèi)存訪問(wèn)
SIGSTKFLT	Linux專(zhuān)用，數(shù)學(xué)協(xié)處理器的棧異常
SIGSTOP	中止進(jìn)程。無(wú)法處理和忽略。
SIGSYS	非法系統(tǒng)調(diào)用
SIGTERM	請(qǐng)求中止進(jìn)程，kill命令缺省發(fā)送
SIGTHAW	Solaris專(zhuān)用，從Suspend恢復(fù)時(shí)候發(fā)送
SIGTRAP	實(shí)現(xiàn)相關(guān)的硬件異常。一般是調(diào)試異常
SIGTSTP	Suspend Key，一般是Ctrl+Z。發(fā)送給所有Foreground Group的進(jìn)程
SIGTTIN	當(dāng)Background Group的進(jìn)程嘗試讀取Terminal的時(shí)候發(fā)送
SIGTTOU	當(dāng)Background Group的進(jìn)程嘗試寫(xiě)Terminal的時(shí)候發(fā)送
SIGURG	當(dāng)out-of-band data接收的時(shí)候可能發(fā)送
SIGUSR1	用戶自定義signal 1
SIGUSR2	用戶自定義signal 2
SIGVTALRM	setitimer函數(shù)設(shè)置的Virtual Interval Timer超時(shí)的時(shí)候
SIGWAITING	Solaris Thread Library內(nèi)部實(shí)現(xiàn)專(zhuān)用
SIGWINCH	當(dāng)Terminal的窗口大小改變的時(shí)候，發(fā)送給Foreground Group的所有進(jìn)程
SIGXCPU	當(dāng)CPU時(shí)間限制超時(shí)的時(shí)候
SIGXFSZ	進(jìn)程超過(guò)文件大小限制
SIGXRES	Solaris專(zhuān)用，進(jìn)程超過(guò)資源限制的時(shí)候發(fā)送

================================================================

????? 2.處理函數(shù)

SIG_DFL

Request for default signal handling.

SIG_ERR

Return value from signal() in case of error.

SIG_HOLD

Request that signal be held.

???????? SIG_IGN

????? Request that signal be ignored.

????? 3. 建立信號(hào)處理函數(shù)

void setupSignalHandlers(void) {struct sigaction act;/* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction is used.* Otherwise, sa_handler is used. */sigemptyset(&act.sa_mask);act.sa_flags = 0;act.sa_handler = sigtermHandler; //信號(hào)中斷處理函數(shù)，使用?sigaction(SIGTERM, &act, NULL);sigaction(SIGTERM, &act, NULL);#ifdef HAVE_BACKTRACEsigemptyset(&act.sa_mask);act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO;act.sa_sigaction = sigsegvHandler;sigaction(SIGSEGV, &act, NULL);sigaction(SIGBUS, &act, NULL);sigaction(SIGFPE, &act, NULL);sigaction(SIGILL, &act, NULL); #endifreturn; }

1.2 創(chuàng)建共享對(duì)象

void createSharedObjects(void) {int j;shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));shared.cnegone = createObject(REDIS_STRING,sdsnew(":-1\r\n"));shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));shared.queued = createObject(REDIS_STRING,sdsnew("+QUEUED\r\n"));shared.emptyscan = createObject(REDIS_STRING,sdsnew("*2\r\n$1\r\n0\r\n*0\r\n"));shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew("-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"));shared.nokeyerr = createObject(REDIS_STRING,sdsnew("-ERR no such key\r\n"));shared.syntaxerr = createObject(REDIS_STRING,sdsnew("-ERR syntax error\r\n"));shared.sameobjecterr = createObject(REDIS_STRING,sdsnew("-ERR source and destination objects are the same\r\n"));shared.outofrangeerr = createObject(REDIS_STRING,sdsnew("-ERR index out of range\r\n"));shared.noscripterr = createObject(REDIS_STRING,sdsnew("-NOSCRIPT No matching script. Please use EVAL.\r\n"));shared.loadingerr = createObject(REDIS_STRING,sdsnew("-LOADING Redis is loading the dataset in memory\r\n"));shared.slowscripterr = createObject(REDIS_STRING,sdsnew("-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n"));shared.masterdownerr = createObject(REDIS_STRING,sdsnew("-MASTERDOWN Link with MASTER is down and slave-serve-stale-data is set to 'no'.\r\n"));shared.bgsaveerr = createObject(REDIS_STRING,sdsnew("-MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist on disk. Commands that may modify the data set are disabled. Please check Redis logs for details about the error.\r\n"));shared.roslaveerr = createObject(REDIS_STRING,sdsnew("-READONLY You can't write against a read only slave.\r\n"));shared.noautherr = createObject(REDIS_STRING,sdsnew("-NOAUTH Authentication required.\r\n"));shared.oomerr = createObject(REDIS_STRING,sdsnew("-OOM command not allowed when used memory > 'maxmemory'.\r\n"));shared.execaborterr = createObject(REDIS_STRING,sdsnew("-EXECABORT Transaction discarded because of previous errors.\r\n"));shared.noreplicaserr = createObject(REDIS_STRING,sdsnew("-NOREPLICAS Not enough good slaves to write.\r\n"));shared.space = createObject(REDIS_STRING,sdsnew(" "));shared.colon = createObject(REDIS_STRING,sdsnew(":"));shared.plus = createObject(REDIS_STRING,sdsnew("+"));for (j = 0; j < REDIS_SHARED_SELECT_CMDS; j++) {char dictid_str[64];int dictid_len;dictid_len = ll2string(dictid_str,sizeof(dictid_str),j);shared.select[j] = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",dictid_len, dictid_str));}shared.messagebulk = createStringObject("$7\r\nmessage\r\n",13);shared.pmessagebulk = createStringObject("$8\r\npmessage\r\n",14);shared.subscribebulk = createStringObject("$9\r\nsubscribe\r\n",15);shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18);shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17);shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19);shared.del = createStringObject("DEL",3);shared.rpop = createStringObject("RPOP",4);shared.lpop = createStringObject("LPOP",4);shared.lpush = createStringObject("LPUSH",5);for (j = 0; j < REDIS_SHARED_INTEGERS; j++) {shared.integers[j] = createObject(REDIS_STRING,(void*)(long)j);shared.integers[j]->encoding = REDIS_ENCODING_INT;}for (j = 0; j < REDIS_SHARED_BULKHDR_LEN; j++) {shared.mbulkhdr[j] = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"*%d\r\n",j));shared.bulkhdr[j] = createObject(REDIS_STRING,sdscatprintf(sdsempty(),"$%d\r\n",j));} }

1.3 改變打開(kāi)的最大文件數(shù)量

/* This function will try to raise the max number of open files accordingly to* the configured max number of clients. It will also account for 32 additional* file descriptors as we need a few more for persistence, listening* sockets, log files and so forth.** If it will not be possible to set the limit accordingly to the configured* max number of clients, the function will do the reverse setting* server.maxclients to the value that we can actually handle. */ void adjustOpenFilesLimit(void) {rlim_t maxfiles = server.maxclients+32;struct rlimit limit;if (getrlimit(RLIMIT_NOFILE,&limit) == -1) {redisLog(REDIS_WARNING,"Unable to obtain the current NOFILE limit (%s), assuming 1024 and setting the max clients configuration accordingly.",strerror(errno));server.maxclients = 1024-32;} else {rlim_t oldlimit = limit.rlim_cur;/* Set the max number of files if the current limit is not enough* for our needs. */if (oldlimit < maxfiles) {rlim_t f;f = maxfiles;while(f > oldlimit) {limit.rlim_cur = f;limit.rlim_max = f;if (setrlimit(RLIMIT_NOFILE,&limit) != -1) break;f -= 128;}if (f < oldlimit) f = oldlimit;if (f != maxfiles) {server.maxclients = f-32;redisLog(REDIS_WARNING,"Unable to set the max number of files limit to %d (%s), setting the max clients configuration to %d.",(int) maxfiles, strerror(errno), (int) server.maxclients);} else {redisLog(REDIS_NOTICE,"Max number of open files set to %d",(int) maxfiles);}}} }

1.4 創(chuàng)建事件處理

aeEventLoop *aeCreateEventLoop(int setsize) {aeEventLoop *eventLoop;int i;if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;eventLoop->setsize = setsize;eventLoop->lastTime = time(NULL);eventLoop->timeEventHead = NULL;eventLoop->timeEventNextId = 0;eventLoop->stop = 0;eventLoop->maxfd = -1;eventLoop->beforesleep = NULL;if (aeApiCreate(eventLoop) == -1) goto err;/* Events with mask == AE_NONE are not set. So let's initialize the* vector with it. */for (i = 0; i < setsize; i++)eventLoop->events[i].mask = AE_NONE;return eventLoop;err:if (eventLoop) {zfree(eventLoop->events);zfree(eventLoop->fired);zfree(eventLoop);}return NULL; }

1.5 綁定監(jiān)聽(tīng)端口

/* Initialize a set of file descriptors to listen to the specified 'port'* binding the addresses specified in the Redis server configuration.** The listening file descriptors are stored in the integer array 'fds'* and their number is set in '*count'.** The addresses to bind are specified in the global server.bindaddr array* and their number is server.bindaddr_count. If the server configuration* contains no specific addresses to bind, this function will try to* bind * (all addresses) for both the IPv4 and IPv6 protocols.** On success the function returns REDIS_OK.** On error the function returns REDIS_ERR. For the function to be on* error, at least one of the server.bindaddr addresses was* impossible to bind, or no bind addresses were specified in the server* configuration but the function is not able to bind * for at least* one of the IPv4 or IPv6 protocols. */ int listenToPort(int port, int *fds, int *count) {int j;/* Force binding of 0.0.0.0 if no bind address is specified, always* entering the loop if j == 0. */if (server.bindaddr_count == 0) server.bindaddr[0] = NULL;for (j = 0; j < server.bindaddr_count || j == 0; j++) {if (server.bindaddr[j] == NULL) {/* Bind * for both IPv6 and IPv4, we enter here only if* server.bindaddr_count == 0. */fds[*count] = anetTcp6Server(server.neterr,port,NULL);if (fds[*count] != ANET_ERR) (*count)++;fds[*count] = anetTcpServer(server.neterr,port,NULL);if (fds[*count] != ANET_ERR) (*count)++;/* Exit the loop if we were able to bind * on IPv4 or IPv6,* otherwise fds[*count] will be ANET_ERR and we'll print an* error and return to the caller with an error. */if (*count) break;} else if (strchr(server.bindaddr[j],':')) {/* Bind IPv6 address. */fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr[j]);} else {/* Bind IPv4 address. */fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr[j]);}if (fds[*count] == ANET_ERR) {redisLog(REDIS_WARNING,"Creating Server TCP listening socket %s:%d: %s",server.bindaddr[j] ? server.bindaddr[j] : "*",server.port, server.neterr);return REDIS_ERR;}(*count)++;}return REDIS_OK; }

1.6 創(chuàng)建redis數(shù)據(jù)庫(kù)，并初始化一些中間狀態(tài)

?/* 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].id = j;
??????? server.db[j].avg_ttl = 0;
??? }
??? server.pubsub_channels = dictCreate(&keylistDictType,NULL);
??? server.pubsub_patterns = listCreate();
??? listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
??? listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
??? server.cronloops = 0;
??? server.rdb_child_pid = -1;
??? server.aof_child_pid = -1;
??? aofRewriteBufferReset();
??? server.aof_buf = sdsempty();
??? server.lastsave = time(NULL); /* At startup we consider the DB saved. */
??? server.lastbgsave_try = 0;??? /* At startup we never tried to BGSAVE. */
??? server.rdb_save_time_last = -1;
??? server.rdb_save_time_start = -1;
??? server.dirty = 0;
??? server.stat_numcommands = 0;
??? server.stat_numconnections = 0;
??? server.stat_expiredkeys = 0;
??? server.stat_evictedkeys = 0;
??? server.stat_starttime = time(NULL);
??? server.stat_keyspace_misses = 0;
??? server.stat_keyspace_hits = 0;
??? server.stat_peak_memory = 0;
??? server.stat_fork_time = 0;
??? server.stat_rejected_conn = 0;
??? server.stat_sync_full = 0;
??? server.stat_sync_partial_ok = 0;
??? server.stat_sync_partial_err = 0;
??? memset(server.ops_sec_samples,0,sizeof(server.ops_sec_samples));
??? server.ops_sec_idx = 0;
??? server.ops_sec_last_sample_time = mstime();
??? server.ops_sec_last_sample_ops = 0;
??? server.unixtime = time(NULL);
??? server.mstime = mstime();
??? server.lastbgsave_status = REDIS_OK;
??? server.repl_good_slaves_count = 0;

?

1.7 創(chuàng)建服務(wù)器時(shí)間事件

/* Create the serverCron() time event, that's our main way to process* background operations. */if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {redisPanic("Can't create the serverCron time event.");exit(1);}----------------------------------------------------------------------------- long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,aeTimeProc *proc, void *clientData,aeEventFinalizerProc *finalizerProc) {long long id = eventLoop->timeEventNextId++;aeTimeEvent *te;te = zmalloc(sizeof(*te));if (te == NULL) return AE_ERR;te->id = id;aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);te->timeProc = proc;te->finalizerProc = finalizerProc;te->clientData = clientData;te->next = eventLoop->timeEventHead;eventLoop->timeEventHead = te;return id; }

補(bǔ)充：servercron

/* This is our timer interrupt, called server.hz times per second.* Here is where we do a number of things that need to be done asynchronously.* For instance:** - Active expired keys collection (it is also performed in a lazy way on* lookup).* - Software watchdog.* - Update some statistic.* - Incremental rehashing of the DBs hash tables.* - Triggering BGSAVE / AOF rewrite, and handling of terminated children.* - Clients timeout of different kinds.* - Replication reconnection.* - Many more...** Everything directly called here will be called server.hz times per second,* so in order to throttle execution of things we want to do less frequently* a macro is used: run_with_period(milliseconds) { .... }*/int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {int j;REDIS_NOTUSED(eventLoop);REDIS_NOTUSED(id);REDIS_NOTUSED(clientData);/* Software watchdog: deliver the SIGALRM that will reach the signal* handler if we don't return here fast enough. */if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);/* We take a cached value of the unix time in the global state because* with virtual memory and aging there is to store the current time* in objects at every object access, and accuracy is not needed.* To access a global var is faster than calling time(NULL) */server.unixtime = time(NULL);server.mstime = mstime();run_with_period(100) trackOperationsPerSecond();/* We have just 22 bits per object for LRU information.* So we use an (eventually wrapping) LRU clock with 10 seconds resolution.* 2^22 bits with 10 seconds resolution is more or less 1.5 years.** Note that even if this will wrap after 1.5 years it's not a problem,* everything will still work but just some object will appear younger* to Redis. But for this to happen a given object should never be touched* for 1.5 years.** Note that you can change the resolution altering the* REDIS_LRU_CLOCK_RESOLUTION define.*/updateLRUClock();/* Record the max memory used since the server was started. */if (zmalloc_used_memory() > server.stat_peak_memory)server.stat_peak_memory = zmalloc_used_memory();/* We received a SIGTERM, shutting down here in a safe way, as it is* not ok doing so inside the signal handler. */if (server.shutdown_asap) {if (prepareForShutdown(0) == REDIS_OK) exit(0);redisLog(REDIS_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");server.shutdown_asap = 0;}/* Show some info about non-empty databases */run_with_period(5000) {for (j = 0; j < server.dbnum; j++) {long long size, used, vkeys;size = dictSlots(server.db[j].dict);used = dictSize(server.db[j].dict);vkeys = dictSize(server.db[j].expires);if (used || vkeys) {redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);/* dictPrintStats(server.dict); */}}}/* Show information about connected clients */if (!server.sentinel_mode) {run_with_period(5000) {redisLog(REDIS_VERBOSE,"%lu clients connected (%lu slaves), %zu bytes in use",listLength(server.clients)-listLength(server.slaves),listLength(server.slaves),zmalloc_used_memory());}}/* We need to do a few operations on clients asynchronously. */clientsCron();/* Handle background operations on Redis databases. */databasesCron();/* Start a scheduled AOF rewrite if this was requested by the user while* a BGSAVE was in progress. */if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&server.aof_rewrite_scheduled){rewriteAppendOnlyFileBackground();}/* Check if a background saving or AOF rewrite in progress terminated. */if (server.rdb_child_pid != -1 || server.aof_child_pid != -1) {int statloc;pid_t pid;if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {int exitcode = WEXITSTATUS(statloc);int bysignal = 0;if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);if (pid == server.rdb_child_pid) {backgroundSaveDoneHandler(exitcode,bysignal);} else if (pid == server.aof_child_pid) {backgroundRewriteDoneHandler(exitcode,bysignal);} else {redisLog(REDIS_WARNING,"Warning, detected child with unmatched pid: %ld",(long)pid);}updateDictResizePolicy();}} else {/* If there is not a background saving/rewrite in progress check if* we have to save/rewrite now */for (j = 0; j < server.saveparamslen; j++) {struct saveparam *sp = server.saveparams+j;/* Save if we reached the given amount of changes,* the given amount of seconds, and if the latest bgsave was* successful or if, in case of an error, at least* REDIS_BGSAVE_RETRY_DELAY seconds already elapsed. */if (server.dirty >= sp->changes &&server.unixtime-server.lastsave > sp->seconds &&(server.unixtime-server.lastbgsave_try >REDIS_BGSAVE_RETRY_DELAY ||server.lastbgsave_status == REDIS_OK)){redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",sp->changes, (int)sp->seconds);rdbSaveBackground(server.rdb_filename);break;}}/* Trigger an AOF rewrite if needed */if (server.rdb_child_pid == -1 &&server.aof_child_pid == -1 &&server.aof_rewrite_perc &&server.aof_current_size > server.aof_rewrite_min_size){long long base = server.aof_rewrite_base_size ?server.aof_rewrite_base_size : 1;long long growth = (server.aof_current_size*100/base) - 100;if (growth >= server.aof_rewrite_perc) {redisLog(REDIS_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);rewriteAppendOnlyFileBackground();}}}/* If we postponed an AOF buffer flush, let's try to do it every time the* cron function is called. */if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);/* Close clients that need to be closed asynchronous */freeClientsInAsyncFreeQueue();/* Replication cron function -- used to reconnect to master and* to detect transfer failures. */run_with_period(1000) replicationCron();/* Run the sentinel timer if we are in sentinel mode. */run_with_period(100) {if (server.sentinel_mode) sentinelTimer();}server.cronloops++;return 1000/server.hz; }

1.8 創(chuàng)建事件處理

??? /* 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)
??????????? {
??????????????? redisPanic(
??????????????????? "Unrecoverable error creating server.ipfd file event.");
??????????? }
??? }
??? if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
??????? acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");

1.9 打開(kāi)aof文件

/* Open the AOF file if needed. */if (server.aof_state == REDIS_AOF_ON) {server.aof_fd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);if (server.aof_fd == -1) {redisLog(REDIS_WARNING, "Can't open the append-only file: %s",strerror(errno));exit(1);}}

1.10 限制32位機(jī)器內(nèi)存，防止crashed

/* 32 bit instances are limited to 4GB of address space, so if there is* no explicit limit in the user provided configuration we set a limit* at 3 GB using maxmemory with 'noeviction' policy'. This avoids* useless crashes of the Redis instance for out of memory. */if (server.arch_bits == 32 && server.maxmemory == 0) {redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");server.maxmemory = 3072LL*(1024*1024); /* 3 GB */server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;}

1.11 復(fù)制初始化

/* ----------------------- REPLICATION SCRIPT CACHE --------------------------* The goal of this code is to keep track of scripts already sent to every* connected slave, in order to be able to replicate EVALSHA as it is without* translating it to EVAL every time it is possible.** We use a capped collection implemented by an hash table for fast lookup* of scripts we can send as EVALSHA, plus a linked list that is used for* eviction of the oldest entry when the max number of items is reached.** We don't care about taking a different cache for every different slave* since to fill the cache again is not very costly, the goal of this code* is to avoid that the same big script is trasmitted a big number of times* per second wasting bandwidth and processor speed, but it is not a problem* if we need to rebuild the cache from scratch from time to time, every used* script will need to be transmitted a single time to reappear in the cache.** This is how the system works:** 1) Every time a new slave connects, we flush the whole script cache.* 2) We only send as EVALSHA what was sent to the master as EVALSHA, without* trying to convert EVAL into EVALSHA specifically for slaves.* 3) Every time we trasmit a script as EVAL to the slaves, we also add the* corresponding SHA1 of the script into the cache as we are sure every* slave knows about the script starting from now.* 4) On SCRIPT FLUSH command, we replicate the command to all the slaves* and at the same time flush the script cache.* 5) When the last slave disconnects, flush the cache.* 6) We handle SCRIPT LOAD as well since that's how scripts are loaded* in the master sometimes.*//* Initialize the script cache, only called at startup. */ void replicationScriptCacheInit(void) {server.repl_scriptcache_size = 10000;server.repl_scriptcache_dict = dictCreate(&replScriptCacheDictType,NULL);server.repl_scriptcache_fifo = listCreate(); }

1.12 初始化lua腳本環(huán)境

/* Initialize the scripting environment.* It is possible to call this function to reset the scripting environment* assuming that we call scriptingRelease() before.* See scriptingReset() for more information. */ void scriptingInit(void) {lua_State *lua = lua_open();luaLoadLibraries(lua);luaRemoveUnsupportedFunctions(lua);/* Initialize a dictionary we use to map SHAs to scripts.* This is useful for replication, as we need to replicate EVALSHA* as EVAL, so we need to remember the associated script. */server.lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);/* Register the redis commands table and fields */lua_newtable(lua);/* redis.call */lua_pushstring(lua,"call");lua_pushcfunction(lua,luaRedisCallCommand);lua_settable(lua,-3);/* redis.pcall */lua_pushstring(lua,"pcall");lua_pushcfunction(lua,luaRedisPCallCommand);lua_settable(lua,-3);/* redis.log and log levels. */lua_pushstring(lua,"log");lua_pushcfunction(lua,luaLogCommand);lua_settable(lua,-3);lua_pushstring(lua,"LOG_DEBUG");lua_pushnumber(lua,REDIS_DEBUG);lua_settable(lua,-3);lua_pushstring(lua,"LOG_VERBOSE");lua_pushnumber(lua,REDIS_VERBOSE);lua_settable(lua,-3);lua_pushstring(lua,"LOG_NOTICE");lua_pushnumber(lua,REDIS_NOTICE);lua_settable(lua,-3);lua_pushstring(lua,"LOG_WARNING");lua_pushnumber(lua,REDIS_WARNING);lua_settable(lua,-3);/* redis.sha1hex */lua_pushstring(lua, "sha1hex");lua_pushcfunction(lua, luaRedisSha1hexCommand);lua_settable(lua, -3);/* redis.error_reply and redis.status_reply */lua_pushstring(lua, "error_reply");lua_pushcfunction(lua, luaRedisErrorReplyCommand);lua_settable(lua, -3);lua_pushstring(lua, "status_reply");lua_pushcfunction(lua, luaRedisStatusReplyCommand);lua_settable(lua, -3);/* Finally set the table as 'redis' global var. */lua_setglobal(lua,"redis");/* Replace math.random and math.randomseed with our implementations. */lua_getglobal(lua,"math");lua_pushstring(lua,"random");lua_pushcfunction(lua,redis_math_random);lua_settable(lua,-3);lua_pushstring(lua,"randomseed");lua_pushcfunction(lua,redis_math_randomseed);lua_settable(lua,-3);lua_setglobal(lua,"math");/* Add a helper function that we use to sort the multi bulk output of non* deterministic commands, when containing 'false' elements. */{char *compare_func = "function __redis__compare_helper(a,b)\n"" if a == false then a = '' end\n"" if b == false then b = '' end\n"" return a<b\n""end\n";luaL_loadbuffer(lua,compare_func,strlen(compare_func),"@cmp_func_def");lua_pcall(lua,0,0,0);}/* Add a helper function we use for pcall error reporting.* Note that when the error is in the C function we want to report the* information about the caller, that's what makes sense from the point* of view of the user debugging a script. */{char *errh_func = "function __redis__err__handler(err)\n"" local i = debug.getinfo(2,'nSl')\n"" if i and i.what == 'C' then\n"" i = debug.getinfo(3,'nSl')\n"" end\n"" if i then\n"" return i.source .. ':' .. i.currentline .. ': ' .. err\n"" else\n"" return err\n"" end\n""end\n";luaL_loadbuffer(lua,errh_func,strlen(errh_func),"@err_handler_def");lua_pcall(lua,0,0,0);}/* Create the (non connected) client that we use to execute Redis commands* inside the Lua interpreter.* Note: there is no need to create it again when this function is called* by scriptingReset(). */if (server.lua_client == NULL) {server.lua_client = createClient(-1);server.lua_client->flags |= REDIS_LUA_CLIENT;}/* Lua beginners ofter don't use "local", this is likely to introduce* subtle bugs in their code. To prevent problems we protect accesses* to global variables. */scriptingEnableGlobalsProtection(lua);server.lua = lua; }

1.13 初始化慢日志

/* Initialize the slow log. This function should be called a single time* at server startup. */ void slowlogInit(void) {server.slowlog = listCreate();server.slowlog_entry_id = 0;listSetFreeMethod(server.slowlog,slowlogFreeEntry); }

1.14 初始化bio

/* Initialize the background system, spawning the thread. */ void bioInit(void) {pthread_attr_t attr;pthread_t thread;size_t stacksize;int j;/* Initialization of state vars and objects */for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {pthread_mutex_init(&bio_mutex[j],NULL);pthread_cond_init(&bio_condvar[j],NULL);bio_jobs[j] = listCreate();bio_pending[j] = 0;}/* Set the stack size as by default it may be small in some system */pthread_attr_init(&attr);pthread_attr_getstacksize(&attr,&stacksize);if (!stacksize) stacksize = 1; /* The world is full of Solaris Fixes */while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;pthread_attr_setstacksize(&attr, stacksize);/* Ready to spawn our threads. We use the single argument the thread* function accepts in order to pass the job ID the thread is* responsible of. */for (j = 0; j < REDIS_BIO_NUM_OPS; j++) {void *arg = (void*)(unsigned long) j;if (pthread_create(&thread,&attr,bioProcessBackgroundJobs,arg) != 0) {redisLog(REDIS_WARNING,"Fatal: Can't initialize Background Jobs.");exit(1);}bio_threads[j] = thread;} }

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

?

轉(zhuǎn)載于:https://www.cnblogs.com/davidwang456/p/3519929.html

總結(jié)

以上是生活随笔為你收集整理的深入redis内部--初始化服务器的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。