客戶端

數(shù)據(jù)結(jié)構(gòu)

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061

/* With multiplexing we need to take per-client state. * Clients are taken in a linked list. */typedef struct client { uint64_t id; /* Client incremental unique ID. */ int fd; /* Client socket. */ redisDb *db; /* Pointer to currently SELECTed DB. */ robj *name; /* As set by CLIENT SETNAME. */ sds querybuf; /* Buffer we use to accumulate client queries. */ size_t qb_pos; /* The position we have read in querybuf. */ sds pending_querybuf; /* If this client is flagged as master, this buffer represents the yet not applied portion of the replication stream that we are receiving from the master. */ size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size. */ int argc; /* Num of arguments of current command. */ robj **argv; /* Arguments of current command. */ struct redisCommand *cmd, *lastcmd; /* Last command executed. */ int reqtype; /* Request protocol type: PROTO_REQ_* */ int multibulklen; /* Number of multi bulk arguments left to read. */ long bulklen; /* Length of bulk argument in multi bulk request. */ list *reply; /* List of reply objects to send to the client. */ unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */ size_t sentlen; /* Amount of bytes already sent in the current buffer or object being sent. */ time_t ctime; /* Client creation time. */ time_t lastinteraction; /* Time of the last interaction, used for timeout */ time_t obuf_soft_limit_reached_time; int flags; /* Client flags: CLIENT_* macros. */ int authenticated; /* When requirepass is non-NULL. */ int replstate; /* Replication state if this is a slave. */ int repl_put_online_on_ack; /* Install slave write handler on ACK. */ int repldbfd; /* Replication DB file descriptor. */ off_t repldboff; /* Replication DB file offset. */ off_t repldbsize; /* Replication DB file size. */ sds replpreamble; /* Replication DB preamble. */ long long read_reploff; /* Read replication offset if this is a master. */ long long reploff; /* Applied replication offset if this is a master. */ long long repl_ack_off; /* Replication ack offset, if this is a slave. */ long long repl_ack_time;/* Replication ack time, if this is a slave. */ long long psync_initial_offset; /* FULLRESYNC reply offset other slaves copying this slave output buffer should use. */ char replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */ int slave_listening_port; /* As configured with: SLAVECONF listening-port */ char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */ int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */ multiState mstate; /* MULTI/EXEC state */ int btype; /* Type of blocking op if CLIENT_BLOCKED. */ blockingState bpop; /* blocking state */ long long woff; /* Last write global replication offset. */ list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */ dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */ list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */ sds peerid; /* Cached peer ID. */ listNode *client_list_node; /* list node in client list */ /* Response buffer */ int bufpos; char buf[PROTO_REPLY_CHUNK_BYTES];} client;

第一眼看上去，發(fā)現(xiàn)，我擦一個(gè)客戶端數(shù)據(jù)結(jié)構(gòu)是真的多啊。這讓我雜記啊，沒(méi)事。不用全部都記住。記幾個(gè)重要的就行。

配置文件

客戶端命令

12345678910111213

"id -- Return the ID of the current connection.","getname -- Return the name of the current connection.","kill -- Kill connection made from .","kill [option value ...] -- Kill connections. Options are:"," addr -- Kill connection made from "," type (normal|master|replica|pubsub) -- Kill connections by type."," skipme (yes|no) -- Skip killing current connection (default: yes).","list [options ...] -- Return information about client connections. Options:"," type (normal|master|replica|pubsub) -- Return clients of specified type.","pause -- Suspend all Redis clients for milliseconds.","reply (on|off|skip) -- Control the replies sent to the current connection.","setname -- Assign the name to the current connection.","unblock [TIMEOUT|ERROR] -- Unblock the specified blocked client."

client list->查看客戶端的信息

其中

• flags表示客戶端類型N-表示普通客戶端，M-表示master等等

1234567891011121314151617

int getClientTypeByName(char *name) { if (!strcasecmp(name,"normal")) return CLIENT_TYPE_NORMAL; else if (!strcasecmp(name,"slave")) return CLIENT_TYPE_SLAVE; else if (!strcasecmp(name,"replica")) return CLIENT_TYPE_SLAVE; else if (!strcasecmp(name,"pubsub")) return CLIENT_TYPE_PUBSUB; else if (!strcasecmp(name,"master")) return CLIENT_TYPE_MASTER; else return -1;}char *getClientTypeName(int class) { switch(class) { case CLIENT_TYPE_NORMAL: return "normal"; case CLIENT_TYPE_SLAVE: return "slave"; case CLIENT_TYPE_PUBSUB: return "pubsub"; case CLIENT_TYPE_MASTER: return "master"; default: return NULL; }}

• obl代表固定緩沖區(qū)的長(zhǎng)度

• oll代表動(dòng)態(tài)緩沖區(qū)列表的長(zhǎng)度

• omem代表使用的字節(jié)數(shù)

• events表示事件類型(r/w)

• cmd記錄最后一次執(zhí)行的命令

具體信息見(jiàn)函數(shù)

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253

/* Concatenate a string representing the state of a client in an human * readable format, into the sds string 's'. */sds catClientInfoString(sds s, client *client) { char flags[16], events[3], *p; int emask; p = flags; if (client->flags & CLIENT_SLAVE) { if (client->flags & CLIENT_MONITOR) *p++ = 'O'; else *p++ = 'S'; } if (client->flags & CLIENT_MASTER) *p++ = 'M'; if (client->flags & CLIENT_PUBSUB) *p++ = 'P'; if (client->flags & CLIENT_MULTI) *p++ = 'x'; if (client->flags & CLIENT_BLOCKED) *p++ = 'b'; if (client->flags & CLIENT_DIRTY_CAS) *p++ = 'd'; if (client->flags & CLIENT_CLOSE_AFTER_REPLY) *p++ = 'c'; if (client->flags & CLIENT_UNBLOCKED) *p++ = 'u'; if (client->flags & CLIENT_CLOSE_ASAP) *p++ = 'A'; if (client->flags & CLIENT_UNIX_SOCKET) *p++ = 'U'; if (client->flags & CLIENT_READONLY) *p++ = 'r'; if (p == flags) *p++ = 'N'; *p++ = '\0'; emask = client->fd == -1 ? 0 : aeGetFileEvents(server.el,client->fd); p = events; if (emask & AE_READABLE) *p++ = 'r'; if (emask & AE_WRITABLE) *p++ = 'w'; *p = '\0'; return sdscatfmt(s, "id=%U addr=%s fd=%i name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U obl=%U oll=%U omem=%U events=%s cmd=%s", (unsigned long long) client->id, getClientPeerId(client), client->fd, client->name ? (char*)client->name->ptr : "", (long long)(server.unixtime - client->ctime), (long long)(server.unixtime - client->lastinteraction), flags, client->db->id, (int) dictSize(client->pubsub_channels), (int) listLength(client->pubsub_patterns), (client->flags & CLIENT_MULTI) ? client->mstate.count : -1, (unsigned long long) sdslen(client->querybuf), (unsigned long long) sdsavail(client->querybuf), (unsigned long long) client->bufpos, (unsigned long long) listLength(client->reply), (unsigned long long) getClientOutputBufferMemoryUsage(client), events, client->lastcmd ? client->lastcmd->name : "NULL");}

info clients–>查看所有客戶端

• connected_clients：代表當(dāng)前Redis節(jié)點(diǎn)的客戶端連接數(shù)，需要重點(diǎn)監(jiān)控，一旦超過(guò)maxclients，新的客戶端連接將被拒絕。

• client_longest_output_list：當(dāng)前所有輸出緩沖區(qū)中隊(duì)列對(duì)象個(gè)數(shù)的最大值。

• client_biggest_input_buf：當(dāng)前所有輸入緩沖區(qū)中占用的最大容量。

• blocked_clients：正在執(zhí)行阻塞命令(例如blpop、brpop、brpoplpush)的客戶端個(gè)數(shù)

客戶端關(guān)閉

調(diào)用client kill命令

不符合協(xié)議格式的命令

客戶端超時(shí)

輸入緩沖區(qū)超過(guò)閾值1G

輸出緩沖區(qū)超出閾值(軟限制和硬限制，超過(guò)硬限制客戶端立即關(guān)閉，超過(guò)軟限制且持續(xù)指定秒后再關(guān)閉)

格式為client-output-buffer-limit

client-output-buffer-limit normal 0 0 0

client-output-buffer-limit replica 256mb 64mb 60

client-output-buffer-limit pubsub 32mb 8mb 60

其中class表示客戶端類型，超過(guò)hard limit客戶端立即關(guān)閉，超過(guò)soft limit且持續(xù)soft seconds后再關(guān)閉。

class取值：

normal -> normal clients including MONITOR clients

slave -> slave clients

pubsub -> clients subscribed to at least one pubsub channel or pattern

注意點(diǎn)：對(duì)于client的檢測(cè)是在serverCron定時(shí)函數(shù)中進(jìn)行。

12345678910111213141516171819202122232425262728293031323334353637

#define CLIENTS_CRON_MIN_ITERATIONS 5void clientsCron(void) { /* Try to process at least numclients/server.hz of clients * per call. Since normally (if there are no big latency events) this * function is called server.hz times per second, in the average case we * process all the clients in 1 second. */ int numclients = listLength(server.clients); //看看這里，如果客戶端很多，這個(gè)數(shù)字將很大，所以這里做了優(yōu)化，看下面的分析 int iterations = numclients/server.hz; mstime_t now = mstime(); /* Process at least a few clients while we are at it, even if we need * to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract * of processing each client once per second. */ if (iterations < CLIENTS_CRON_MIN_ITERATIONS) iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ? numclients : CLIENTS_CRON_MIN_ITERATIONS; while(listLength(server.clients) && iterations--) { client *c; listNode *head; /* Rotate the list, take the current head, process. * This way if the client must be removed from the list it's the * first element and we don't incur into O(N) computation. */ listRotate(server.clients); head = listFirst(server.clients); c = listNodeValue(head); /* The following functions do different service checks on the client. * The protocol is that they return non-zero if the client was * terminated. */ if (clientsCronHandleTimeout(c,now)) continue; if (clientsCronResizeQueryBuffer(c)) continue; if (clientsCronTrackExpansiveClients(c)) continue; }}

由于redis是單線程的，所以他不可能一直循環(huán)來(lái)檢測(cè)客戶端。其中客戶端超時(shí)檢測(cè)是在serverCron的clientsCron中進(jìn)行，serverCron是一個(gè)周期函數(shù)，每100ms執(zhí)行一次。server.hz表示serverCron函數(shù)的調(diào)用頻率，默認(rèn)為10。clientsCron函數(shù)中為了每秒鐘都能循環(huán)一次所有客戶端，所以每次循環(huán)次數(shù)為iterations = numclients/server.hz。如果客戶端多的話，可能會(huì)導(dǎo)致redis主線程的阻塞。因此，5.0引入了動(dòng)態(tài)hz，見(jiàn)配置文件dynamic-hz，默認(rèn)打開(kāi)。

服務(wù)端

數(shù)據(jù)庫(kù)

數(shù)據(jù)庫(kù)存儲(chǔ)在redisDb結(jié)構(gòu)中，而在服務(wù)端redisServer結(jié)構(gòu)中保存著redisDb對(duì)象和個(gè)數(shù)，個(gè)數(shù)可以在配置文件中進(jìn)行更新(見(jiàn)配置文件中的dbnum)

數(shù)據(jù)結(jié)構(gòu)

12345678910

typedef struct redisDb { dict *dict; /* The keyspace for this DB */ dict *expires; /* Timeout of keys with a timeout set */ dict *blocking_keys; /* Keys with clients waiting for data (BLPOP)*/ dict *ready_keys; /* Blocked keys that received a PUSH */ dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */ int id; /* Database ID */ long long avg_ttl; /* Average TTL, just for stats */ list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */} redisDb;

而redisServer中都包含redisDb數(shù)據(jù)結(jié)構(gòu)。表示當(dāng)前使用的是哪個(gè)db。

12345678910111213141516

struct redisServer { /* General */ pid_t pid; /* Main process pid. */ char *configfile; /* Absolute config file path, or NULL */ char *executable; /* Absolute executable file path. */ char **exec_argv; /* Executable argv vector (copy). */ int dynamic_hz; /* Change hz value depending on # of clients. */ int config_hz; /* Configured HZ value. May be different than the actual 'hz' field value if dynamic-hz is enabled. */ int hz; /* serverCron() calls frequency in hertz */ //看這里 redisDb *db; ...}

數(shù)據(jù)庫(kù)切換

而數(shù)據(jù)庫(kù)的切換是使用select來(lái)執(zhí)行的。當(dāng)執(zhí)行select命令時(shí)，實(shí)際上是將第n個(gè)db指向客戶端的db對(duì)象，c->db = &server.db[id]；

123456

int selectDb(client *c, int id) { if (id < 0 || id >= server.dbnum) return C_ERR; c->db = &server.db[id]; return C_OK;}

鍵空間

由前文可知redis可以有多個(gè)數(shù)據(jù)庫(kù)，而不同數(shù)據(jù)庫(kù)之間是互不影響的，那如何做到的呢？鍵空間可以理解成C++里面的名稱空間，用來(lái)隔離。數(shù)據(jù)存儲(chǔ)在redisDb中的dict對(duì)象。因此對(duì)鍵的操作，基本都是基于鍵空間來(lái)操作的。

本質(zhì)其實(shí)就是每個(gè)redisDd中的dict對(duì)象。很好理解，因?yàn)檫x擇了不同的db，那肯定這個(gè)對(duì)象下面的數(shù)據(jù)也不一樣。

12345678910

void setKey(redisDb *db, robj *key, robj *val) { if (lookupKeyWrite(db,key) == NULL) { dbAdd(db,key,val); } else { dbOverwrite(db,key,val); } incrRefCount(val); removeExpire(db,key); signalModifiedKey(db,key);}

過(guò)期鍵

在redis中過(guò)期鍵保存在redisDb中expires變量里，expires是dict指針類型。

儲(chǔ)存方式

• Key保存的是數(shù)據(jù)庫(kù)的鍵對(duì)象

• Value保存的是數(shù)據(jù)庫(kù)鍵對(duì)象的過(guò)期時(shí)間，長(zhǎng)整型Unix時(shí)間。

12345

de = dictAddOrFind(db->expires,dictGetKey(kde)); dictSetSignedIntegerVal(de,when);#define dictSetSignedIntegerVal(entry, _val_) \ do { (entry)->v.s64 = _val_; } while(0)

設(shè)置過(guò)期時(shí)間

相對(duì)方式：expire,單位秒；pexpire,單位毫秒

絕對(duì)方式：expireat,單位秒；pexpireat< timestamp >,單位毫秒。其他命令都會(huì)轉(zhuǎn)成pexpireat

如expire命令：setExpire(c,c->db,key,when)—->de = dictFind(db->dict,key->ptr)—->dictAddOrFind(db->expires,dictGetKey(de))

刪除過(guò)期時(shí)間

persist?，具體函數(shù)為removeExpire(c->db,c->argv[1])—->dictDelete(db->expires,key->ptr)

查看鍵剩余時(shí)間

查看過(guò)期時(shí)間：

ttl?，秒單位；pttl，毫秒

lookupKeyReadWithFlags—->getExpire—->ttl = expire-mstime();

鍵過(guò)期策略

redis中采用的是定期刪除和惰性刪除兩種

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163

void activeExpireCycle(int type) { /* This function has some global state in order to continue the work * incrementally across calls. */ static unsigned int current_db = 0; /* Last DB tested. */ //timelimit_exit，超時(shí)退出 static int timelimit_exit = 0; /* Time limit hit in previous call? */ static long long last_fast_cycle = 0; /*上一次執(zhí)行快速定期刪除的時(shí)間點(diǎn) When last fast cycle ran. */ int j, iteration = 0; int dbs_per_call = CRON_DBS_PER_CALL; long long start = ustime(), timelimit, elapsed; /* When clients are paused the dataset should be static not just from the * POV of clients not being able to write, but also from the POV of * expires and evictions of keys not being performed. */ if (clientsArePaused()) return; if (type == ACTIVE_EXPIRE_CYCLE_FAST) { /* Don't start a fast cycle if the previous cycle did not exit * for time limit. Also don't repeat a fast cycle for the same period * as the fast cycle total duration itself. */ if (!timelimit_exit) return; if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return; last_fast_cycle = start; } /* We usually should test CRON_DBS_PER_CALL per iteration, with * two exceptions: * * 1) Don't test more DBs than we have. * 2) If last time we hit the time limit, we want to scan all DBs * in this iteration, as there is work to do in some DB and we don't want * expired keys to use memory for too much time. */ if (dbs_per_call > server.dbnum || timelimit_exit) dbs_per_call = server.dbnum; /* We can use at max ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC percentage of CPU time * per iteration. Since this function gets called with a frequency of * server.hz times per second, the following is the max amount of * microseconds we can spend in this function. */ // 最多允許25%的CPU時(shí)間用于過(guò)期Key清理 // 若hz=1，則一次activeExpireCycle最多只能執(zhí)行250ms // 若hz=10，則一次activeExpireCycle最多只能執(zhí)行25ms timelimit = 1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100; timelimit_exit = 0; if (timelimit <= 0) timelimit = 1; //如果是ACTIVE_EXPIRE_CYCLE_FAST，時(shí)間限制為ACTIVE_EXPIRE_CYCLE_FAST_DURATION if (type == ACTIVE_EXPIRE_CYCLE_FAST) timelimit = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; /* in microseconds. */ /* Accumulate some global stats as we expire keys, to have some idea * about the number of keys that are already logically expired, but still * existing inside the database. */ long total_sampled = 0; long total_expired = 0; //timelimit_exit超時(shí)的話，我們也不會(huì)循環(huán)遍歷下一個(gè) for (j = 0; j < dbs_per_call && timelimit_exit == 0; j++) { int expired; redisDb *db = server.db+(current_db % server.dbnum); /* Increment the DB now so we are sure if we run out of time * in the current DB we'll restart from the next. This allows to * distribute the time evenly across DBs. */ current_db++; /* Continue to expire if at the end of the cycle more than 25% * of the keys were expired. */ do { unsigned long num, slots; long long now, ttl_sum; int ttl_samples; iteration++; /* If there is nothing to expire try next DB ASAP. */ if ((num = dictSize(db->expires)) == 0) { db->avg_ttl = 0; break; } slots = dictSlots(db->expires); now = mstime(); /* When there are less than 1% filled slots getting random * keys is expensive, so stop here waiting for better times... * The dictionary will be resized asap. */ if (num && slots > DICT_HT_INITIAL_SIZE && (num*100/slots < 1)) break; /* The main collection cycle. Sample random keys among keys * with an expire set, checking for expired ones. */ expired = 0; ttl_sum = 0; ttl_samples = 0; // 一次取ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP個(gè)Key，判斷是否過(guò)期 if (num > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP) num = ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP; while (num--) { dictEntry *de; long long ttl; //隨機(jī)從過(guò)期鍵表中抽出一個(gè)鍵，并獲取他的ttl if ((de = dictGetRandomKey(db->expires)) == NULL) break; ttl = dictGetSignedIntegerVal(de)-now; //判斷他是否過(guò)期，如果過(guò)期就對(duì)其進(jìn)行刪除操作，至于是同步還是異步，那就隨便你 if (activeExpireCycleTryExpire(db,de,now)) expired++; if (ttl > 0) { /* We want the average TTL of keys yet not expired. */ ttl_sum += ttl; ttl_samples++; } total_sampled++; } total_expired += expired; /* Update the average TTL stats for this database. */ if (ttl_samples) { long long avg_ttl = ttl_sum/ttl_samples; /* Do a simple running average with a few samples. * We just use the current estimate with a weight of 2% * and the previous estimate with a weight of 98%. */ if (db->avg_ttl == 0) db->avg_ttl = avg_ttl; db->avg_ttl = (db->avg_ttl/50)*49 + (avg_ttl/50); } /* We can't block forever here even if there are many keys to * expire. So after a given amount of milliseconds return to the * caller waiting for the other active expire cycle. * * 每迭代16次就來(lái)計(jì)算函數(shù)已經(jīng)運(yùn)行的時(shí)間，如果這個(gè)時(shí)間超過(guò)了之前的限定時(shí)間timelimit， * 就將timelimit_exit這個(gè)標(biāo)志置為1，說(shuō)明程序超時(shí)，需要強(qiáng)制退出了 * */ if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */ elapsed = ustime()-start; if (elapsed > timelimit) { timelimit_exit = 1; server.stat_expired_time_cap_reached_count++; break; } } /* We don't repeat the cycle if there are less than 25% of keys * found expired in the current DB. */ /** * ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP是我們每個(gè)循環(huán)希望查到的過(guò)期鍵的個(gè)數(shù)，如果我們每次循環(huán)過(guò)后，被清理的過(guò)期鍵的個(gè)數(shù)超過(guò)了我們期望的四分之一，我們就會(huì)繼續(xù)這個(gè)循環(huán)，因?yàn)檫@說(shuō)明當(dāng)前數(shù)據(jù)庫(kù)中過(guò)期鍵的個(gè)數(shù)比較多 * ，需要繼續(xù)清理，如果沒(méi)有達(dá)到我們期望的四分之一，就跳出while循環(huán)，遍歷下一個(gè)數(shù)據(jù)庫(kù) * */ } while (expired > ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4); } elapsed = ustime()-start; latencyAddSampleIfNeeded("expire-cycle",elapsed/1000); /* Update our estimate of keys existing but yet to be expired. * Running average with this sample accounting for 5%. */ double current_perc; if (total_sampled) { current_perc = (double)total_expired/total_sampled; } else current_perc = 0; server.stat_expired_stale_perc = (current_perc*0.05)+ (server.stat_expired_stale_perc*0.95);}

• 定期刪除分為兩種模式，分別是在不同場(chǎng)景下調(diào)用的。因此結(jié)束判斷不一樣。

  在定時(shí)任務(wù)serverCron中的databasesCron中，為ACTIVE_EXPIRE_CYCLE_SLOW，這就意味著我們可以花費(fèi)更多的時(shí)間來(lái)處理。而在事件循環(huán)前的beforeSleep函數(shù)中則為ACTIVE_EXPIRE_CYCLE_FAST，因?yàn)椴荒苡绊懱幚硎录Ｒ虼?#xff0c;還做了一個(gè)優(yōu)化

  12345678

  if (type == ACTIVE_EXPIRE_CYCLE_FAST) { /* Don't start a fast cycle if the previous cycle did not exit * for time limit. Also don't repeat a fast cycle for the same period * as the fast cycle total duration itself. */ if (!timelimit_exit) return; if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return; last_fast_cycle = start; }

• REDIS_EXPIRELOOKUPS_TIME_PERC是單位時(shí)間內(nèi)分配給activeExpireCycle函數(shù)執(zhí)行的CPU比例，默認(rèn)為25.

• 每次循環(huán)最多16個(gè)庫(kù)。

• 每個(gè)庫(kù)要求找到過(guò)期鍵達(dá)到5個(gè)就行(需要注意的是，因?yàn)槊總€(gè)庫(kù)是隨機(jī)選取一個(gè)key的，所以數(shù)據(jù)量不能太少，太少隨機(jī)效果不好)。

• 在每個(gè)庫(kù)里面，每次隨機(jī)選取20個(gè)key。檢查是否是過(guò)期鍵，過(guò)期就計(jì)數(shù)一次。每隔16次檢查一下時(shí)間是否超過(guò)限制，如果超過(guò)需要退出循環(huán)。不在繼續(xù)查找。

對(duì)于過(guò)期鍵對(duì)RDB/AOF/主從的影響，我將會(huì)融入在后面具體模塊中。

數(shù)據(jù)庫(kù)通知

數(shù)據(jù)庫(kù)通知主要是利用redis中支持發(fā)布訂閱模式，讓客戶端通過(guò)訂閱給定的頻道或者模式(保存在client和server的pubsub_channels和pubsub_patterns，但保存的內(nèi)容不一樣)來(lái)獲悉數(shù)據(jù)庫(kù)中鍵的變化。主要分為鍵空間通知和事件通知兩類，要開(kāi)啟該功能需要在配置文件中設(shè)置參數(shù)。見(jiàn)notify-keyspace-events

鍵空間通知——某個(gè)鍵執(zhí)行了什么命令

keyspace@:

事件通知——某個(gè)事件被哪些命令執(zhí)行

keyevent@:

123456789101112131415161718192021222324252627282930313233343536373839404142

void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) { sds chan; robj *chanobj, *eventobj; int len = -1; char buf[24]; /* If any modules are interested in events, notify the module system now. * This bypasses the notifications configuration, but the module engine * will only call event subscribers if the event type matches the types * they are interested in. */ moduleNotifyKeyspaceEvent(type, event, key, dbid); /* If notifications for this class of events are off, return ASAP. */ if (!(server.notify_keyspace_events & type)) return; eventobj = createStringObject(event,strlen(event)); /* __keyspace@__: notifications. */ if (server.notify_keyspace_events & NOTIFY_KEYSPACE) { chan = sdsnewlen("__keyspace@",11); len = ll2string(buf,sizeof(buf),dbid); chan = sdscatlen(chan, buf, len); chan = sdscatlen(chan, "__:", 3); chan = sdscatsds(chan, key->ptr); chanobj = createObject(OBJ_STRING, chan); pubsubPublishMessage(chanobj, eventobj); decrRefCount(chanobj); } /* __keyevent@__: notifications. */ if (server.notify_keyspace_events & NOTIFY_KEYEVENT) { chan = sdsnewlen("__keyevent@",11); if (len == -1) len = ll2string(buf,sizeof(buf),dbid); chan = sdscatlen(chan, buf, len); chan = sdscatlen(chan, "__:", 3); chan = sdscatsds(chan, eventobj->ptr); chanobj = createObject(OBJ_STRING, chan); pubsubPublishMessage(chanobj, key); decrRefCount(chanobj); } decrRefCount(eventobj);}

參數(shù)說(shuō)明

當(dāng)然要啟用的話，必須要在配置文件中進(jìn)行設(shè)置：格式如下notify-keyspace-events Elg，若要啟用配置K和E必須要至少開(kāi)啟一個(gè)。

其中一個(gè)應(yīng)用就是：當(dāng)鍵過(guò)期或者被刪除時(shí)，需要通知應(yīng)用方，這個(gè)時(shí)候就可以xe兩個(gè)事件

總結(jié)

以上是生活随笔為你收集整理的redis源码分析 ppt_Redis源码分析之客户端+数据库的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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