Redis源码阅读(十) 字符串对象

1. 编码

字符串对象的编码可以是int、raw、embstr三种。

• int: 对象保存的是整数值，并且可以使用long类型进行表示，那么ptr中存储的就是这个long的地址，并将编码设置为int，[0, 10000)范围的可以使用共享整数。

• raw: 对象保存的是字符串，且长度大于44字节，使用raw编码。

• embstr: 对象保存的是字符串，若长度小等于44字节，则使用embstr类型编码，embstr是一种sds字符串，使用固定的SDS_TYPE_8进行存储，只进行一次内存分配，分配redis对象内存空间和sds对象内存空间，它们是连续存储的。

Redis中浮点类型作为字符串值来保存。

2. 命令

命令	作用
SET	设置指定 key 的值 NX:仅key不存在 XX: 仅key存在 EX: 过期秒数 PX: 过期毫秒数
GET	获取指定 key 的值。
SETNX	只有在 key 不存在时设置 key 的值
SETEX	设置key的值，并给定过期时间seconds
GETSET	将给定 key 的值设为 value ，并返回 key 的旧值(old value)
SETRANGE	用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始
GETRANGE	获取key 中字符串值的[start,end]的值
MSET	同时设置一个或多个 key-value 对
HGET	获取所有(一个或多个)给定 key 的值
INCR	将 key 中储存的数字值增一
INCRBY	将 key 所储存的值加上给定的增量值（increment）
INCRBYFLOAT	将 key 所储存的值加上给定的浮点增量值（increment）
DECR	将 key 中储存的数字值增一
DECRBY	key 所储存的值减去给定的减量值（decrement）
APPEND	追加value到key所存储字符串的后面，key不存在则添加
STRLEN	获取key所存储字符串的长度

• 相同命令不同的编码区分

命令	int 编码	embstr编码	raw编码
SET	底层使用整数值进行存储，[0,1000)使用共享整数存储	使用简单动态字符串存储，SDS_TYPE_8	使用简单动态字符串存储，长度超过44字节时使用
GET	拷贝底层存储的整数值，将其转换为字符串后返回给客户端	直接返回字符串给客户端	直接返回字符串给客户端
INCR	对整数值进行加法，计算结果仍能用在整数值范围，则直接存储，否则转换编码方式为RAW	取出字符串并尝试转换成long long类型，失败则告诉客户端，成功则创建新的字符串保存	取出字符串并尝试转换成long long类型，失败则告诉客户端，成功则创建新的字符串保存
INCRBYFLOAT	取出整数转换成long double，对齐进行浮点操作后，创建新的字符串对象保存起来	取出整数转换成long double，对齐进行浮点操作后，创建新的字符串对象保存起来	取出整数转换成long double，对齐进行浮点操作后，创建新的字符串对象保存起来
APPEND	调用 sdscatlen 函数，将字符串添加到末尾	调用 sdscatlen 函数，将字符串添加到末尾	调用 sdscatlen 函数，将字符串添加到末尾
STRLEN	直接获取字符串的长度	直接获取字符串的长度	直接获取字符串的长度

若对象进行了共享，那么修改对象的操作则会拷贝一个新的对象，而不是用原来的共享对象

3. 源码剖析

• 创建字符串对象

/**
 * 根据创建的字符串长度，判断是否使用OBJ_ENCODING_EMBSTR编码
 * 使用jemalloc分配内存，会分配2的幂次方字节，所以我们会分配64字节
 * redisOob大小16字节，sdshdr8大小3字节，加1字节结束标识，共20字节
 * 所以embstr的限制为64-20 = 44字节
 */
#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44

/**
 * 超过44字节使用RAW编码方式，否则使用EMBSTR编码
 */
robj *createStringObject(const char *ptr, size_t len) {
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)
        return createEmbeddedStringObject(ptr,len);
    else
        return createRawStringObject(ptr,len);
}

• 通过long long类型创建字符串对象

robj *createStringObjectFromLongLong(long long value) {
    robj *o;

    // [0, 10000) 范围内整数共享
    if (value >= 0 && value < OBJ_SHARED_INTEGERS) {
        // 添加整数的引用计数
        incrRefCount(shared.integers[value]);
        // 获取整数值
        o = shared.integers[value];
    } else { // 不在整数共享范围
        // [LONG_MIN, LONG_MAX], 使用字符串对象，编码使用OBJ_ENCODING_INT
        if (value >= LONG_MIN && value <= LONG_MAX) {
            o = createObject(OBJ_STRING, NULL);
            o->encoding = OBJ_ENCODING_INT;
            // 直接指向整数值
            o->ptr = (void*)((long)value);
        } else {
            // 不在[LONG_MIN,LONG_MAX] 范围则使用sds动态字符串存储, 编码OBJ_ENCODING_RAW
            o = createObject(OBJ_STRING,sdsfromlonglong(value));
        }
    }
    return o;
}

• 尝试对字符串对象进行编码优化

robj *tryObjectEncoding(robj *o) {
    long value;
    sds s = o->ptr;
    size_t len;

    /* Make sure this is a string object, the only type we encode
     * in this function. Other types use encoded memory efficient
     * representations but are handled by the commands implementing
     * the type. */
    serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);

    /* We try some specialized encoding only for objects that are
     * RAW or EMBSTR encoded, in other words objects that are still
     * in represented by an actually array of chars. */

    // 只有是RAW或EMBSTR编码是，才尝试重新编码
    if (!sdsEncodedObject(o)) return o;

    /* It's not safe to encode shared objects: shared objects can be shared
     * everywhere in the "object space" of Redis and may end in places where
     * they are not handled. We handle them only as values in the keyspace. */

    // 编码共享对象是不安全的
    // 若对象是共享的，则不进行编码
     if (o->refcount > 1) return o;

    /* Check if we can represent this string as a long integer.
     * Note that we are sure that a string larger than 20 chars is not
     * representable as a 32 nor 64 bit integer. */

     // 判断字符串是否可以用整数表示
     // 注意：超过20位的字符串则无法用整数表示
    len = sdslen(s);
    if (len <= 20 && string2l(s,len,&value)) {
        /* This object is encodable as a long. Try to use a shared object.
         * Note that we avoid using shared integers when maxmemory is used
         * because every object needs to have a private LRU field for the LRU
         * algorithm to work well. */

        // [0, 10000) 尝试使用共享整数
        if ((server.maxmemory == 0 ||
            !(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) &&
            value >= 0 &&
            value < OBJ_SHARED_INTEGERS)
        {
            // 是否之前的对象
            decrRefCount(o);
            incrRefCount(shared.integers[value]);
            return shared.integers[value];
        } else {
            // 否则使用OBJ_ENCODING_INT编码
            if (o->encoding == OBJ_ENCODING_RAW) sdsfree(o->ptr);
            o->encoding = OBJ_ENCODING_INT;
            o->ptr = (void*) value;
            return o;
        }
    }

    /* If the string is small and is still RAW encoded,
     * try the EMBSTR encoding which is more efficient.
     * In this representation the object and the SDS string are allocated
     * in the same chunk of memory to save space and cache misses. */

    // 若字符串长度小于EMBSTR的限制
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT) {
        robj *emb;

        // 编码是EMBSTR则返回
        if (o->encoding == OBJ_ENCODING_EMBSTR) return o;

        // 否则使用EMBSTR编码
        emb = createEmbeddedStringObject(s,sdslen(s));
        // 释放之前的对象
        decrRefCount(o);
        return emb;
    }

    /* We can't encode the object...
     *
     * Do the last try, and at least optimize the SDS string inside
     * the string object to require little space, in case there
     * is more than 10% of free space at the end of the SDS string.
     *
     * We do that only for relatively large strings as this branch
     * is only entered if the length of the string is greater than
     * OBJ_ENCODING_EMBSTR_SIZE_LIMIT. */

    // 无法进行编码优化，但若空间浪费过多(有1/10的空间未使用)，则释放未使用的内存空间
    if (o->encoding == OBJ_ENCODING_RAW &&
        sdsavail(s) > len/10)
    {
        o->ptr = sdsRemoveFreeSpace(o->ptr);
    }

    /* Return the original object. */
    return o;
}

• SET命令底层实现

void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
    long long milliseconds = 0; /* initialized to avoid any harmness warning */

    // 若存在过期时间对象, 则取出过期时间
    if (expire) {
        // 从exprie对象中获取long long类型的过期时间
        if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
            return;
        if (milliseconds <= 0) {
            addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
            return;
        }

        // redis实际以毫秒形式保存过期时间
        // 若输入时间为秒，则转换为毫秒
        if (unit == UNIT_SECONDS) milliseconds *= 1000;
    }

    // NX：表示key不存在
    // XX：表示key存在
    // 若使用NX，但找到给key；若使用XX，但未找到该key
    // 回复错误给客户端
    if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
        (flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
    {
        addReply(c, abort_reply ? abort_reply : shared.nullbulk);
        return;
    }

    // 添加键值对到当前db
    setKey(c->db,key,val);

    // 设置数据库为脏(dirty)，每次修改key后，都会dirty++
    server.dirty++;

    // 设置key的过期时间
    if (expire) setExpire(c,c->db,key,mstime()+milliseconds);

    // 发送"set"事件通知，用于发布订阅模式，通知客户端接收消息
    notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);

    // 发送"expire"事件通知
    if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
        "expire",key,c->db->id);

    // 回复客户端
    addReply(c, ok_reply ? ok_reply : shared.ok);
}

• INCR和DECR底层实现

void incrDecrCommand(client *c, long long incr) {
    long long value, oldvalue;
    robj *o, *new;

    // 寻址key对象
    o = lookupKeyWrite(c->db,c->argv[1]);

    // 若未找到或不是字符串对象则返回
    if (o != NULL && checkType(c,o,OBJ_STRING)) return;

    // 取出对象的整数值，保存在value中
    if (getLongLongFromObjectOrReply(c,o,&value,NULL) != C_OK) return;

    // 判断是否有溢出发生
    oldvalue = value;
    if ((incr < 0 && oldvalue < 0 && incr < (LLONG_MIN-oldvalue)) ||
        (incr > 0 && oldvalue > 0 && incr > (LLONG_MAX-oldvalue))) {
        addReplyError(c,"increment or decrement would overflow");
        return;
    }

    // 计算新值
    value += incr;

    // 对象未共享，且超过共享整数编码范围或用long可以表示, 直接存储
    if (o && o->refcount == 1 && o->encoding == OBJ_ENCODING_INT &&
        (value < 0 || value >= OBJ_SHARED_INTEGERS) &&
        value >= LONG_MIN && value <= LONG_MAX)
    {
        new = o;
        o->ptr = (void*)((long)value);
    } else {
        // 新建字符串对象
        new = createStringObjectFromLongLong(value);
        if (o) {
            dbOverwrite(c->db,c->argv[1],new);
        } else {
            dbAdd(c->db,c->argv[1],new);
        }
    }

    // 向数据库发送键被修改的信号
    signalModifiedKey(c->db,c->argv[1]);

    // 发送"incrby"事件通知
    notifyKeyspaceEvent(NOTIFY_STRING,"incrby",c->argv[1],c->db->id);

    // 服务器为脏
    server.dirty++;

    // 回复客户端
    addReply(c,shared.colon);
    addReply(c,new);
    addReply(c,shared.crlf);
}