Redis源码剖析之对象

以下涉及到的源码均为redis5.0-rc3版本的代码【点击查看官方源码】

redis对象

在redis中，定义了5中基本对象，分别为string、list、set、zset、hash。而为了方便管理与操作，redis又对这5种对象进行了一次外围封装，如下所示（server.h头文件）：

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;    /* 指向对象的指针 */
} robj;

由此可见，是通过redisObject这个结构来统一对象的入口，由type字段来标识当前对象的本质（string、list、set、zset、hash），并由encoding字段来标识对象的底层具体实现数据结构，refcount字段则是用于引用计数，使得对象可以共享。
redis对象类型的宏定义如下所示：

/* Redis 对象 */
//redis实际的对象
#define OBJ_STRING 0    /* 字符串对象 */
#define OBJ_LIST 1      /* 列表对象 */
#define OBJ_SET 2       /* 集合对象 */
#define OBJ_ZSET 3      /* 有序集合对象 */
#define OBJ_HASH 4      /* 哈希对象 */
/* The "module" object type is a special one that signals that the object
 * is one directly managed by a Redis module. In this case the value points
 * to a moduleValue struct, which contains the object value (which is only
 * handled by the module itself) and the RedisModuleType struct which lists
 * function pointers in order to serialize, deserialize, AOF-rewrite and
 * free the object.
 *
 * Inside the RDB file, module types are encoded as OBJ_MODULE followed
 * by a 64 bit module type ID, which has a 54 bits module-specific signature
 * in order to dispatch the loading to the right module, plus a 10 bits
 * encoding version. */
#define OBJ_MODULE 5    /* Module object*/
#define OBJ_STREAM 6    /* Stream object. */

redis编码类型的宏定义如下所示：

/* 对象编码 */

#define OBJ_ENCODING_RAW 0     /* 原始的简单动态字符串 */
#define OBJ_ENCODING_INT 1     /* 整型 */
#define OBJ_ENCODING_HT 2      /* 哈希表 */
#define OBJ_ENCODING_ZIPMAP 3  /* 压缩图 */
#define OBJ_ENCODING_LINKEDLIST 4 /* No longer used: old list encoding. */
#define OBJ_ENCODING_ZIPLIST 5 /* 压缩列表 */
#define OBJ_ENCODING_INTSET 6  /* 整型集合*/
#define OBJ_ENCODING_SKIPLIST 7  /* 跳跃表 */
#define OBJ_ENCODING_EMBSTR 8  /* embstr编码的简单动态字符串 */
#define OBJ_ENCODING_QUICKLIST 9 /* 链式压缩列表编码 */
#define OBJ_ENCODING_STREAM 10 /* 编码为listpacks的基数树 */

redis对象与基本对象及其底层编码实现之间的关系可以简单得用一个图来简略的表述，如下所示：

特点

1. 引用技术和对象共享。从redisObject的结构中可以看出，有一个refcount字段，这个字段主要是用于对象的计数，从而达到对象的共享，以及对象释放。
2. 命令的类型检查。由于有些命令只能在部分结构上使用，而通过redisObject对象来操作的时候便需要进行区分，这时候type字段就有用了，它标识了具体的对象，便能表明命令是否可执行。
3. 命令多态。因redis的基本对象都不止由一种类型编码实现，所以在执行一个操作的时候，都存在多态的情况，这时encoding字段的重要性就凸显了。

源码解读

这里粗略的解读一下redis对象操作的源码，以此来说明以下几个点：

1. 对象的创建
2. 类型检查
3. 多态
4. 引用技术与对象共享
5. 内存释放

（注：如下相关的代码如非特定说明则均存在object.c文件中）

对象创建

通过查看此部分代码，可以知道每种基本对象的底层数据结构编码情况。
创建对象的基本代码如下所示：

robj *createObject(int type, void *ptr) {
    robj *o = zmalloc(sizeof(*o));
    o->type = type;//标记类型
    o->encoding = OBJ_ENCODING_RAW; //标记编码
    o->ptr = ptr;   //赋值对象指针
    o->refcount = 1; //引用计数

    /* Set the LRU to the current lruclock (minutes resolution), or
     * alternatively the LFU counter. */
    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
    } else {
        o->lru = LRU_CLOCK();
    }
    return o;
}

创建String类型的对象

在创建字符串的时候用OBJ_ENCODING_EMBSTR_SIZE_LIMIT来标明区分具体的底层实现是sds或者emdstr类型。

/* raw类型编码 */
robj *createRawStringObject(const char *ptr, size_t len) {
    return createObject(OBJ_STRING, sdsnewlen(ptr,len));
}

/* embstr类型编码： Create a string object with encoding OBJ_ENCODING_EMBSTR, that is
 * an object where the sds string is actually an unmodifiable string
 * allocated in the same chunk as the object itself. */
robj *createEmbeddedStringObject(const char *ptr, size_t len) {
    robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr8)+len+1);
    struct sdshdr8 *sh = (void*)(o+1);

    o->type = OBJ_STRING;
    o->encoding = OBJ_ENCODING_EMBSTR;
    o->ptr = sh+1;
    o->refcount = 1;
    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
    } else {
        o->lru = LRU_CLOCK();
    }

    sh->len = len;
    sh->alloc = len;
    sh->flags = SDS_TYPE_8;
    if (ptr == SDS_NOINIT)
        sh->buf[len] = '\0';
    else if (ptr) {
        memcpy(sh->buf,ptr,len);
        sh->buf[len] = '\0';
    } else {
        memset(sh->buf,0,len+1);
    }
    return o;
}

/* Create a string object with EMBSTR encoding if it is smaller than
 * OBJ_ENCODING_EMBSTR_SIZE_LIMIT, otherwise the RAW encoding is
 * used.
 *
 * The current limit of 44 is chosen so that the biggest string object
 * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
 //embstr类型字符串的字节限定数宏定义
#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44
robj *createStringObject(const char *ptr, size_t len) {
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)
        return createEmbeddedStringObject(ptr,len);
    else
        return createRawStringObject(ptr,len);
}

robj *createStringObjectFromLongLong(long long value) {
    robj *o;
    if (value >= 0 && value < OBJ_SHARED_INTEGERS) {
        incrRefCount(shared.integers[value]);
        o = shared.integers[value];
    } else {
        if (value >= LONG_MIN && value <= LONG_MAX) {
            o = createObject(OBJ_STRING, NULL);
            o->encoding = OBJ_ENCODING_INT;
            o->ptr = (void*)((long)value);
        } else {
            o = createObject(OBJ_STRING,sdsfromlonglong(value));
        }
    }
    return o;
}

/* Create a string object from a long double. If humanfriendly is non-zero
 * it does not use exponential format and trims trailing zeroes at the end,
 * however this results in loss of precision. Otherwise exp format is used
 * and the output of snprintf() is not modified.
 *
 * The 'humanfriendly' option is used for INCRBYFLOAT and HINCRBYFLOAT. */
robj *createStringObjectFromLongDouble(long double value, int humanfriendly) {
    char buf[MAX_LONG_DOUBLE_CHARS];
    int len = ld2string(buf,sizeof(buf),value,humanfriendly);
    return createStringObject(buf,len);
}

/* Duplicate a string object, with the guarantee that the returned object
 * has the same encoding as the original one.
 *
 * This function also guarantees that duplicating a small integere object
 * (or a string object that contains a representation of a small integer)
 * will always result in a fresh object that is unshared (refcount == 1).
 *
 * The resulting object always has refcount set to 1. */
robj *dupStringObject(const robj *o) {
    robj *d;

    serverAssert(o->type == OBJ_STRING);

    switch(o->encoding) {
    case OBJ_ENCODING_RAW:
        return createRawStringObject(o->ptr,sdslen(o->ptr));
    case OBJ_ENCODING_EMBSTR:
        return createEmbeddedStringObject(o->ptr,sdslen(o->ptr));
    case OBJ_ENCODING_INT:
        d = createObject(OBJ_STRING, NULL);
        d->encoding = OBJ_ENCODING_INT;
        d->ptr = o->ptr;
        return d;
    default:
        serverPanic("Wrong encoding.");
        break;
    }
}

创建List类型的对象

/* quicklist类型编码 */
robj *createQuicklistObject(void) {
    quicklist *l = quicklistCreate();
    robj *o = createObject(OBJ_LIST,l);
    o->encoding = OBJ_ENCODING_QUICKLIST;
    return o;
}
/* ziplist类型编码 */
robj *createZiplistObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_LIST,zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

创建Set类型的对象

/* ht类型编码 */
robj *createSetObject(void) {
    dict *d = dictCreate(&setDictType,NULL);
    robj *o = createObject(OBJ_SET,d);
    o->encoding = OBJ_ENCODING_HT;
    return o;
}
/* intset类型编码 */
robj *createIntsetObject(void) {
    intset *is = intsetNew();
    robj *o = createObject(OBJ_SET,is);
    o->encoding = OBJ_ENCODING_INTSET;
    return o;
}

创建ZSet类型的对象

/* skiplist类型编码 */
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;
}
/* ziplist类型编码 */
robj *createZsetZiplistObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_ZSET,zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

创建Hash类型的对象

/* ziplist类型编码 */
robj *createHashObject(void) {
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_HASH, zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

创建Module类型的对象

robj *createModuleObject(moduleType *mt, void *value) {
    moduleValue *mv = zmalloc(sizeof(*mv));
    mv->type = mt;
    mv->value = value;
    return createObject(OBJ_MODULE,mv);
}

创建Stream类型的对象

robj *createStreamObject(void) {
    stream *s = streamNew();
    robj *o = createObject(OBJ_STREAM,s);
    o->encoding = OBJ_ENCODING_STREAM;
    return o;
}

内存释放

对象的释放体现了引用计数的重要性，也体现了对象共享的问题。如下时各对象的释放操作。这里我们重点关注decrRefCount操作函数，其通过判断一个对象是否仍然被共享，如果没有就进行内存释放。

void decrRefCount(robj *o) {
    if (o->refcount == 1) {
        switch(o->type) {
        case OBJ_STRING: freeStringObject(o); break;
        case OBJ_LIST: freeListObject(o); break;
        case OBJ_SET: freeSetObject(o); break;
        case OBJ_ZSET: freeZsetObject(o); break;
        case OBJ_HASH: freeHashObject(o); break;
        case OBJ_MODULE: freeModuleObject(o); break;
        case OBJ_STREAM: freeStreamObject(o); break;
        default: serverPanic("Unknown object type"); break;
        }
        zfree(o);
    } else {
        if (o->refcount <= 0) serverPanic("decrRefCount against refcount <= 0");
        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;
    }
}

引用计数限定

redis对对象共享的引用计数进行了限定，在server.h头文件中宏定义了最大引用计数：

#define OBJ_SHARED_REFCOUNT INT_MAX

当一个对象的被引用数量没有到达最大值时，才可以被继续引用。如下所示：

void incrRefCount(robj *o) {
    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
}

对象共享

引用计数体现了对象共享，除了创建的对象的引用计数外，redis还默认初始化了一些共享的常量，仔细读过前面内容的会发现，在创建字符串对象中我贴出了如下的这个函数：

robj *createStringObjectFromLongLong(long long value) {
    robj *o;
    if (value >= 0 && value < OBJ_SHARED_INTEGERS) {
        //共享
        incrRefCount(shared.integers[value]);
        o = shared.integers[value];
    } else {
        if (value >= LONG_MIN && value <= LONG_MAX) {
            o = createObject(OBJ_STRING, NULL);
            o->encoding = OBJ_ENCODING_INT;
            o->ptr = (void*)((long)value);
        } else {
            o = createObject(OBJ_STRING,sdsfromlonglong(value));
        }
    }
    return o;
}

这个函数中就体现了redis默认初始化的共享量，当整数值小于OBJ_SHARED_INTEGERS常量时，则直接进行共享引用计数加1，而该常量在server.h头文件宏定义如下：

#define OBJ_SHARED_INTEGERS 10000

如下时redis默认初始化的共享量（server.h头文件中），其具体的初始化在createSharedObjects函数中实现（server.c文件中）。

struct sharedObjectsStruct {
    robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space,
    *colon, *nullbulk, *nullmultibulk, *queued,
    *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
    *outofrangeerr, *noscripterr, *loadingerr, *slowscripterr, *bgsaveerr,
    *masterdownerr, *roslaveerr, *execaborterr, *noautherr, *noreplicaserr,
    *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
    *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink,
    *rpop, *lpop, *lpush, *zpopmin, *zpopmax, *emptyscan,
    *select[PROTO_SHARED_SELECT_CMDS],
    *integers[OBJ_SHARED_INTEGERS],
    *mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */
    *bulkhdr[OBJ_SHARED_BULKHDR_LEN];  /* "$<value>\r\n" */
    sds minstring, maxstring;
};