1. 前言

Redis程式碼優美，註釋也很到位，閱讀起來會賞心悅目，大大降低了理解門檻。由於redis單執行緒幾乎完成所有工作，整體邏輯是相當複雜的，涉及了太多狀態，作者的技術深厚可見一斑。

Redis的單執行緒設計給出了一種優雅實現高效能服務思路，在實踐中值得借鑑。需要注意Redis並不是嚴格的單執行緒，實際上它是多程序+多執行緒。為解決IO和慢速操作帶的效能毛刺和卡頓，Redis實現引入的多程序和多執行緒。但是它的主體仍然是單執行緒的，單個執行緒完成網路IO操作和其它操作。

疑問：Redis選擇slot數為16384，是一個偶數，沒有采用質數，這個可能並不是一個最好的主意。

2. 名詞

RESP	REdis Serialization Protocol	Redis序列化協議
AE	A simple Event drived programming library	一個簡單的事務驅動程式設計庫
ASAP
AOF	Append Only File	僅僅追加寫檔案
BIO	Backgroup IO	後臺IO操作，三種涉及BIO：FSYNC、關閉檔案和記憶體free，均為阻塞或慢操作

3. dict.c

雜湊表的實現，雜湊函式使用了siphash雜湊演算法。

3.1. siphash演算法

一種非加密的64位雜湊演算法。

3.2. 核心函式

Redis中非常核心的演算法——雜湊表的實現在這個檔案中，很多命令都有用到，比如set/hset等，它是除記憶體分配管理外的最基礎實現。核心函式包含但不限於：dictFind、dictNext、dictAdd、dictDelete、dictFetchValue、dictGetHash、dictReplace、dictAddRaw、dictCreate、dictDelete、dictRelease、dictUnlink、dictRehash、dictEmpty、dictResize等。

3.3. 核心巨集

CLUSTER_SLOTS	定義Redis叢集Slots個數，值為16384
DICT_HT_INITIAL_SIZE	定義雜湊表預設大小巨集，值為4
dictSetVal	設定Value
dictSetKey	設定Key
dictCompareKeys	比較兩個Key
dictHashKey	對Key求雜湊
dictGetKey	取Key
dictGetVal	取Value
dictSlots	得到槽（slot）個數
dictSize	得到已用槽（slot）個數
dictGetSignedIntegerVal	取8位元組有符號整數值
dictGetUnsignedIntegerVal	取8位元組無符號整數值
dictGetDoubleVal	取double型別值

3.4. 核心結構體

3.4.1. dictEntry

定義了雜湊節點的資料結構：

typedef struct dictEntry {

void *key; // Key

union { // 支援4種類型的值

void *val; // 字串值，或二進位制值

uint64_t u64; // 8位元組無符號整數值

int64_t s64; // 8位元組有符號整數值

double d; // 雙精度浮點值

} v;

struct dictEntry *next; // 下一節點，鏈地址衝突解決法

} dictEntry;

4. Redis命令

各具體的命令並不做持久化（寫AOF等）和傳播給Slaves，這兩項是公共操作，在上層統一執行，具體函式為server.c中的“void call(client *c, int flags)”，具體執行函式為server.c中的“void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int flags)”。

4.1. SELECT命令

群裡有人問select跟普通命令的效率，猜想select沒有效率問題，閱讀原始碼也證實了這一點。因為它只是改變指標指向，所以不存在效率問題，自然比普通命令效率要高。

Redis支援的所有命令，都儲存在型別為redisCommand函式表中，函式表名為redisCommandTable的陣列中。

4.1.1. redisCommand結構體

如果以C++來理解redisCommand，可以將redisCommand看抽象基類，它定義了兩個虛擬函式proc和getkeys_proc，各種command是它的具體實現。

// server.h

struct redisCommand {

char *name; // 命令名，比如：GET

redisCommandProc *proc; // 命令處理過程（函式指標）

// 命令引數的個數，用於檢查命令請求的格式是否正確

// 如果這個值為負數-N，那麼表示引數的數量大於等於N

// 注意：命令的名字本身也是一個引數。

int arity;

// 字串形式的標識值，這個值記錄了命令的屬性

// 比如是讀命令還是寫命令，是否允許在載入資料時使用，是否允許在 Lua 指令碼中使用等

char *sflags; /* Flags as string representation, one char per flag. */

// sflags對應的二進位制值，方便程式“與”、“或”等操作操作

int flags; /* The actual flags, obtained from the 'sflags' field. */

/* Use a function to determine keys arguments in a command line.

* Used for Redis Cluster redirect. */

redisGetKeysProc *getkeys_proc; // 函式指標

/* What keys should be loaded in background when calling this command? */

int firstkey; /* The first argument that's a key (0 = no keys) */

int lastkey; /* The last argument that's a key */

int keystep; /* The step between first and last key */

// 伺服器執行這個命令所耗費的總時長

long long microseconds;;

// 伺服器總共執行了多少次這個命令

long long calls;

};

4.1.2. redisCommandTable變數

redisCommandTable的部分定義：

// server.c

struct redisCommand redisCommandTable[] = {

{"module",moduleCommand,-2,"as",0,NULL,0,0,0,0,0},

{"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},

{"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},

。。。。。。

{"select",selectCommand,2,"lF",0,NULL,0,0,0,0,0},

。。。。。。

{"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},

{"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},

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

};

4.1.3. selectCommand函式

// server.c

void selectCommand(client *c) {

long id;

// 第一個引數值為DB的ID

if (getLongFromObjectOrReply(c, c->argv[1], &id,

"invalid DB index") != C_OK)

return;

// 叢集不支援SELECT命令

if (server.cluster_enabled && id != 0) {

addReplyError(c,"SELECT is not allowed in cluster mode");

return;

}

if (selectDb(c,id) == C_ERR) { // 切換DB

addReplyError(c,"DB index is out of range");

} else {

addReply(c,shared.ok);

}

// db.c

int selectDb(client *c, int id) {

if (id < 0 || id >= server.dbnum) // dbnum值由redis.conf中的databases決定，預設為16

return C_ERR;

c->db = &server.db[id];

return C_OK;

}

4.2. SET命令

SET命令主要是字典操作（dict）。

4.2.1. setCommand函式

// t_string.c

/* SET key value [NX] [XX] [EX <seconds>] [PX <milliseconds>] */

void setCommand(client *c) {

int j;

robj *expire = NULL;

int unit = UNIT_SECONDS;

int flags = OBJ_SET_NO_FLAGS;

// 以下一長段均為引數選項解析，

// 第一3個開始，因為第一個為命令SET自身，第二個為KEY，第三個為VALUE

for (j = 3; j < c->argc; j++) {

char *a = c->argv[j]->ptr;

robj *next = (j == c->argc-1) ? NULL : c->argv[j+1];

if ((a[0] == 'n' || a[0] == 'N') &&

(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&

!(flags & OBJ_SET_XX))

{

flags |= OBJ_SET_NX;

} else if ((a[0] == 'x' || a[0] == 'X') &&

(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&

!(flags & OBJ_SET_NX))

{

flags |= OBJ_SET_XX;

} else if ((a[0] == 'e' || a[0] == 'E') &&

(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&

!(flags & OBJ_SET_PX) && next)

{

flags |= OBJ_SET_EX;

unit = UNIT_SECONDS;

expire = next;

j++;

} else if ((a[0] == 'p' || a[0] == 'P') &&

(a[1] == 'x' || a[1] == 'X') && a[2] == '\0' &&

!(flags & OBJ_SET_EX) && next)

{

flags |= OBJ_SET_PX;

unit = UNIT_MILLISECONDS;

expire = next;

j++;

} else {

addReply(c,shared.syntaxerr); // 語法錯誤

return;

}

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

setGenericCommand(c,flags,c->argv[1],c->argv[2],expire,unit,NULL,NULL);

}

// t_string.c

void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {

。。。。。。

setKey(c->db,key,val);

server.dirty++;

if (expire) setExpire(c,c->db,key,mstime()+milliseconds);

notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);

if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);

addReply(c, ok_reply ? ok_reply : shared.ok);

}

// db.c

void setKey(redisDb *db, robj *key, robj *val) {

if (lookupKeyWrite(db,key) == NULL) { // 效能開銷主要在這，參見dictFind函式

dbAdd(db,key,val);

} else {

dbOverwrite(db,key,val);

}

incrRefCount(val);

removeExpire(db,key);

signalModifiedKey(db,key);

}

// db.c

/* Add the key to the DB. It's up to the caller to increment the reference

* counter of the value if needed.

* The program is aborted if the key already exists. */

void dbAdd(redisDb *db, robj *key, robj *val) {

sds copy = sdsdup(key->ptr);

int retval = dictAdd(db->dict, copy, val);

serverAssertWithInfo(NULL,key,retval == DICT_OK);

if (val->type == OBJ_LIST ||

val->type == OBJ_ZSET)

signalKeyAsReady(db, key);

if (server.cluster_enabled) slotToKeyAdd(key);

}

// dict.h

#define dictSetVal(d, entry, _val_) do { \

if ((d)->type->valDup) \

(entry)->v.val = (d)->type->valDup((d)->privdata, _val_); \

else \

(entry)->v.val = (_val_); \

} while(0)

// db.c

// KEY存在時覆蓋寫

/* Overwrite an existing key with a new value. Incrementing the reference

* count of the new value is up to the caller.

* This function does not modify the expire time of the existing key.

* The program is aborted if the key was not already present. */

void dbOverwrite(redisDb *db, robj *key, robj *val) {

dictEntry *de = dictFind(db->dict,key->ptr);

serverAssertWithInfo(NULL,key,de != NULL);

dictEntry auxentry = *de;

robj *old = dictGetVal(de);

// 記憶體策略由redis.conf中的maxmemory-policy決定

// LFU策略是4.0開始引入的（Least Frequently Used，最不經常使用）

// #define MAXMEMORY_FLAG_LRU (1<<0)

// #define MAXMEMORY_FLAG_LFU (1<<1)

// #define MAXMEMORY_FLAG_ALLKEYS (1<<2)

// #define MAXMEMORY_FLAG_NO_SHARED_INTEGERS (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU)

// #define MAXMEMORY_VOLATILE_LRU ((0<<8)|MAXMEMORY_FLAG_LRU)

// #define MAXMEMORY_VOLATILE_LFU ((1<<8)|MAXMEMORY_FLAG_LFU)

// #define MAXMEMORY_VOLATILE_TTL (2<<8)

// #define MAXMEMORY_VOLATILE_RANDOM (3<<8)

// #define MAXMEMORY_ALLKEYS_LRU ((4<<8)|MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_ALLKEYS)

// #define MAXMEMORY_ALLKEYS_LFU ((5<<8)|MAXMEMORY_FLAG_LFU|MAXMEMORY_FLAG_ALLKEYS)

// #define MAXMEMORY_ALLKEYS_RANDOM ((6<<8)|MAXMEMORY_FLAG_ALLKEYS)

// #define MAXMEMORY_NO_EVICTION (7<<8)

if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {

val->lru = old->lru;

}

dictSetVal(db->dict, de, val);

if (server.lazyfree_lazy_server_del) {

freeObjAsync(old);

dictSetVal(db->dict, &auxentry, NULL);

}

dictFreeVal(db->dict, &auxentry);

}

4.2.2. dictAdd函式

// dict.c

// KEY不存在時新增

/* Add an element to the target hash table */

int dictAdd(dict *d, void *key, void *val)

{

dictEntry *entry = dictAddRaw(d,key,NULL);

if (!entry) return DICT_ERR;

dictSetVal(d, entry, val);

return DICT_OK;

}

4.2.3. dictFind函式

字典查詢實現：

dictEntry *dictFind(dict *d, const void *key)

{

dictEntry *he;

uint64_t h, idx, table;

if (d->ht[0].used + d->ht[1].used == 0) return NULL; /* dict is empty */

if (dictIsRehashing(d)) _dictRehashStep(d);

// #define dictHashKey(d, key) (d)->type->hashFunction(key)

h = dictHashKey(d, key); // 對key求雜湊值

for (table = 0; table <= 1; table++) {

idx = h & d->ht[table].sizemask;

he = d->ht[table].table[idx];

while(he) { // 鏈式衝突解決法

if (key==he->key || dictCompareKeys(d, key, he->key))

return he;

he = he->next;

}

if (!dictIsRehashing(d)) return NULL; // 重雜湊

}

return NULL;

}

typedef struct dictType {

uint64_t (*hashFunction)(const void *key);

void *(*keyDup)(void *privdata, const void *key);

void *(*valDup)(void *privdata, const void *obj);

int (*keyCompare)(void *privdata, const void *key1, const void *key2);

void (*keyDestructor)(void *privdata, void *key);

void (*valDestructor)(void *privdata, void *obj);

} dictType;

4.3. HSETNX命令

4.3.1. hsetnxCommand函式

// t_hash.c

void hsetnxCommand(client *c) {

robj *o;

if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;

hashTypeTryConversion(o,c->argv,2,3);

if (hashTypeExists(o, c->argv[2]->ptr)) {

addReply(c, shared.czero);

} else {

hashTypeSet(o,c->argv[2]->ptr,c->argv[3]->ptr,HASH_SET_COPY);

addReply(c, shared.cone);

signalModifiedKey(c->db,c->argv[1]);

notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id);

server.dirty++;

}

4.3.2. hashTypeSet函式

// t_hash.c

int hashTypeSet(robj *o, sds field, sds value, int flags) {

if (o->encoding == OBJ_ENCODING_ZIPLIST) {

// ziplistInsert

// ziplistPush

} else if (o->encoding == OBJ_ENCODING_HT) {

。。。。。。

dictAdd(o->ptr,f,v);

} else {

serverPanic("Unknown hash encoding");

}

5. Redis多路複用機制

5.1. AE是什麼？

“AE”為“A simple event-driven programming library”的縮寫，翻譯成中文，即一個簡單的事件驅動程式設計庫。就Linux而言，可簡單理解為對epoll的封裝。

5.2. 多路複用選擇aeApiPoll

Redis支援select、epoll、kqueue和evport四種模式，編譯Redis時決定採用哪一種，因此執行時只會有一種有效。優先順序是：evport -> epoll -> kqueue -> select，其中select為兜底用，因為所有系統都會支援select。

對應的原始碼檔案，分別為：

evport	ae_evport.c	Solaris
epoll	ae_epoll.c	Linux系統
kqueue	ae_kqueue.c	BSD類系統
select	ae_select.c	其它不支援epoll、evport和kqueue系統

橋接它們的檔案則是ae.c，橋接程式碼：

/* Include the best multiplexing layer supported by this system.

* The following should be ordered by performances, descending. */

#ifdef HAVE_EVPORT

#include "ae_evport.c"

#else

#ifdef HAVE_EPOLL

#include "ae_epoll.c"

#else

#ifdef HAVE_KQUEUE

#include "ae_kqueue.c"

#else

#include "ae_select.c"

#endif

巨集HAVE_EVPORT、HAVE_EPOLL、HAVE_KQUEUE則在config.h中決議出：

#ifdef __sun

#include <sys/feature_tests.h>

#ifdef _DTRACE_VERSION

#define HAVE_EVPORT 1

#endif

#ifdef __linux__

#define HAVE_EPOLL 1

#endif

#if (defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)

#define HAVE_KQUEUE 1

#endif

5.3. 多路複用呼叫aeCreateFileEvent

自main函式開始的呼叫過程（讀事件和寫事件的順序，作者在這裡用了個小技巧，支援優先響應是查詢，即立即返回查詢結果）：

int main(int argc, char **argv) { // server.c

。。。。。。

// aeMain(server.el);

void aeMain(aeEventLoop *eventLoop) // ae.c

{

while (!eventLoop->stop) {

if (eventLoop->beforesleep != NULL)

eventLoop->beforesleep(eventLoop); // 這很重要，寫AOF檔案在這裡進行

// aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);

int aeProcessEvents(aeEventLoop *eventLoop, int flags) { // ae.c

// linux實際呼叫的是ae_epoll.c中的aeApiPoll函式

numevents = aeApiPoll(eventLoop, tvp);

if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)

eventLoop->aftersleep(eventLoop);

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

// 一般優先處理讀事件，然後再寫事件，但有時為立即響應是查詢則寫優先

/* Normally we execute the readable event first,

and the writable event laster. */

int invert = fe->mask & AE_BARRIER;

/* Fire the readable event if the call sequence is not inverted. */

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

/* Fire the writable event. */

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

/* If we have to invert the call,

fire the readable event now after the writable one. */

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

// 函式aeCreateFileEvent中設定rfileProc和wfileProc

// rfileProc和wfileProc的初始化均在aeCreateFileEvent中完成

}

。。。。。。

}

// 初始化rfileProc和wfileProc

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,

aeFileProc *proc, void *clientData)

{

if (fd >= eventLoop->setsize) {

errno = ERANGE;

return AE_ERR;

}

aeFileEvent *fe = &eventLoop->events[fd]; // events是一個以fd為下標的陣列，0查詢

if (aeApiAddEvent(eventLoop, fd, mask) == -1) // fd放到epoll或select等中

return AE_ERR;

fe->mask |= mask; // 讀寫事件

if (mask & AE_READABLE) fe->rfileProc = proc;

if (mask & AE_WRITABLE) fe->wfileProc = proc;

fe->clientData = clientData; // 附加資料

if (fd > eventLoop->maxfd)

eventLoop->maxfd = fd; // 這個主要給select時用，epoll等不需要

return AE_OK;

}

6. 網路收發過程

6.1. 過程簡要

main

-> initServer

-> aeCreateFileEvent(acceptTcpHandler)

-> createClient

-> aeCreateFileEvent(readQueryFromClient)

-> read

Redis原始碼筆記-初步

目錄

1. 前言

2. 名詞

3. dict.c

3.1. siphash演算法

3.2. 核心函式

3.3. 核心巨集

3.4. 核心結構體

3.4.1. dictEntry

4. Redis命令

4.1. SELECT命令

4.1.1. redisCommand結構體

4.1.2. redisCommandTable變數

4.1.3. selectCommand函式

4.2. SET命令

4.2.1. setCommand函式

4.2.2. dictAdd函式

4.2.3. dictFind函式

4.3. HSETNX命令

4.3.1. hsetnxCommand函式

4.3.2. hashTypeSet函式

5. Redis多路複用機制

5.1. AE是什麼？

5.2. 多路複用選擇aeApiPoll

5.3. 多路複用呼叫aeCreateFileEvent

6. 網路收發過程

6.1. 過程簡要

相關推薦