Redis源码分析（九）--- t_list，t_string的分析

          今天我是看完了t_list和t_string的代码，从名字知道，这也是和之前的t_hash非常类似的，无非就是各种形式，转化来转化去的。先讲讲t_list，对比于昨天的t_hash,t_hash是ziplist和dict之间的转换，t_list则是描述的是ziplist压缩表和linedlist普通链表，换句话说，当client这个robj作为参数传入的时候，都分为ENCODING_ZIIPLIST和ENCODING_LINKEDLIST 2类编码方式处理。还有一个在t_list出现的一个 比较新颖的东西是出现了锁的概念,操作系统的东西在这里也会碰到了，这里的代码也非常值得学习。下面，我们看看一般的t_list中包括的一些方法和命令：

/* List方法 */
/* list的操作分为2种，LINKEDLIST普通链表和ZIPLIST压缩列表的相关操作 */
void listTypeTryConversion(robj *subject, robj *value) /* 判断是否需要将ziplist转为linkedlist */
void listTypePush(robj *subject, robj *value, int where) /* 在头部或尾部插入value元素 */
robj *listTypePop(robj *subject, int where)  /* 在列表的头部或尾弹出元素 */
unsigned long listTypeLength(robj *subject) /* 列表的长度 */
listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction) /* 返回列表迭代器，方向有头尾之分 */
void listTypeReleaseIterator(listTypeIterator *li) /* 释放列表迭代器 */
int listTypeNext(listTypeIterator *li, listTypeEntry *entry) /* 根据列表迭代器，获取下一个元素 */
robj *listTypeGet(listTypeEntry *entry) /* 获取listType元素，有ziplist和linkedlist */
void listTypeInsert(listTypeEntry *entry, robj *value, int where) /* listType了类型插入元素操作 */
int listTypeEqual(listTypeEntry *entry, robj *o) /* 判断2个元素是否相等 */
void listTypeDelete(listTypeEntry *entry) /* listType类型删除元素 */
void listTypeConvert(robj *subject, int enc) /* listType类型的转换操作，这里指的是往linkedList上转 */
	
/* List的相关命令 */
void pushGenericCommand(redisClient *c, int where) /* 插入操作命令的原始操作 */
void lpushCommand(redisClient *c) /* 左边插入元素命令 */
void rpushCommand(redisClient *c) /* 右边插入元素命令 */
void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) /* 有返回状态的插入操作命令，假设首先都是能够实现插入命令的 */
void lpushxCommand(redisClient *c) /* 左边插入元素有返回消息的命令 */
void rpushxCommand(redisClient *c) /* 右边插入元素有返回消息的命令 */
void linsertCommand(redisClient *c) /* 列表指定位置插入元素操作命令 */
void llenCommand(redisClient *c) /* 列表返回长度命令 */
void lindexCommand(redisClient *c) /* 获取index位置的上的元素 */
void lsetCommand(redisClient *c) /* listType类型设置value命令 */
void popGenericCommand(redisClient *c, int where) /* 实现弹出操作的原始命令 */
void lpopCommand(redisClient *c) /* 左边弹出元素命令 */
void rpopCommand(redisClient *c) /* 右边弹出操作命令 */
void lrangeCommand(redisClient *c) /* 移动listType位置操作 */
void ltrimCommand(redisClient *c) /* listType实现截取操作，把多余范围的元素删除 */
void lremCommand(redisClient *c) /* 移除在listType中出现的与指定的元素相等的元素 */
void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) /* 元素从一个obj右边弹出，在从左侧推入另一个obj列表操作 */
void rpoplpushCommand(redisClient *c) /* 右边弹出，左边推入元素的命令 */
void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) >/* 设置客户端为阻塞模式，并设置超时时间，当请求特定的key元素时 */
void unblockClientWaitingData(redisClient *c) /* 客户端解锁操作 */
void signalListAsReady(redisDb *db, robj *key) /* 将key存入server中，后续可以用于客户端的存取 */
int serveClientBlockedOnList(redisClient *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int where) /* 根据server，Client的key,value情况，判断server此时能否服务于Client，否则Client将被阻塞 */
void handleClientsBlockedOnLists(void) /* 服务端解除阻塞住的Client */
int getTimeoutFromObjectOrReply(redisClient *c, robj *object, time_t *timeout) /* 获取超时时间 */ 
void blockingPopGenericCommand(redisClient *c, int where) /* 阻塞弹出命令的原始操作 */
void blpopCommand(redisClient *c) /* 左边弹出数据的阻塞式命令 */
void brpopCommand(redisClient *c) /* 右边弹出数据的阻塞式命令 */
void brpoplpushCommand(redisClient *c) /* 弹出推入阻塞式命令 */

看到这么多API，是否有被吓到的感觉，但里面其实很多的方法都很简单，我只挑几个比较典型的方法做一下分析，向什么返回长度了，获取，设置键值的大家可以自己看具体的方法，不难的。一般的t_list的处理流程，比如获取迭代器的方法：

/* Stores pointer to current the entry in the provided entry structure
 * and advances the position of the iterator. Returns 1 when the current
 * entry is in fact an entry, 0 otherwise. */
int listTypeNext(listTypeIterator *li, listTypeEntry *entry) {
    /* Protect from converting when iterating */
    redisAssert(li->subject->encoding == li->encoding);

    entry->li = li;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
        entry->zi = li->zi;
        if (entry->zi != NULL) {
            if (li->direction == REDIS_TAIL)
            	//根据方向调用pre或next的方法
                li->zi = ziplistNext(li->subject->ptr,li->zi);
            else
                li->zi = ziplistPrev(li->subject->ptr,li->zi);
            return 1;
        }
    } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
        entry->ln = li->ln;
        if (entry->ln != NULL) {
            if (li->direction == REDIS_TAIL)
            	//普通的列表的调用方式同上
                li->ln = li->ln->next;
            else
                li->ln = li->ln->prev;
            return 1;
        }
    } else {
        redisPanic("Unknown list encoding");
    }
    return 0;
}

很多API和t_hash的方法都是极其类似的，我就不多说了，我们将关注点，放在列表的锁控制上，先讲个前提，在redis客户端执行操作的时候，会有个请求超时时间，在这个请求的时间内，客户端如果没有找到key对应的数据，是会被阻塞的，什么意思呢，比如：

/* 设置客户端为阻塞模式，并设置超时时间，当请求特定的key元素时 */
/* 这个客户端阻塞的意思：当客户端请求list中某个特定key值时，如果key存在且列表非空，当然 */
/* 当然不会阻塞，正常返回数据，如果当客户端请求某个key不存在，或列表为empty的时候，客户端将被阻塞 */
/* 只有当有这个key被写入的时候，客户端才会被解锁 *

这个其实就是操作系统中的PV操作类似的原理，跟java里的wait(),signal()方法的模型是一样的，他的实现代码如下：

/* This is how the current blocking POP works, we use BLPOP as example:
 * - If the user calls BLPOP and the key exists and contains a non empty list
 *   then LPOP is called instead. So BLPOP is semantically the same as LPOP
 *   if blocking is not required.
 * - If instead BLPOP is called and the key does not exists or the list is
 *   empty we need to block. In order to do so we remove the notification for
 *   new data to read in the client socket (so that we'll not serve new
 *   requests if the blocking request is not served). Also we put the client
 *   in a dictionary (db->blocking_keys) mapping keys to a list of clients
 *   blocking for this keys.
 * - If a PUSH operation against a key with blocked clients waiting is
 *   performed, we mark this key as "ready", and after the current command,
 *   MULTI/EXEC block, or script, is executed, we serve all the clients waiting
 *   for this list, from the one that blocked first, to the last, accordingly
 *   to the number of elements we have in the ready list.
 */

/* Set a client in blocking mode for the specified key, with the specified
 * timeout */
/* 设置客户端为阻塞模式，并设置超时时间，当请求特定的key元素时 */
/* 这个客户端阻塞的意思：当客户端请求list中某个特定key值时，如果key存在且列表非空，当然 */
/* 当然不会阻塞，正常返回数据，如果当客户端请求某个key不存在，或列表为empty的时候，客户端将被阻塞 */
/* 只有当有这个key被写入的时候，客户端才会被解锁 */		
void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) {
    dictEntry *de;
    list *l;
    int j;

    //设置超时时间
    c->bpop.timeout = timeout;
    c->bpop.target = target;

    if (target != NULL) incrRefCount(target);

    for (j = 0; j < numkeys; j++) {
    	//下面为为找到的key上锁
        /* If the key already exists in the dict ignore it. */
        //如果此时，某些key已经存在了，直接忽略
        if (dictAdd(c->bpop.keys,keys[j],NULL) != DICT_OK) continue;
        incrRefCount(keys[j]);

        /* And in the other "side", to map keys -> clients */
        //根据key找到对于请求该key的客户端，也就是将要被阻塞的Client
        de = dictFind(c->db->blocking_keys,keys[j]);
        if (de == NULL) {
            int retval;

            /* For every key we take a list of clients blocked for it */
            l = listCreate();
            //为c客户端添加一个阻塞的key
            retval = dictAdd(c->db->blocking_keys,keys[j],l);
            //增加key的引用次数
            incrRefCount(keys[j]);
            redisAssertWithInfo(c,keys[j],retval == DICT_OK);
        } else {
            l = dictGetVal(de);
        }
        listAddNodeTail(l,c);
    }

    /* Mark the client as a blocked client */
    /* 标记Client为阻塞的客户端 */
    c->flags |= REDIS_BLOCKED;
    //服务端的阻塞客户端计数递增
    server.bpop_blocked_clients++;
}

所有的key未来都会存在于server.readykeys里面，客户端判断自己所请求的key是否存在于服务端中的readykeys中，如果不存在就会被阻塞了。将key存入服务单的数据库的操作：

/* If the specified key has clients blocked waiting for list pushes, this
 * function will put the key reference into the server.ready_keys list.
 * Note that db->ready_keys is a hash table that allows us to avoid putting
 * the same key again and again in the list in case of multiple pushes
 * made by a script or in the context of MULTI/EXEC.
 *
 * The list will be finally processed by handleClientsBlockedOnLists() */
/* 将key存入server中，后续可以用于客户端的存取 */
void signalListAsReady(redisDb *db, robj *key) {
    readyList *rl;

    /* No clients blocking for this key? No need to queue it. */
    /* 如果没有客户端为此key所阻塞的，直接不添加 */
    if (dictFind(db->blocking_keys,key) == NULL) return;

    /* Key was already signaled? No need to queue it again. */
    /* 如果key已经存在，不不要重复添加 */
    if (dictFind(db->ready_keys,key) != NULL) return;

    /* Ok, we need to queue this key into server.ready_keys. */
    rl = zmalloc(sizeof(*rl));
    rl->key = key;
    rl->db = db;
    incrRefCount(key);
    //添加到list列表尾部
    listAddNodeTail(server.ready_keys,rl);

    /* We also add the key in the db->ready_keys dictionary in order
     * to avoid adding it multiple times into a list with a simple O(1)
     * check. */
    incrRefCount(key);
    redisAssert(dictAdd(db->ready_keys,key,NULL) == DICT_OK);
}

其实这里我不免会有个小问题，为什么list出现Client被阻塞的操作，而在t_hash中不会出现类似的操作呢？略疑惑。

      下面说说t_string的用法，t_string里面没有涉及什么特定的结构体，就是最直接的键值对的设置，直接改变数据库中的键值，先列出里面的主要方法:

/* 方法 API */
static int checkStringLength(redisClient *c, long long size) /* 检查字符串长度是否超出限制最大长度，512*1024*1024个字节长度 */
void setGenericCommand(redisClient *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) /* 设置的泛型指令操作 */
void setCommand(redisClient *c) /* 设置的综合方法，设置都是根据Client中的参数而定 */
void setnxCommand(redisClient *c) /* key不存在的时候才设置值，flag为REDIS_SET_NX时 */
void setexCommand(redisClient *c) /* key存在的时候才设置值，flag为REDIS_SET_NO_FLAGS，命令到期时间单位为秒 */
void psetexCommand(redisClient *c) /* key存在的时候才设置值，flag为REDIS_SET_NO_FLAGS，命令到期时间单位为毫秒 */
int getGenericCommand(redisClient *c) /* 获取命令的泛型命令 */
void getCommand(redisClient *c) /* 获取命令的操作 */
void getsetCommand(redisClient *c) /* 获取set命令的操作 */
void setrangeCommand(redisClient *c) /* 设置rangge命令的操作 */
void getrangeCommand(redisClient *c) /* 获取range命令的操作 */
void mgetCommand(redisClient *c) /* 执行多次getCommand指令 */
void msetGenericCommand(redisClient *c, int nx) /* 一次运行多个设置操作泛型命令 */
void msetCommand(redisClient *c) /* 多次设置指令的非NX模式 */
void msetnxCommand(redisClient *c)  /* 多次设置指令的NX模式,即只在不存在的时候才设置 */
void incrDecrCommand(redisClient *c, long long incr) /* 增加或减少固定值的操作，减少其实就是增加-1 */
void incrCommand(redisClient *c) /* 递增1操作，incr递增设置为1 */
void decrCommand(redisClient *c) /* 递减1操作，incr递增设置为-1 */
void incrbyCommand(redisClient *c) /* 增加指令，从Client中获取递增值 */
void decrbyCommand(redisClient *c) /* 减少指令，从Client中获取减少值 */
void incrbyfloatCommand(redisClient *c) /* 增加float类型值的操作命令，在获取原始值的时候，获取的方法不一样，为getLongDoubleFromObjectOrReply*/
void appendCommand(redisClient *c) /* 追加命令操作，其实也是key:value的形式 */
void strlenCommand(redisClient *c) /* 获取命令长度 */

里面都是一些键值对的简单操作，有一个不同点是，里面的设置操作分为3种：

/* SET操作的一些FLAG */
#define REDIS_SET_NO_FLAGS 0
#define REDIS_SET_NX (1<<0)     /* Set if key not exists. */
#define REDIS_SET_XX (1<<1)     /* Set if key exists. */

如上面所说，nx模式指的是key不存在的时候才设置值，xx为存在的时候设置，这种模式设置有命令到期时间的限制，分为单位为秒级和毫秒级，而nx模式下，没有时间上的限制，调用的泛型方法:

void setGenericCommand(redisClient *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply)

注意其中的expire,和时间单位unit:

/* key不存在的时候才设置值，flag为REDIS_SET_NX时 */
void setnxCommand(redisClient *c) {
    c->argv[2] = tryObjectEncoding(c->argv[2]);
    setGenericCommand(c,REDIS_SET_NX,c->argv[1],c->argv[2],NULL,0,shared.cone,shared.czero);
}

/* key存在的时候才设置值，flag为REDIS_SET_NO_FLAGS，命令到期时间单位为秒 */
void setexCommand(redisClient *c) {
    c->argv[3] = tryObjectEncoding(c->argv[3]);
    setGenericCommand(c,REDIS_SET_NO_FLAGS,c->argv[1],c->argv[3],c->argv[2],UNIT_SECONDS,NULL,NULL);
}

/* key存在的时候才设置值，flag为REDIS_SET_NO_FLAGS，命令到期时间单位为毫秒 */
void psetexCommand(redisClient *c) {
    c->argv[3] = tryObjectEncoding(c->argv[3]);
    setGenericCommand(c,REDIS_SET_NO_FLAGS,c->argv[1],c->argv[3],c->argv[2],UNIT_MILLISECONDS,NULL,NULL);
}

nx模式时expire和unit参数分别为NULL，0，所以我做出了如下猜想，可能是因为，当键值对存在的时候，考虑多并发设置的情况，有些Client可能被阻塞，所以有超时时间的存在，而key不存在的时候，就看谁更快了吧，就是直接设置。下面还有一个这个文件比较有意思的方法，有点批处理的味道，换句话说，一个方法里执行多次set操作：

/* 一次运行多个设置操作泛型命令 */
void msetGenericCommand(redisClient *c, int nx) {
    int j, busykeys = 0;

    /* 设置操作一定是成对出现的 */
    if ((c->argc % 2) == 0) {
        addReplyError(c,"wrong number of arguments for MSET");
        return;
    }
    /* Handle the NX flag. The MSETNX semantic is to return zero and don't
     * set nothing at all if at least one already key exists. */
    if (nx) {
        for (j = 1; j < c->argc; j += 2) {
            if (lookupKeyWrite(c->db,c->argv[j]) != NULL) {
                busykeys++;
            }
        }
        
        //如果key存在，返回0
        if (busykeys) {
            addReply(c, shared.czero);
            return;
        }
    }

	//隔2个执行设置key操作指令一次
    for (j = 1; j < c->argc; j += 2) {
        c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
        setKey(c->db,c->argv[j],c->argv[j+1]);
        notifyKeyspaceEvent(REDIS_NOTIFY_STRING,"set",c->argv[j],c->db->id);
    }
    server.dirty += (c->argc-1)/2;
    addReply(c, nx ? shared.cone : shared.ok);
}

也不是什么特别的操作，应该redis编写者为了方便效率的提高吧。其他的一些方法，请读者可以学习t_string.c的文件，还有提醒一点，在redis数据库中，存储的形式都是K-V形式的，所有的获取，设置都是通过KEY的形式操作，你会看到很多这样的方法：

dbAdd(c->db,c->argv[1],new);
setKey(c->db,c->argv[j],c->argv[j+1]);