Redis源码学习：Redis对象和5种数据类型的工作原理

Redis 提供 5 种基本数据类型：String（字符串）、List（列表）、Set（集合）、Hash（哈希）、Zset（有序集合），这些数据类型可以供用户直接使用。

RedisObject

RedisObject 是 Redis 中的一个数据结构，表示 Redis 中的所有数据类型（字符串、列表、集合、有序集合和哈希）的统一抽象。定义如下：

typedef struct redisObject {
	// 表示对象的类型，比如字符串、列表、集合等
    unsigned type:4;
    // 表示对象的编码方式，不同的编码方式在内存中的存储形式不同
    unsigned encoding:4;
    // 用于 LRU 或 LFU 过期策略
    unsigned lru:LRU_BITS;
    // 引用计数，用于内存管理
    int refcount;
    // 指向实际数据的指针
    void *ptr;
} robj;

5种数据类型及实现

String

字符串是最常用的 Redis 数据类型，可以存储普通文本或二进制数据。字符串的实现主要有三种编码方式：

• RAW：使用普通的动态字符串（SDS，Simple Dynamic String），存储上限时 512 mb（但是，你不能这么干）。
• EMBSTR：优化的 SDS 存储方式，用于小于等于 44 字节的字符串，此时 object head 与 SDS 是一段连续的空间。申请内存时只需要一次内存分片，效率更高。
• INT：用于表示可以用整数存储的字符串，且大小在 LONG_MAX 范围内，直接将数据保存在 RedisObject 的 ptr 指针位置（8字节），不再需要 SDS 了。

List

列表是一种有序的字符串链表，现在只使用 quicklist 实现 List。

在 object.c 文件中可以找到创建列表对象的代码：

robj *createQuicklistObject(void) {
    quicklist *l = quicklistCreate();
    robj *o = createObject(OBJ_LIST,l);
    o->encoding = OBJ_ENCODING_QUICKLIST;
    return o;
}

在插入元素时，使用的也是 quicklist：

void listTypePush(robj *subject, robj *value, int where) {
    if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
        int pos = (where == LIST_HEAD) ? QUICKLIST_HEAD : QUICKLIST_TAIL;
        if (value->encoding == OBJ_ENCODING_INT) {
            char buf[32];
            ll2string(buf, 32, (long)value->ptr);
            quicklistPush(subject->ptr, buf, strlen(buf), pos);
        } else {
            quicklistPush(subject->ptr, value->ptr, sdslen(value->ptr), pos);
        }
    } else {
        serverPanic("Unknown list encoding");
    }
}

Set

集合是一组无序的字符串集合，集合中的元素是唯一的，查询效率高。集合的实现有两种主要的编码方式：

• INTSET：整数集合，用于存储数据都是整数，数量不超过 set-max-intset-entries
• HT：哈希表，用于存储字符串的大集合，dic 中的 key 存储元素，value 都是 null。

创建集合

在创建集合对象时，通过对 value 的类型判断，来选择不同的编码：

robj *setTypeCreate(sds value) {
	// 判断 value 是否是数值类型Long Long，如果是数值类型，采用 IntSet 编码
	// 否则，采用 HT 编码
    if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
        return createIntsetObject();
    return createSetObject();
}

选择编码方式的逻辑

int setTypeAdd(robj *subject, sds value) {
    long long llval;
    if (subject->encoding == OBJ_ENCODING_HT) {
    // 已经是 HT 编码，直接添加元素
        dict *ht = subject->ptr;
        dictEntry *de = dictAddRaw(ht,value,NULL);
        if (de) {
            dictSetKey(ht,de,sdsdup(value));
            dictSetVal(ht,de,NULL);
            return 1;
        }
    } else if (subject->encoding == OBJ_ENCODING_INTSET) {
        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
        // 目前是 IntSet
        // 判断 value 是否是整数
            uint8_t success = 0;
            // 是整数，直接添加元素到 set
            subject->ptr = intsetAdd(subject->ptr,llval,&success);
            if (success) {
            // 当intset 元素数量超过 set_max_intset_entries，则转为 HT
                /* Convert to regular set when the intset contains
                 * too many entries. */
                size_t max_entries = server.set_max_intset_entries;
                /* limit to 1G entries due to intset internals. */
                if (max_entries >= 1<<30) max_entries = 1<<30;
                if (intsetLen(subject->ptr) > max_entries)
                    setTypeConvert(subject,OBJ_ENCODING_HT);
                return 1;
            }
        } else {
	        // 不是整数，直接转为 HT
            /* Failed to get integer from object, convert to regular set. */
            setTypeConvert(subject,OBJ_ENCODING_HT);

            /* The set *was* an intset and this value is not integer
             * encodable, so dictAdd should always work. */
            serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
            return 1;
        }
    } else {
        serverPanic("Unknown set encoding");
    }
    return 0;
}

Zset

有序集合（ZSet）的实现主要有两种编码方式：压缩列表（ziplist）和跳表（skiplist）。

定义

在 server.h 文件中，zset 结构体的定义如下：

typedef struct zset {
    dict *dict;
    zskiplist *zsl;
} zset;

创建集合

在 t_zset.c 文件中，可以找到创建有序集合对象的代码：

robj *createZsetObject(void) {
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;

    zs->dict = dictCreate(&zsetDictType, NULL);
    zs->zsl = zslCreate();
    o = createObject(OBJ_ZSET, zs);
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

当元素数量不多时，HT 和 Skiplist 的优势不明显，而且更耗内存。因此，在满足以下条件的情况下，Zset 会采用 Ziplist 来节省内存。

• 元素数量小于 zset-max-ziplist-entries，默认值 128
• 每个元素都小于 zset-max-ziplist-value 字节，默认值 64

// zadd 添加元素时，先根据 key 找到 zset，不存在则创建新的 zset
zobj = lookupKeyWrite(c->db,key);
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
if (zobj == NULL) {
	if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
	// zset_max_ziplist_entries 设置为0就是禁用了ziplist
	// 或者 value 大小超过了 zset_max_ziplist_value，采用 HT + Skiplist
	if (server.zset_max_ziplist_entries == 0 ||
		server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
	{
		zobj = createZsetObject();
	} else {
		// 否则，采用 ziplist
		zobj = createZsetZiplistObject();
	}
	dbAdd(c->db,key,zobj);
}

// 采用 ziplist 编码
robj *createZsetZiplistObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_ZSET, zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

选择编码方式的逻辑

在有序集合的插入操作中，Redis 会根据特定条件选择编码方式。在插入操作中，Redis 会检查 zobj 的编码类型，如果是 ZIPLIST 编码，会进行 ziplist 相关的操作；如果是 SKIPLIST 编码，会进行 skiplist 相关的操作。源码如下：

int zsetAdd(robj *zobj, double score, sds ele, int *incr, double *newscore) {
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        // 使用 ziplist 进行插入操作的代码
        // 如果 ziplist 超过一定大小，会转换为 skiplist
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
        // 使用 skiplist 进行插入操作的代码
    }
    return 1;
}

判断是否需要转换

在 t_zset.c 文件中的 zsetAdd 函数中，有一个逻辑会判断是否需要将 ziplist 转换为 skiplist：

if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
    // 如果 ziplist 的元素数量超过一定阈值，或者单个元素的长度超过一定阈值
    if (ziplistLen(zobj->ptr) > server.zset_max_ziplist_entries ||
        ziplistBlobLen(zobj->ptr) > server.zset_max_ziplist_value)
    {
        zsetConvert(zobj, OBJ_ENCODING_SKIPLIST);
    }
}

ziplist本身没有排序功能，而且没有键值对的概念，因此需要有 zset 通过编码实现：

• ziplist是连续内存，因此 score 和 element 是紧挨在一起的两个 entry，element 在前， score 在后
• score 越小越接近队首，score 越大越接近队尾，按照 score 值升序排列

Hash

Hash 底层采用的编码与 Zset 基本一致，只需要把排序相关的 Skiplist 去掉即可。

创建哈希

在 t_hash.c 文件中，可以找到创建哈希对象的代码：

robj *createHashObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_HASH, zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

Hash 结构默认采用 ziplist 编码，用来节省内存。 ziplist 中相邻的两个 entry 分别保存 field 和 value。

选择编码方式的逻辑

在 t_hash.c 文件中的 hashTypeSet 函数中，会判断是否需要将 ziplist 转换为 hashtable

void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {
    int i;
    size_t sum = 0;
	// 已经是 ziplist 编码了，就什么都不做
    if (o->encoding != OBJ_ENCODING_ZIPLIST) return;
	// 依次遍历命令种的 field、value
    for (i = start; i <= end; i++) {
        if (!sdsEncodedObject(argv[i]))
            continue;
        size_t len = sdslen(argv[i]->ptr);
        // 如果 field 或 value 超过了hash_max_ziplist_value，则转为 hashtable
        if (len > server.hash_max_ziplist_value) {
            hashTypeConvert(o, OBJ_ENCODING_HT);
            return;
        }
        sum += len;
    }
    // ziplist 大小超过 1g，也要转
    if (!ziplistSafeToAdd(o->ptr, sum))
        hashTypeConvert(o, OBJ_ENCODING_HT);
}

当满足4个条件中的任意一个时，需要转换：

• ziplist 的元素数量是否超过 server.hash_max_ziplist_entries（默认 512 ）
• ziplist 的大小超过 1G
• field 或 value 的长度是否超过 server.hash_max_ziplist_value（默认 64 字节）

总结

本文讲解了什么Redis对象，已经面向用户的5种常用数据类型的底层逻辑，希望对你有帮助。