Redis 底层数据结构 dict（hashtable）的实现机制
Redis 可以被看作是一个存储了 key 和 value 的映射关系的字典。无论是哪种数据类型，其 value 最终都会存储到这个大的字典当中，而它的 key 则被用作索引。
Redis 中包含多个数据库（具体数量上限可以通过配置文件修改），每一个数据库在底层都对应一个 redisDb 结构。Redis 客户端即是通过与数据库实例对应的 redisDb 进行交互而将数据存入 Redis 当中。
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 */
unsigned long expires_cursor; /* Cursor of the active expire cycle. */
list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */
} redisDb;
可以看出，redisDb 中有一项 dict *dict ，数据库中的数据即是存储在这个结构中，而 dict 的底层实现则是通过 hashtable（这里需要强调，在 Redis 中，hashtable 是一种数据类型，但这里所说的 hashtable 是指数据结构，也就是 Redis 中的 dict。下文中提到的 hashtable 如果没有特别说明都是值的数据结构中的 hashtable）。
dict 数据结构介绍
- dictht & dictEntry
Redis 中 hashtable 具体由 dictht 来实现。其中，table 结构是一个 dictEntry 类型的数组，而 dictEntry 是一个单向链表中的节点，代表的是 hashtable 中的一条记录，包含了该条记录的 key、value 以及指向下一个节点（下一条记录）的指针（hash 碰撞）。
typedef struct dictht {
dictEntry **table; /* 链表数组 */
unsigned long size; /* bucket 的数量 */
unsigned long sizemask;
unsigned long used; /* 存储元素的数量 */
} dictht;
typedef struct dictEntry {
void *key;
union {
void *val;
uint64_t u64;
int64_t s64;
double d;
} v;
struct dictEntry *next;
} dictEntry;
这里需要注意，dictEntry 中的 key 和 val 都是 void 类型，意味着可以存储任意数据类型。所以，即使 Redis 中一些数据类型本身的底层实现也是 hashtable（set、zset、hash 等），但仍然可以存储在 redisDb 中的 dict 中。
dictht 中的 size 表示 hashtable 中 bucket 的数量，sizemask 是 size 的位掩码，用于计算对应的 key 应该存储在哪个 bucket 当中（位运算的效率高于取模运算）。而 used 则用来存储当前 hashtable 中实际存储的记录（元素）的数量。
- dict & dictType
由于 Redis 是单进程单线程模型，为了避免在 hash 扩容以及 rehash 的过程中阻塞服务，dict 结构中的 ht 定义成了一个数组，维护两个 dictht 。rehashidx 则是在 rehash 过程中被用作 ht[0] 中的 bucket 的索引。pauserehash 用来标识 rehash 是否暂停。
typedef struct dict {
dictType *type;
void *privdata;
dictht ht[2];
long rehashidx; /* rehashing not in progress if rehashidx == -1 */
int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
} dict;
再来看 dictType ，dictType 中定义了一些函数，这些函数共同决定了 dict 中的 key 和 value 的行为。在实际应用中，针对不同的数据类型以及应用场景，Redis 又对 dictType 做了一些约束，例如：对于 set 数据类型，dict 中只有 key 而没有 value 。
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);
int (*expandAllowed)(size_t moreMem, double usedRatio);
} dictType;
extern dictType objectKeyPointerValueDictType;
extern dictType objectKeyHeapPointerValueDictType;
extern dictType setDictType;
extern dictType zsetDictType;
extern dictType clusterNodesDictType;
extern dictType clusterNodesBlackListDictType;
extern dictType dbDictType;
extern dictType shaScriptObjectDictType;
extern dictType hashDictType;
extern dictType replScriptCacheDictType;
extern dictType dbExpiresDictType;
extern dictType modulesDictType;
extern dictType sdsReplyDictType;
dictType setDictType = {
dictSdsHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsDestructor, /* key destructor */
NULL /* val destructor */
};
⒈ dict 的创建
dict 的创建只是对数据结构进行初始化，但并不会初始化 hashtable 中的 bucket 。
// 为 dict 分配内存空间
dict *dictCreate(dictType *type, void *privDataPtr)
{
dict *d = zmalloc(sizeof(*d));
_dictInit(d,type,privDataPtr);
return d;
}
// 初始化 dict
int _dictInit(dict *d, dictType *type, void *privDataPtr)
{
_dictReset(&d->ht[0]);
_dictReset(&d->ht[1]);
d->type = type;
d->privdata = privDataPtr;
d->rehashidx = -1;
d->pauserehash = 0;
return DICT_OK;
}
// 初始化 dictht
static void _dictReset(dictht *ht)
{
ht->table = NULL;
ht->size = 0;
ht->sizemask = 0;
ht->used = 0;
}
⒉ 新增元素
当往 dict 中新增元素时，首先需要在 dictht 的 table 中为新增的元素分配一个 dictEntry ，然后再往这个新分配的 dictEntry 中填充数据。
在新增元素时，如果 dict 正在进行 rehash ，则 dictEntry 会分配到 dict.ht[1] 中，否则会分配到 dict.ht[0] 中。
另外，对于刚初始化完成的 dict，在第一次新增元素时还会顺带初始化 bucket 。
// 向 dict 中新增元素
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;
}
// 分配 dictEntry
dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
long index;
dictEntry *entry;
dictht *ht;
if (dictIsRehashing(d)) _dictRehashStep(d);
/* Get the index of the new element, or -1 if
* the element already exists. */
if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
return NULL;
/* Allocate the memory and store the new entry.
* Insert the element in top, with the assumption that in a database
* system it is more likely that recently added entries are accessed
* more frequently. */
ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
entry = zmalloc(sizeof(*entry));
entry->next = ht->table[index];
ht->table[index] = entry;
ht->used++;
/* Set the hash entry fields. */
dictSetKey(d, entry, key);
return entry;
}
在分配 dictEntry 的过程中，首先通过函数 dictHashKey 计算新增元素的 key 的 hash 值。然后通过函数 _dictKeyIndex 找到新增元素应该被分配到的 bucket 的索引位置。在此过程中，如果是刚创建的 dict，那么 dict.ht[0].table 的长度（hashtable 中 bucket 的数量）会被初始化为 4。如果当前 dict 正在进行 rehash ，那么新的 dictEntry 会被分配到 dict.ht[1].table 中。
// 找到 dictEntry 应该被分配到的 bucket 的索引
static long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing)
{
unsigned long idx, table;
dictEntry *he;
if (existing) *existing = NULL;
/* 如果需要，则对 hashtable 进行扩容 */
if (_dictExpandIfNeeded(d) == DICT_ERR)
return -1;
for (table = 0; table <= 1; table++) {
idx = hash & d->ht[table].sizemask;
/* Search if this slot does not already contain the given key */
he = d->ht[table].table[idx];
while(he) {
if (key==he->key || dictCompareKeys(d, key, he->key)) {
if (existing) *existing = he;
return -1;
}
he = he->next;
}
// 如果当前 dict 没有进行 rehash，则分配到 ht[0] 中
if (!dictIsRehashing(d)) break;
}
return idx;
}
// 如果需要，对 hashtable 进行扩容
#define DICT_HT_INITIAL_SIZE 4
static int dict_can_resize = 1;
static unsigned int dict_force_resize_ratio = 5;
static int _dictExpandIfNeeded(dict *d)
{
/* 正在进行 rehash，则返回 */
if (dictIsRehashing(d)) return DICT_OK;
/* 如果 hashtable 为空，则初始化其 size，默认为 4 */
if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);
/* hashtable 扩容的条件：
* hashtable 中元素数量与 bucket 数量达到 1:1 ，并且可以进行扩容操作
* 或者
* hashtable 中元素数量与 bucket 的数量之比超过阈值并且当前内存空间足够支持扩容
*/
if (d->ht[0].used >= d->ht[0].size &&
(dict_can_resize ||
d->ht[0].used/d->ht[0].size > dict_force_resize_ratio) &&
dictTypeExpandAllowed(d))
{
return dictExpand(d, d->ht[0].used + 1);
}
return DICT_OK;
}
找到 bucket 的索引位置之后，将为 dictEntry 分配内存空间，然后将 dictEntry 添加到 bucket 中链表的头部位置，最后将数据填充到新分配的 dictEntry 中。
在函数 dictAddRaw() 的运行过程中，如果给定的 key 在 hashtable 中已经存在，则此时会找到给定 key 对应的 dictEntry 并将其赋值给 existing 。existing 是一个指针类型， 对 existing 中的 v 的更新操作即是对 hashtable 中元素的更新操作。Redis 中 hashtable 的更新操作即是此种实现机制，只不过在更新完成以后还需要释放旧的 v 所占用的内存空间。
⒊ 删除元素
hashtable 中的元素删除有两种实现方式：dictDelete() 和 dictUnlink() 。二者的区别在于，dictDelete() 将 dictEntry 从 hashtable 中移除的同时还会释放其所占用的内存空间，而 dictUnlink() 只是将 dictEntry 从 hashtable 中移除，然后将其返回。如果还需要释放其所占用的内存空间，则需要额外调用 dictFreeUnlinkedEntry() 。
如果在彻底删除元素之前还需要对元素进行一些操作，那么此时应该选择 dictUnlink() 来进行元素删除。dictUnlink() 在将元素从 hashtable 中移除后会将 dictEntry 返回，调用者可以对 dictEntry 进行期望的操作，然后调用 dictFreeUnlinkedEntry() 释放其所占用的内存空间。如果使用 dictDelete() ，则调用者需要先查找 hashtable，找到期望的 dictEntry，然后对其进行期望的操作，然后再调用 dictDelete() 进行删除操作。综上可以看出，第一种方案比第二种方案少进行了一次 hashtable 的查找操作。
static dictEntry *dictGenericDelete(dict *d, const void *key, int nofree) {
uint64_t h, idx;
dictEntry *he, *prevHe;
int table;
if (d->ht[0].used == 0 && d->ht[1].used == 0) return NULL;
if (dictIsRehashing(d)) _dictRehashStep(d);
h = dictHashKey(d, key);
for (table = 0; table <= 1; table++) {
idx = h & d->ht[table].sizemask;
he = d->ht[table].table[idx];
prevHe = NULL;
while(he) {
if (key==he->key || dictCompareKeys(d, key, he->key)) {
/* Unlink the element from the list */
if (prevHe)
prevHe->next = he->next;
else
d->ht[table].table[idx] = he->next;
if (!nofree) {
dictFreeKey(d, he);
dictFreeVal(d, he);
zfree(he);
}
d->ht[table].used--;
return he;
}
prevHe = he;
he = he->next;
}
if (!dictIsRehashing(d)) break;
}
return NULL; /* not found */
}
dictDelete() 和 dictUnlink() 都是通过调用 dictGenericDelete() 来实现删除功能。区别在于 nofree 的传参，前者 nofree 参数为 0，后者为 1。当 nofree 值为 0 时，在从 hashtable 中移除 dictEntry 的同时会释放其所占用的内存空间。
⒋ 扩容
上文已经提到过，在内存空间足够的前提下，如果 dictht 中的元素数量与 bucket 数量达到 1:1 并且当前 dict 可以进行扩容操作（dict_can_resize = 1）或者dictht 中的元素数量与 bucket 数量的比值超过阈值（dict_force_resize_ratio = 5），则 dict 需要进行扩容操作。
那么 Redis 是如何判断当前内存空间是否能够满足扩容的需求的呢？在 Redis 的配置文件中会有相应的配置项来设置分配给 Redis 的最大内存空间，如果这个值不设置，则默认分配给 Redis 的内存空间没有限制。在当前 hashtable 的元素数量与 bucket 数量之比还没有达到 Redis 所设置的最大负载因子的前提下，Redis 会判断系统所分配的最大内存空间减去当前已经分配的空间之后的值是否能满足扩容后所需要的空间；如果还无法满足，那么 Redis 会判断加上已分配但还没有使用的空间是否足够，如果还不够，那么本次扩容请求不被允许。
基于上述扩容的条件判断逻辑，如果一次扩容需要的内存空间很大，那么在本次扩容所需要的内存空间分配完成之前，Redis 服务会阻塞。另外，如果当前 Redis 的可用内存空间已经无法满足扩容所需要的内存空间，那么 Redis 还会清除一些数据以使得总的内存消耗不超过配置文件中所设置的上限。
/* 判断内存空间是否足够扩容的逻辑 */
static int dictTypeExpandAllowed(dict *d) {
if (d->type->expandAllowed == NULL) return 1;
return d->type->expandAllowed(
_dictNextPower(d->ht[0].used + 1) * sizeof(dictEntry*),
(double)d->ht[0].used / d->ht[0].size);
}
#define HASHTABLE_MAX_LOAD_FACTOR 1.618
/* 如果当前元素数量与 bucket 数量之比不超过最大负载因子。判断可用内存空间是否足够扩容
* 如果超过了股灾因子，则无论如何都允许扩容
*/
int dictExpandAllowed(size_t moreMem, double usedRatio) {
if (usedRatio <= HASHTABLE_MAX_LOAD_FACTOR) {
return !overMaxmemoryAfterAlloc(moreMem);
} else {
return 1;
}
}
/* 判断当前可用内存空间是否足够扩容 */
int overMaxmemoryAfterAlloc(size_t moremem) {
if (!server.maxmemory) return 0; /* No limit. */
/* Check quickly. */
size_t mem_used = zmalloc_used_memory();
if (mem_used + moremem <= server.maxmemory) return 0;
size_t overhead = freeMemoryGetNotCountedMemory();
mem_used = (mem_used > overhead) ? mem_used - overhead : 0;
return mem_used + moremem > server.maxmemory;
}
由于 Redis 采用单进程单线程模型，在扩容过程中需要避免阻塞 Redis 服务，所以，dict.ht 被设计成一个长度为 2 的数组。Redis 服务与客户端的交互都在 dict.ht[0] 中进行，而扩容操作在 dict.ht[1] 中进行，这样就避免了扩容操作可能引起的服务阻塞。扩容完成之后会对 dict.ht[0] 中存储的元素进行 rehash 操作，将其中的元素移动到 dict.ht[1] 中，待 rehash 完成之后，再将 dict.ht[1] 的值赋给 dict.ht[0] ，然后将 dict.ht[1] 重置，以备后续的扩容操作。
同样是为了避免阻塞 Redis 服务，rehash 过程也不是一次执行完成的，而是在每次操作 dict 时（读/写）最多移动一个 bucket 的数据（n =1）。另外，由于 hash 的特点，不可避免的会有空的 bucket 出现，为了尽可能快的响应客户端，如果在 rehash 过程中连续遇到 10 * n 个空的 bucket，则本次 rehash 操作终止。在 rehash 的过程中，新添加的元素会直接添加到 ht[1] 中。
int _dictExpand(dict *d, unsigned long size, int* malloc_failed)
{
if (malloc_failed) *malloc_failed = 0;
/* 扩容后的 size 不能小于当前元素的个数 */
if (dictIsRehashing(d) || d->ht[0].used > size)
return DICT_ERR;
dictht n; /* the new hash table */
/* 扩容后的 bucket 数量一定是 2 的 n 次方 */
unsigned long realsize = _dictNextPower(size);
/* 防止溢出 */
if (realsize < size || realsize * sizeof(dictEntry*) < realsize)
return DICT_ERR;
/* 如果扩容后的 size 与当前 size 相同，则 rehash 没有意义 */
if (realsize == d->ht[0].size) return DICT_ERR;
n.size = realsize;
n.sizemask = realsize-1;
if (malloc_failed) {
n.table = ztrycalloc(realsize*sizeof(dictEntry*));
*malloc_failed = n.table == NULL;
if (*malloc_failed)
return DICT_ERR;
} else
n.table = zcalloc(realsize*sizeof(dictEntry*));
n.used = 0;
/* 如果是新创建的 dict，则直接赋值给 ht[0] */
if (d->ht[0].table == NULL) {
d->ht[0] = n;
return DICT_OK;
}
/* 否则赋值给 ht[1] 准备 rehash */
d->ht[1] = n;
d->rehashidx = 0;
return DICT_OK;
}
// 计算扩容后的 bucket 数量
static unsigned long _dictNextPower(unsigned long size)
{
unsigned long i = DICT_HT_INITIAL_SIZE;
if (size >= LONG_MAX) return LONG_MAX + 1LU;
while(1) {
if (i >= size)
return i;
i *= 2;
}
}
int dictRehash(dict *d, int n) {
int empty_visits = n*10; /* 最多遇到的空的 bucket 的数量 */
if (!dictIsRehashing(d)) return 0;
while(n-- && d->ht[0].used != 0) {
dictEntry *de, *nextde;
assert(d->ht[0].size > (unsigned long)d->rehashidx);
while(d->ht[0].table[d->rehashidx] == NULL) {
d->rehashidx++;
/* 连续遇到的空的 bucket 数量达到 10 个，则直接返回 */
if (--empty_visits == 0) return 1;
}
de = d->ht[0].table[d->rehashidx];
/* 移动 bucket 中的数据 */
while(de) {
uint64_t h;
nextde = de->next;
/* 计算元素在 ht[1] 中的索引位置 */
h = dictHashKey(d, de->key) & d->ht[1].sizemask;
de->next = d->ht[1].table[h];
d->ht[1].table[h] = de;
d->ht[0].used--;
d->ht[1].used++;
de = nextde;
}
d->ht[0].table[d->rehashidx] = NULL;
d->rehashidx++;
}
/* 如果 ht[0] 中所有的 bucket 都已经移动到 ht[1]，则将 ht[1] 赋值给 ht[0] 并重置 ht[1] */
if (d->ht[0].used == 0) {
zfree(d->ht[0].table);
d->ht[0] = d->ht[1];
_dictReset(&d->ht[1]);
d->rehashidx = -1;
return 0;
}
/* More to rehash... */
return 1;
}
⒌ 后台任务 Cron
Redis 服务通过函数 databasesCron 维护一个后台定时任务，该函数的运行频率通过配置文件中的配置项 hz 来设置，该值设置了函数 databasesCron 每秒钟运行的次数，默认为 10。这个后台定时任务主要负责清理过期的 key 、hashtable 的缩容以及在 Redis 服务空闲时进行 rehash 。
- rehash
前面提到，为了确保 Redis 服务尽快响应客户端的请求，在每次操作 dict 时最多移动一个 bucket 。但如果 Redis 服务处于空闲状态，则 rehash 可以不受此限制。
当 Redis 服务处于空闲状态时，Redis 会使用 1ms 的时间对 dict 进行 rehash 操作。这期间，每次 rehash 时 n 的值设为 100，在每次操作完成之后会检查当前时间距离开始时间是否超过 1 ms，如果是，那么停止 rehash ，否则继续。
int incrementallyRehash(int dbid) {
/* Keys dictionary */
if (dictIsRehashing(server.db[dbid].dict)) {
dictRehashMilliseconds(server.db[dbid].dict,1);
return 1; /* already used our millisecond for this loop... */
}
/* ... ... */
return 0;
}
int dictRehashMilliseconds(dict *d, int ms) {
if (d->pauserehash > 0) return 0;
long long start = timeInMilliseconds();
int rehashes = 0;
while(dictRehash(d,100)) {
rehashes += 100;
if (timeInMilliseconds()-start > ms) break;
}
return rehashes;
}
- 缩容
前面提到，当 dict 中的元素数量与 bucket 数量之比达到一定值时，dict 会进行扩容。同样，当元素数量与 bucket 数量之比小于一定值（10%），并且此时 bucket 的数量还大于 Redis 服务所允许的最小值（4）时，Redis 服务会对 dict 进行缩容操作。
void tryResizeHashTables(int dbid) {
if (htNeedsResize(server.db[dbid].dict))
dictResize(server.db[dbid].dict);
/* ... ... */
}
#define HASHTABLE_MIN_FILL 10
int htNeedsResize(dict *dict) {
long long size, used;
size = dictSlots(dict);
used = dictSize(dict);
return (size > DICT_HT_INITIAL_SIZE &&
(used*100/size < HASHTABLE_MIN_FILL));
}
int dictResize(dict *d)
{
unsigned long minimal;
if (!dict_can_resize || dictIsRehashing(d)) return DICT_ERR;
minimal = d->ht[0].used;
if (minimal < DICT_HT_INITIAL_SIZE)
minimal = DICT_HT_INITIAL_SIZE;
return dictExpand(d, minimal);
}
缩容操作与扩容操作逻辑相同，只不过缩容是将 bucket 数量减小到能容纳当前 dict 中所有元素同时又是 2 的 n 次方的最小值。