（九）— t_list，t_string的分析
最后更新于：2022-04-01 20:20:31
今天我是看完了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]);
~~~
';