自学内容网 自学内容网

Redis 散列

1. 数据结构

我们自底向上来描述redis散列涉及的数据结构。

首先是负责存储键值的结构,Java 中存储的结构叫 Entry,redis里也差不多,叫dictEntry:

typedef struct dictEntry {
    void *key;                // 键,它是一个指针类型,所以我们可以将其指向sds的指针
    union {                   // 这是一个联合类型,也就是你可以选择任意一个字段来存储
        void *val;            // 因为散列也是redis数据库的底层结构,它是使用val字段
        uint64_t u64;         // redis还用散列保存键过期时间,此时就用u64存储过期时间
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next;    // Java中为了处理哈希冲突,是采用拉链法,redis也是一样
} dictEntry;

我们向上,就是 dictEntry 的存储容器,也叫 dictht ,我们就叫他哈希

typedef struct dictht {
    dictEntry **table;         // 指针数组               
    unsigned long size;        // 底层数组长度,或者会所是slot数量
    unsigned long sizemask;    // 它比size小1,位运算比取模快,用它加速定位hash值索引
    unsigned long used;        // 这个是当前哈希表里所有数据的总和,哈希冲突的也算
} dictht;

然后,其实到这里基本可以用了。但是redis依然在 dicthst 之上抽象了 dict 结构,我们叫他散列:

typedef struct dict {
    dictType *type;         // redis将数据和函数分离,dictType 是 API 抽象
    void *privdata;         // 网上说是和 type 一起用的,我们初始化时都是NULL
    dictht ht[2];           // len=2的dictht数组,主要用0,当rehast时会开放1
    long rehashidx;         // rehashidx == -1 则处于rehashing
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
} dict;

下面是作为数据库键值对存储时候用到的 dictType

/* Db->dict, keys are sds strings, vals are Redis objects. */
dictType dbDictType = {
    dictSdsHash,                /* hash function */
    NULL,                       /* key dup */
    NULL,                       /* val dup */
    dictSdsKeyCompare,          /* key compare */
    dictSdsDestructor,          /* key destructor */
    dictObjectDestructor,       /* val destructor */
    dictExpandAllowed           /* allow to expand */
};

下面是作为失效控制存储时候用到的 dictType

/* Db->expires */
dictType dbExpiresDictType = {
    dictSdsHash,                /* hash function */
    NULL,                       /* key dup */
    NULL,                       /* val dup */
    dictSdsKeyCompare,          /* key compare */
    NULL,                       /* key destructor */
    NULL,                       /* val destructor */
    dictExpandAllowed           /* allow to expand */
};

下面还有针对事务watch专用的 dictType

/* Keylist hash table type has unencoded redis objects as keys and
 * lists as values. It's used for blocking operations (BLPOP) and to
 * map swapped keys to a list of clients waiting for this keys to be loaded. */
dictType keylistDictType = {
    dictObjHash,                /* hash function */
    NULL,                       /* key dup */
    NULL,                       /* val dup */
    dictObjKeyCompare,          /* key compare */
    dictObjectDestructor,       /* key destructor */
    dictListDestructor,         /* val destructor */
    NULL                        /* allow to expand */
};

最后,这部分是redis服务器上的DB字段,我们关注散列,就是因为它可以用作数据库的实现,所以这里也列举一下:

struct redisServer {
    redisDb *db;
}
/* Redis database representation. There are multiple databases identified
 * by integers from 0 (the default database) up to the max configured
 * database. The database number is the 'id' field in the structure. */
typedef struct redisDb {
    dict *dict;                 // DB 的 keyspace,键空间,当然不是只存储键
    dict *expires;              // DB 中键的过期时间
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP)*/
    dict *ready_keys;           /* Blocked keys that received a PUSH */
    dict *watched_keys;         /* 事务中通过watch观察的键
    int id;                     // 0-15,redis支持16个数据库
    long long avg_ttl;          // 平均ttl。没啥用
    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;

需要注意的是,redis 支持的map结构,底层的实现可能是ziplist,即压缩列表,也可能是dict,即散列。我们将名词规范化。

2.生命周期:初始化

我们先看作为数据库键值对存储时的初始化,这是在redis服务器启动时的一部分(DB 里直接使用dict,他压根没想用ziplist):

for (j = 0; j < server.dbnum; j++) {
    server.db[j].dict = dictCreate(&dbDictType,NULL);
    server.db[j].expires = dictCreate(&dbExpiresDictType,NULL);
    server.db[j].expires_cursor = 0;
    server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
    server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
    server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
    server.db[j].id = j;
    server.db[j].avg_ttl = 0;
    server.db[j].defrag_later = listCreate();
    listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
}

我们知道redis默认支持16个库,这里就是for循环生成16个数据库,然后我们进入 dictCreate 方法

/* Create a new hash table */
dict *dictCreate(dictType *type,
        void *privDataPtr)
{
    dict *d = zmalloc(sizeof(*d));         // 使用malloc 分配内存,返回指针

    _dictInit(d,type,privDataPtr);         // 委托其他方法
    return d;
}

官方注释也很清晰,这就是创建一个新的哈希表的方法,我们传入了两个参数,一个就是 dictType ,后续针对该哈希表的操作的一些可以自定义的 API,都是由这个结构体来处理的。

我理解这个非常像 Java 里的抽象类里未实现的方法,交给子类去自定义。

/* Initialize the hash table */
int _dictInit(dict *d, dictType *type,
        void *privDataPtr)
{
    _dictReset(&d->ht[0]);         // 初始化 dict.ht[0]
    _dictReset(&d->ht[1]);         // 初始化 dict.ht[1]
    d->type = type;                // 这个玩意就是 dictType
    d->privdata = privDataPtr;     // 这个玩意是个NULL
    d->rehashidx = -1;             // rehash索引,-1表示没有发生rehash
    d->pauserehash = 0;
    return DICT_OK;
}
/* Reset a hash table already initialized with ht_init().
 * NOTE: This function should only be called by ht_destroy(). */
static void _dictReset(dictht *ht)
{
    ht->table = NULL;        // dictEntry **table 哈希里的字段声明
    ht->size = 0;
    ht->sizemask = 0;
    ht->used = 0;
}

dictCreate 委托给 _dictInit 方法来初始化我们刚生成的 dict,这里的工作,其实很简答,就是初始化,需要注意的是,此时 table 数组是 NULL,等真正 set 值时才会初始化。

3.生命周期:请求

3.1 hset 命令

Redis 负责处理 hset 命令的函数就叫 hsetCommand ,而且redis里的命令都是这个起名法:

typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                            * LFU data (least significant 8 bits frequency
                            * and most significant 16 bits access time). */
    int refcount;        // 这个字段不用太关注
    void *ptr;           // 指向真实结构体的指针,比如我们现在讲散列,那么它可能指向
                         // ziplist,也可能指向 dict
} robj;

robj 是贯穿 redis 的一个容器类型,你将其想象成 Java 里的 List,Optional。

主要关注type,encoding以及ptr

  • type:该对象的类型,字符串,列表,集合,有序集合,map,stream
  • encoding:每种type会因为各种性能考虑,在不同场景下有多重编码方式,啥是编码??就是底层数据结构的类型。打个比方,Java 里 List 在底层可以用数组,也可以用链表实现
  • ptr:指针结构type+encoding选择出的结构体的指针

void hsetCommand(client *c) {
    int i, created = 0;
    robj *o;
    // 如果参数数量是奇数,那么命令输入错误,这是hset的格式:hset key field value
    if ((c->argc % 2) == 1) {
        addReplyErrorFormat(c,"wrong number of arguments for '%s' command",c->cmd->name);
        return;
    }
    // 查询传入的key对应的散列对象的robj,如果不存在,则创建一个合适的散列对象并返回
    // 我们提前讲:如果该key不存在,那么会创建一个ziplist,他其实是个char数组的指针
    // 然后用robj对ziplist做封装
    if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
    hashTypeTryConversion(o,c->argv,2,c->argc-1);

    for (i = 2; i < c->argc; i += 2)
        created += !hashTypeSet(o,c->argv[i]->ptr,c->argv[i+1]->ptr,HASH_SET_COPY);

    /* HMSET (deprecated) and HSET return value is different. */
    char *cmdname = c->argv[0]->ptr;
    if (cmdname[1] == 's' || cmdname[1] == 'S') {
        /* HSET */
        addReplyLongLong(c, created);
    } else {
        /* HMSET */
        addReply(c, shared.ok);
    }
    signalModifiedKey(c,c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_HASH,"hset",c->argv[1],c->db->id);
    server.dirty += (c->argc - 2)/2;
}

3.1.1 hashTypeLookupWriteOrCreate

我们先看第一个内部函数,就是根据 hset 传入的 key 查得对应的哈希散列

robj *hashTypeLookupWriteOrCreate(client *c, robj *key) {
    // 根据传入的key【robj】查回对应的val【robj】,传入的key就是map的key
    robj *o = lookupKeyWrite(c->db,key);
    // 由于key对应的应该是一个散列,所以这里强制匹配,即robj.type == HASH
    if (checkType(c,o,OBJ_HASH)) return NULL;
    // 如果返回的robj是NULL,则说明不存在该散列,那么就创建一个散列。,否则直接返回
    if (o == NULL) {
        // 当在db里没有查到这个key,那么创建一个map对象,默认初始化的,都是ziplist编码的
        o = createHashObject();
        // 将新创建的robj以及key,设置到db中,key用于定位插槽【slot】的
        // 而o,也就是新创建的这个map对象,会和key一起,再组成一个dictEntry对象
        // 存储到db这个大dict中,db.dict[0]
        dbAdd(c->db,key,o);
    }
    return o;
}

里面比较核心的方法是 createHashObject中,这个方法只生成一个空的map对象【robj】,不会向里面设置任何键值对,这里会涉及到散列的编码选择:

robj *createHashObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_HASH, zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}
/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
    unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE;
    unsigned char *zl = zmalloc(bytes);
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    ZIPLIST_LENGTH(zl) = 0;
    zl[bytes-1] = ZIP_END;
    return zl;
}

可以看到,默认redis初始化的map,就是 ziplist,然后等真正插入键值对时,再看是否需要转换为哈希。

然后,由于可能是新生成了一个map对象,所以还涉及将其挂到db上的动作:

void dbAdd(redisDb *db, robj *key, robj *val) {
    // key是hset命令的key,val则是刚生成的map对象【robj】
    sds copy = sdsdup(key->ptr);
    // 该函数内部会讲key+val构造为一个dictEntry,挂到db.dict.ht[0]中
    // 按照顺序看,DB -> dict 散列 -> dictha 哈希 -> dictEntry 
    int retval = dictAdd(db->dict, copy, val);

    serverAssertWithInfo(NULL,key,retval == DICT_OK);
    signalKeyAsReady(db, key, val->type);
    if (server.cluster_enabled) slotToKeyAdd(key->ptr);
}

这部分代码其实与前面讲的 dictAdd 是基本相同的实现,毕竟db也就是个散列

3.1.2 hashTypeTryConversion

然后我们来看第二个函数 hashTypeTryConversion ,简单的将就是判断插入新的值后,是否可以将原本是 ziplist 存储的散列结构,转换为哈希存储的散列。

散列支持两种编码,即 ziplist 和 hash,当满足下面两个条件时,redis会使用ziplist:

  1. key-value 结构的所有键值对的字符串长度都小于 hash-max-ziplist-value(默认值64),可以通过配置文件修改
  2. 散列对象保存的键值对的个数(1个键值对记为1)小于 hash-max-ziplist-entries(默认值512),可以通过配置文件修改

redis 只支持 ziplist 到 hash 的转换,不支持反向。

3.1.3 hashTypeSet

然后我们来看第三个函数 hashTypeSet ,他就是插入键值对的。

#define HASH_SET_TAKE_FIELD (1<<0)
#define HASH_SET_TAKE_VALUE (1<<1)
#define HASH_SET_COPY 0
int hashTypeSet(robj *o, sds field, sds value, int flags) {
    int update = 0;
    // 如果值【robj】的编码方式是 ziplist 时
    if (o->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl, *fptr, *vptr;

        zl = o->ptr;
        fptr = ziplistIndex(zl, ZIPLIST_HEAD);
        if (fptr != NULL) {
            fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
            if (fptr != NULL) {
                /* Grab pointer to the value (fptr points to the field) */
                vptr = ziplistNext(zl, fptr);
                serverAssert(vptr != NULL);
                update = 1;

                /* Replace value */
                zl = ziplistReplace(zl, vptr, (unsigned char*)value,
                        sdslen(value));
            }
        }

        if (!update) {
            /* Push new field/value pair onto the tail of the ziplist */
            zl = ziplistPush(zl, (unsigned char*)field, sdslen(field),
                    ZIPLIST_TAIL);
            zl = ziplistPush(zl, (unsigned char*)value, sdslen(value),
                    ZIPLIST_TAIL);
        }
        o->ptr = zl;

        /* Check if the ziplist needs to be converted to a hash table */
        if (hashTypeLength(o) > server.hash_max_ziplist_entries)
            hashTypeConvert(o, OBJ_ENCODING_HT);
    } else if (o->encoding == OBJ_ENCODING_HT) {
        // 如果值【robj】的编码方式为哈希时,传入命令里的 field ,查回匹配的entry
        // 如果没匹配,就设置NULL,需要注意,dictEntry 保存了键【key】和值【val】
        dictEntry *de = dictFind(o->ptr,field);
        if (de) {
            // 如果map里有该field,则进入这段代码,首先获取目前field对应的值并释放内存
            sdsfree(dictGetVal(de));
            if (flags & HASH_SET_TAKE_VALUE) {
                // 然后将新的val设置到指针上,dictGetVal(de)返回的是指针
                dictGetVal(de) = value;
                value = NULL;
            } else {
                // 这个是将value复制了一遍,然后设置到val指针上
                dictGetVal(de) = sdsdup(value);
            }
            update = 1;
        } else {
            // 如果map中未查得field
            sds f,v;
            // 构建field结构体
            if (flags & HASH_SET_TAKE_FIELD) {
                f = field;
                field = NULL;
            } else {
                f = sdsdup(field);
            }
            // 构建value结构体
            if (flags & HASH_SET_TAKE_VALUE) {
                v = value;
                value = NULL;
            } else {
                v = sdsdup(value);
            }
            // 将field-value设置到map中
            dictAdd(o->ptr,f,v);
        }
    } else {
        serverPanic("Unknown hash encoding");
    }

    /* Free SDS strings we did not referenced elsewhere if the flags
     * want this function to be responsible. */
    if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field);
    if (flags & HASH_SET_TAKE_VALUE && value) sdsfree(value);
    return update;
}

这里的 dictAdd 负责将键值设到 map 上

int dictAdd(dict *d, void *key, void *val)
{   // 设置entry,如果entry已存在,这里返回NULL
    dictEntry *entry = dictAddRaw(d,key,NULL);
    if (!entry) return DICT_ERR;
    // 设置值
    dictSetVal(d, entry, val);
    return DICT_OK;
}

dictAddRaw 此函数添加条目,但不是设置值,而是向用户返回 dictEntry 结构,这将确保按用户的意愿填充值字段。

这个函数也直接暴露给用户API,主要是为了在哈希值中存储非指针,前面我看到过, dictEntry 里有个联合体,即可以放 *val,也可以放数字等。

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;
}

3.1.4 signalModifiedKey

我们来看第四个函数,signalModifiedKey ,这个函数是后置处理,涉及事物 watck key 和失效。

void signalModifiedKey(client *c, redisDb *db, robj *key) {
    touchWatchedKey(db,key);
    trackingInvalidateKey(c,key,1);
}
/* "Touch" a key, so that if this key is being WATCHed by some client the
 * next EXEC will fail. */
// watch key 检查,这样客户端在执行 exec 时就失败
void touchWatchedKey(redisDb *db, robj *key) {
    list *clients;
    listIter li;
    listNode *ln;
    // 如果db的watched_keys(它也是一个散列)为空,则返回,没有事物在watch
    if (dictSize(db->watched_keys) == 0) return;
    // 查询watch了该key的客户端,这里返回的是个list,这里查询很简单,就是哈希取模再比较key
    clients = dictFetchValue(db->watched_keys, key);
    if (!clients) return;
    // 将所有watch了该key的客户端的flag标签,都或等 CLIENT_DIRTY_CAS
    /* Mark all the clients watching this key as CLIENT_DIRTY_CAS */
    /* Check if we are already watching for this key */
    listRewind(clients,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        c->flags |= CLIENT_DIRTY_CAS;
    }
}
// 我也不知道这个函数是干嘛的
void trackingInvalidateKey(client *c, robj *keyobj, int bcast) {
    if (TrackingTable == NULL) return;

    unsigned char *key = (unsigned char*)keyobj->ptr;
    size_t keylen = sdslen(keyobj->ptr);

    if (bcast && raxSize(PrefixTable) > 0)
        trackingRememberKeyToBroadcast(c,(char *)key,keylen);

    rax *ids = raxFind(TrackingTable,key,keylen);
    if (ids == raxNotFound) return;

    raxIterator ri;
    raxStart(&ri,ids);
    raxSeek(&ri,"^",NULL,0);
    while(raxNext(&ri)) {
        uint64_t id;
        memcpy(&id,ri.key,sizeof(id));
        client *target = lookupClientByID(id);
        /* Note that if the client is in BCAST mode, we don't want to
         * send invalidation messages that were pending in the case
         * previously the client was not in BCAST mode. This can happen if
         * TRACKING is enabled normally, and then the client switches to
         * BCAST mode. */
        if (target == NULL ||
            !(target->flags & CLIENT_TRACKING)||
            target->flags & CLIENT_TRACKING_BCAST)
        {
            continue;
        }

        /* If the client enabled the NOLOOP mode, don't send notifications
         * about keys changed by the client itself. */
        if (target->flags & CLIENT_TRACKING_NOLOOP &&
            target == server.current_client)
        {
            continue;
        }

        /* If target is current client and it's executing a command, we need schedule key invalidation.
         * As the invalidation messages may be interleaved with command
         * response and should after command response. */
        if (target == server.current_client && server.fixed_time_expire) {
            incrRefCount(keyobj);
            listAddNodeTail(server.tracking_pending_keys, keyobj);
        } else {
            sendTrackingMessage(target,(char *)keyobj->ptr,sdslen(keyobj->ptr),0);
        }
    }
    raxStop(&ri);

    /* Free the tracking table: we'll create the radix tree and populate it
     * again if more keys will be modified in this caching slot. */
    TrackingTableTotalItems -= raxSize(ids);
    raxFree(ids);
    raxRemove(TrackingTable,(unsigned char*)key,keylen,NULL);
}

3.2 hlen 命令

hlen 命令是由下面这个函数来处理的

void hlenCommand(client *c) {
    robj *o;
    // 这行很简单,就是从数据库里查询key对应的robj对象,其中,因为这map结构的方法,
    // 所以检查robj头里的type是否是HASH,否则命令就用错了
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,OBJ_HASH)) return;
    // hashTypeLength 返回一个长度,然后addReplyLongLong将结果写到buf中
    addReplyLongLong(c,hashTypeLength(o));
}

所以核心的方法都是在下面这个函数内:

/* Return the number of elements in a hash. */
unsigned long hashTypeLength(const robj *o) {
    unsigned long length = ULONG_MAX;
    // 如果是ziplist编码的,那么这里有优点特殊了
    // 当 zllen 小于65535时,那么ziplist的len=zllen
    // 当 zllen 等于 65535,那么就得一个个算,直到END标记处
    if (o->encoding == OBJ_ENCODING_ZIPLIST) {
        // 这里使用ziplistLen函数,该函数就是直接拿ziplist数据结构的zllen字段
        // 这里除以2,是因为这是保存的键值对,是成对出现的
        length = ziplistLen(o->ptr) / 2;
    } else if (o->encoding == OBJ_ENCODING_HT) {
        // 如果时hash编码,其dictht里的used字段记录了当前有多少键值对
        length = dictSize((const dict*)o->ptr);
    } else {
        serverPanic("Unknown hash encoding");
    }
    return length;
}

下面引用ziplist的内存布局图

area        |<---- ziplist header ---->|<----------- entries ------------->|<-end->|

size          4 bytes  4 bytes  2 bytes    ?        ?        ?        ?     1 byte
            +---------+--------+-------+--------+--------+--------+--------+-------+
component   | zlbytes | zltail | zllen | entry1 | entry2 |  ...   | entryN | zlend |
            +---------+--------+-------+--------+--------+--------+--------+-------+
                                       ^                          ^        ^
address                                |                          |        |
                                ZIPLIST_ENTRY_HEAD                |   ZIPLIST_ENTRY_END
                                                                  |
                                                         ZIPLIST_ENTRY_TAIL

长度/类型域的值
zlbytesuint32_t整个 ziplist 占用的内存字节数,对 ziplist 进行内存重分配,或者计算末端时使用。
zltailuint32_t到达 ziplist 表尾节点的偏移量。 通过这个偏移量,可以在不遍历整个 ziplist 的前提下,弹出表尾节点。
zllenuint16_tziplist 中节点的数量。 当这个值小于 UINT16_MAX (65535)时,这个值就是 ziplist 中节点的数量; 当这个值等于 UINT16_MAX 时,节点的数量需要遍历整个 ziplist 才能计算得出。
entryX?ziplist 所保存的节点,各个节点的长度根据内容而定。
zlenduint8_t255 的二进制值 1111 1111 (UINT8_MAX) ,用于标记 ziplist 的末端。

3.3 增量rehash

增量式rehash是redis时间事件处理函数中的一部分:

/* Rehash */
if (server.activerehashing) {
    // dbs_per_call 这个值是min(16,你设置的数据库数量)
    for (j = 0; j < dbs_per_call; j++) {
        // 增量式rehash
        int work_done = incrementallyRehash(rehash_db);
        // work_done 标识该在rehash过程中是否真的干活了
        if (work_done) {
            /* If the function did some work, stop here, we'll do
             * more at the next cron loop. */
            break;
        } else {
            /* If this db didn't need rehash, we'll try the next one. */
            // 如果该库没有做任何工作,那说明该库无需rehash,那++处理下一个数据
            // 这里取了个模,好实现环型处理,0,1,2,3...15,0,1,2...
            rehash_db++;
            rehash_db %= server.dbnum;
        }
    }
}

我们主要关注下面这个增量式rehash核心方法:

int incrementallyRehash(int dbid) {
    /* Keys dictionary */
    // 如果键空间在rehash,那么就执行1ms的
    if (dictIsRehashing(server.db[dbid].dict)) {
        dictRehashMilliseconds(server.db[dbid].dict,1);
        return 1; /* already used our millisecond for this loop... */
    }
    /* Expires */
    // 如果键空间没有rehash,则检查失效空间是否在rehash,是的话也执行1ms的
    if (dictIsRehashing(server.db[dbid].expires)) {
        dictRehashMilliseconds(server.db[dbid].expires,1);
        return 1; /* already used our millisecond for this loop... */
    }
    return 0;
}

如果检测到该db处于散列中,那么就调用下面这个方法,执行1ms的任务:

/* Rehash in ms+"delta" milliseconds. The value of "delta" is larger 
 * than 0, and is smaller than 1 in most cases. The exact upper bound 
 * depends on the running time of dictRehash(d,100).*/
int dictRehashMilliseconds(dict *d, int ms) {
    // 如果 dict 的 pauserehash>0,那么就暂停rehash
    if (d->pauserehash > 0) return 0;
    long long start = timeInMilliseconds();
    int rehashes = 0;
    // while循环,调用 dictRehash,每次执行100任务量,执行完后就计算下剩余时间,不够就结束
    while(dictRehash(d,100)) {
        rehashes += 100;
        if (timeInMilliseconds()-start > ms) break;
    }
    return rehashes;
}

核心方法,rehash,我们送进来的n=100

int dictRehash(dict *d, int n) {
    // n * 100,这是最大允许访问的空插槽
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    unsigned long s0 = d->ht[0].size;
    unsigned long s1 = d->ht[1].size;
    if (dict_can_resize == DICT_RESIZE_FORBID || !dictIsRehashing(d)) return 0;
    if (dict_can_resize == DICT_RESIZE_AVOID && 
        ((s1 > s0 && s1 / s0 < dict_force_resize_ratio) ||
         (s1 < s0 && s0 / s1 < dict_force_resize_ratio)))
    {
        return 0;
    }
    // while循环,迭代次数最大为n,同时确保ht[0]剩余简键值对>0
    while(n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        // 需要注意的是rehashidx是插槽索引,简单说就是dictht结构体数组指针这个数组的索引
        while(d->ht[0].table[d->rehashidx] == NULL) {
            // 当插槽位置为NULL,那么rehashidx++
            d->rehashidx++;
            // 如果空插槽碰到的太多,超过100*n,也不做了,怕时间花销太大
            if (--empty_visits == 0) return 1;
        }
        // 取出rehashidx处的头部dictEntry
        de = d->ht[0].table[d->rehashidx];
        // 将该插槽下所有的dictEntry迁移到ht[1]中
        while(de) {
            uint64_t h;
            nextde = de->next;
            // 从dictEntry中取出key,然后与ht[1]的sizemask做取模得到索引
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            // 这里怕迷糊,de是当前entry,nextde是下一个entry
            // 首先将de设置到当前entry的next指针设置为ht[1]中插槽头entry
            de->next = d->ht[1].table[h];
            // 然后将ht[1]的头entry换成当前entry
            d->ht[1].table[h] = de;
            // used 统计计算
            d->ht[0].used--;
            d->ht[1].used++;
            // 然后重用de变量,存储下一个entry,然后循环起来,直刀最后一个冲突键为止
            de = nextde;
        }
        // 将当前rehashidx位置设置NULL(毕竟都迁移完了),然后 rehashidx++
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }
    // 如果整个表都被rehash完了,那么释放内存,然后ht[0]和ht[1]对掉,并下掉rehashidx标志
    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;
}

3.4 单步rehash

其实,除了定时时间任务里的增量rehash之外,在正常命令处理时,也会随时做一点rehash,我称呼它为单步rehahs,主要是为了贴合它起的名字:

/* This function performs just a step of rehashing, and only if hashing has
 * not been paused for our hash table. When we have iterators in the
 * middle of a rehashing we can't mess with the two hash tables otherwise
 * some element can be missed or duplicated.
 *
 * This function is called by common lookup or update operations in the
 * dictionary so that the hash table automatically migrates from H1 to H2
 * while it is actively used. */
static void _dictRehashStep(dict *d) {
    if (d->pauserehash == 0) dictRehash(d,1);
}

该函数与前面增量rehahs不同的点在于,他n 不是100次,而是 1次。除此之外,没啥任何区别。

我们下面大概列举一下会出发单步rehash的场景:

  • 向db中插入新的键值对
  • 向db删除一个键
  • 向db发起查询操作

3.5 迭代器

Jababoy 应该比较数据,首先就是获取迭代器,Redis 中有好多种 iterator,我们首先介绍散列类型的,就是db内的某个map上的hkeys

hashTypeIterator *hashTypeInitIterator(robj *subject) {
    // 函数传入的subject,就是我们的迭代目标,比方我们这里时在处理散列结构
    // 那么送入的就可能是ziplist,也可能是hash编码的
    hashTypeIterator *hi = zmalloc(sizeof(hashTypeIterator));
    hi->subject = subject;
    hi->encoding = subject->encoding;
    
    if (hi->encoding == OBJ_ENCODING_ZIPLIST) {
        // 如果是ziplist编码
        hi->fptr = NULL;
        hi->vptr = NULL;
    } else if (hi->encoding == OBJ_ENCODING_HT) {
        // 如果是hash编码
        hi->di = dictGetIterator(subject->ptr);
    } else {
        serverPanic("Unknown hash encoding");
    }
    return hi;
}
dictIterator *dictGetIterator(dict *d)
{
    // 初始化一个dictIterator结构
    dictIterator *iter = zmalloc(sizeof(*iter));

    iter->d = d;
    iter->table = 0;
    iter->index = -1;
    iter->safe = 0;
    iter->entry = NULL;
    iter->nextEntry = NULL;
    return iter;
}

我们看一下迭代器的数据结构:hashTypeIterator 内部嵌套了一个 dictIterator结构,原因是因为map有两种编码,所以为了给使用者一个公共API,就出现了hashTypeIterator,由它来负责调度是走ziplist的迭代器,还是hash的迭代器。

typedef struct {
    robj *subject;
    int encoding;

    unsigned char *fptr, *vptr;

    dictIterator *di;
    dictEntry *de;
} hashTypeIterator;
typedef struct dictIterator {
    dict *d;
    long index;
    int table, safe;
    dictEntry *entry, *nextEntry;
    /* unsafe iterator fingerprint for misuse detection. */
    unsigned long long fingerprint;
} dictIterator;

初始化完,我们继续看下一个函数 hashTypeNext,对应 Java 的就是 next() 方法

int hashTypeNext(hashTypeIterator *hi) {
    // 如果是ziplist编码的迭代器
    if (hi->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl;
        unsigned char *fptr, *vptr;
        // hi 是迭代器,subject是ziplist的robj,用ptr可以直接指针到ziplist
        zl = hi->subject->ptr;
        // 下面这俩玩意吃实话时就是NULL
        fptr = hi->fptr;
        vptr = hi->vptr;
        
        if (fptr == NULL) {
            // 初始化cursor
            serverAssert(vptr == NULL);
            fptr = ziplistIndex(zl, 0);
        } else {
            /* Advance cursor */
            serverAssert(vptr != NULL);
            fptr = ziplistNext(zl, vptr);
        }
        if (fptr == NULL) return C_ERR;

        /* Grab pointer to the value (fptr points to the field) */
        vptr = ziplistNext(zl, fptr);
        serverAssert(vptr != NULL);

        /* fptr, vptr now point to the first or next pair */
        // 
        hi->fptr = fptr;
        hi->vptr = vptr;
    } else if (hi->encoding == OBJ_ENCODING_HT) {
        // 如果是哈希编码的迭代器
        if ((hi->de = dictNext(hi->di)) == NULL) return C_ERR;
    } else {
        serverPanic("Unknown hash encoding");
    }
    return C_OK;
}

由于ziplist本身就是有序的,它是按照插入顺序存储的,所以如果是ziplis编码的话,迭代器就是直接使用了ziplist本身的特性进行迭代。

而对于hash编码的case,我们需要重点研究dictNex函数的实现:

dictEntry *dictNext(dictIterator *iter)
{
    while (1) {
        if (iter->entry == NULL) {
            // 初始化时,iter->table 就是0,表示选择ht[0]开始
            dictht *ht = &iter->d->ht[iter->table];
            // 其实下面这俩判断,就是index和table的初始值,当首次执行时,会根据
            // 迭代饿模式,来区别设置,如果是安全模式,则设置dict的pasuerRehash,
            // 如果是非安全模式,则计算一个词是dict各个状态字段汇聚出来的一个指纹哈希值
            if (iter->index == -1 && iter->table == 0) {
                if (iter->safe)
                    dictPauseRehashing(iter->d);
                else
                    iter->fingerprint = dictFingerprint(iter->d);
            }
            // index++,开始扫描哈希,它就像rehash里的rehashidx
            iter->index++;
            // 这个条件是判断是否当前index已经超过当前扫描的dictht的size最大插槽数了
            if (iter->index >= (long) ht->size) {
                // 当dict处于rehash中,且现在是扫描0号表,那么就切换到1号表,index归零
                if (dictIsRehashing(iter->d) && iter->table == 0) {
                    iter->table++;
                    iter->index = 0;
                    ht = &iter->d->ht[1];
                } else {
                    break;
                }
            }
            // 将该插槽处的dictEntry设置到entry上,然后
            iter->entry = ht->table[iter->index];
        } else {
            // 这种case是因为之前设置过entry,这里又进入循环了
            iter->entry = iter->nextEntry;
        }
        //  如果entry非null,则设置好nextEntry,就可以返回了
        //  如果entry位null,会进入while循环,继续查找
        if (iter->entry) {
            /* We need to save the 'next' here, the iterator user
             * may delete the entry we are returning. */
            iter->nextEntry = iter->entry->next;
            return iter->entry;
        }
    }
    return NULL;
}

我们需要理解 dictIterator 的内部字段的含义:

  • d 就是本迭代的 dict 对象【robj】
  • entry 当前迭代到的元素
  • nextEntry 下一个元素
  • safe 是否安全模式迭代,安全模式会暂停rehash
  • fingerprint 指纹,当不是安全模式时,redis将dict里的used,size等字段进行哈希计算,得到一个指纹

我理解这个迭代器很容易理解,线循环哈希插槽,然后在插槽内再循环链表。


原文地址:https://blog.csdn.net/qq_31179577/article/details/140463899

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!