顺风车运营研发团队 谭淼
一、brpoplpush
brpoplpush是rpoplpush的阻塞版本，当给定列表 source 不为空时， brpoplpush的表现和rpoplpush一样。
当列表 source 为空时，brpoplpush命令将阻塞连接，直到等待超时，或有另一个客户端对 source 执行lpush或rpush命令为止。
brpoplpush的源码如下：
void brpoplpushCommand(client *c) {
mstime_t timeout;
//首先将命令的第三个参数（超时时间）存在timeout变量中
if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_SECONDS)
!= C_OK) return;
//命令的第一个参数是source key，从数据库字典中查询source key，获取source key的value
robj *key = lookupKeyWrite(c->db, c->argv[1]);
if (key == NULL) {
//如果list为空，则判断该命令是否是事务
if (c->flags & CLIENT_MULTI) {
//如果是事务，则直接返回
addReply(c, shared.nullbulk);
} else {
//如果不是事务，则执行等待操作
blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]);
}
} else {
//如果list不为空，则判断value是否是列表类型
if (key->type != OBJ_LIST) {
//如果不是列表类型，则返回错误
addReply(c, shared.wrongtypeerr);
} else {
//如果是list则执行rpoplpush命令
serverAssertWithInfo(c,key,listTypeLength(key) > 0);
rpoplpushCommand(c);
}
}
}brpoplpush的关键点是rpoplpushCommand()函数。
rpoplpushCommand()函数是执行rpoplpush命令的函数，在执行brpoplpush命令时，如果不发生阻塞行为，也是调用的这个函数，
void rpoplpushCommand(client *c) {
robj *sobj, *value;
//命令的第一个参数是source key，从数据库字典中查询source key，获取source key的value，并赋值给sobj
//对sobj进行检查，如果sobj值为null或者sobj不是列表类型，则返回。
if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
checkType(c,sobj,OBJ_LIST)) return;
//如果sobj的长度为0，返回
if (listTypeLength(sobj) == 0) {
/* This may only happen after loading very old RDB files. Recent
* versions of Redis delete keys of empty lists. */
addReply(c,shared.nullbulk);
} else {
//取出第二个参数（目标key）所对应的目标列表，并赋值给dobj
robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
robj *touchedkey = c->argv[1];
//对dobj进行检查，如果dobj值为null或者dobj不是列表类型，则返回。
if (dobj && checkType(c,dobj,OBJ_LIST)) return;
//从源列表的尾部弹出一个元素，保存在value中
value = listTypePop(sobj,LIST_TAIL);
/* We saved touched key, and protect it, since rpoplpushHandlePush
* may change the client command argument vector (it does not
* currently). */
incrRefCount(touchedkey);
//将从源列表的尾部弹出的元素push到目标列表中
rpoplpushHandlePush(c,c->argv[2],dobj,value);
/* listTypePop returns an object with its refcount incremented */
decrRefCount(value);
/* Delete the source list when it is empty */
//键空间事件通知
notifyKeyspaceEvent(NOTIFY_LIST,"rpop",touchedkey,c->db->id);
//如果弹出元素后源列表长度为0，则删除源key和value
if (listTypeLength(sobj) == 0) {
dbDelete(c->db,touchedkey);
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",
touchedkey,c->db->id);
}
signalModifiedKey(c->db,touchedkey);
decrRefCount(touchedkey);
server.dirty++;
}
}二、lindex
返回列表 key 中，下标为 index 的元素。
下标(index)参数 start 和 stop 都以 0 为底，也就是说，以 0 表示列表的第一个元素，以 1 表示列表的第二个元素，以此类推。
你也可以使用负数下标，以 -1 表示列表的最后一个元素， -2 表示列表的倒数第二个元素，以此类推。
如果 key 不是列表类型，返回一个错误。
lindex的源码如下：
void lindexCommand(client *c) {
//首先获取查询的key的列表
robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk);
//对列表进行检查，若列表为空或者类型不是OBJ_LIST，则返回
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
long index;
robj *value = NULL;
//读取命令中的index参数，保存在index变量中
if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != C_OK))
return;
if (o->encoding == OBJ_ENCODING_QUICKLIST) {
//初始化entry，并从列表中找到第index个元素，保存到entry变量中
quicklistEntry entry;
if (quicklistIndex(o->ptr, index, &entry)) {
//entry是一个quicklistEntry类型的变量，若取第index个元素成功，则会执行下面的代码
if (entry.value) {
//若列表中的第index个元素是一个string，则会创建一个StringObject，保存在value中
value = createStringObject((char*)entry.value,entry.sz);
} else {
//若列表中的第index个元素是一个数字，则会将数字转化为字符串，保存在value中
value = createStringObjectFromLongLong(entry.longval);
}
addReplyBulk(c,value);
decrRefCount(value);
} else {
//若取第index个元素失败，则会返回空
addReply(c,shared.nullbulk);
}
} else {
serverPanic("Unknown list encoding");
}
}lindexCommand()方法的核心是quicklistIndex()方法。
/* 该方法的核心思想：
* （1）首先根据idx是否大于0，决定遍历的顺序。
* （2）根据idx和每个quicklist节点中ziplist元素的个数，找到包含第idx个元素的quicklist节点。
* （3）在包含第idx个元素的ziplist中找到表示第idx个元素的entry字符串，读取并返回。*/
int quicklistIndex(const quicklist *quicklist, const long long idx,
quicklistEntry *entry) {
quicklistNode *n;
unsigned long long accum = 0;
unsigned long long index;
//根据idx是否大于0，决定遍历的顺序。
int forward = idx < 0 ? 0 : 1; /* < 0 -> reverse, 0+ -> forward */
initEntry(entry);
entry->quicklist = quicklist;
if (!forward) {
//如果是逆序遍历，获得逆序遍历的index，并将当前指针n指向quicklist的尾部
index = (-idx) - 1;
n = quicklist->tail;
} else {
//如果是顺序遍历，index不变，并将当前指针n指向quicklist的头部
index = idx;
n = quicklist->head;
}
//若index的值超出了整个列表的长度，则直接返回未找到
if (index >= quicklist->count)
return 0;
//根据index和每个quicklist节点中ziplist元素的个数，找到包含第idx个元素的quicklist节点，保存在遍历n中
while (likely(n)) {
if ((accum + n->count) > index) {
break;
} else {
D("Skipping over (%p) %u at accum %lld", (void *)n, n->count,
accum);
accum += n->count;
n = forward ? n->next : n->prev;
}
}
if (!n)
return 0;
D("Found node: %p at accum %llu, idx %llu, sub+ %llu, sub- %llu", (void *)n,
accum, index, index - accum, (-index) - 1 + accum);
entry->node = n;
//计算所要寻找的元素在ziplist中的偏移量
if (forward) {
/* forward = normal head-to-tail offset. */
entry->offset = index - accum;
} else {
/* reverse = need negative offset for tail-to-head, so undo
* the result of the original if (index < 0) above. */
entry->offset = (-index) - 1 + accum;
}
//对quicklist解码
quicklistDecompressNodeForUse(entry->node);
//根据偏移量获取元素的值，并保存在entry变量中
entry->zi = ziplistIndex(entry->node->zl, entry->offset);
ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval);
/* The caller will use our result, so we don't re-compress here.
* The caller can recompress or delete the node as needed. */
return 1;
}quicklistIndex()方法的思想是先找到目标元素所在的quicklist节点的位置，再从该节点中找到该元素。
以从一个长度为18的list中找到第-10个元素为例，其中假设该list有三个quicklist节点，每个节点中的ziplist分别存储了6、7、5个entry。
1、首先根据idx是否大于0，决定遍历的顺序，并计算出index的值。由于-10小于0，则从尾节点开始遍历，同时计算index值为-(-10)-1=9
2、根据index和每个quicklist节点中ziplist元素的个数，找到包含第idx个元素的quicklist节点。
while (likely(n)) {
if ((accum + n->count) > index) {
break;
} else {
D("Skipping over (%p) %u at accum %lld", (void *)n, n->count, accum);
accum += n->count;
n = forward ? n->next : n->prev;
}
}- （1）首先accum=0，n指向尾节点，故n->count的值为5，accum + n->count小于9，因此走到else条件中；
- （2）在else条件中，accum更新为5，当前节点n更新为前一个节点；
- （3）重新进入到while循环中，accum +
n->count的值为5+7=12，大于index的值9，故退出while循环，此时n指向第二个节点，accum的值为5。 - （4）随后计算offset偏移量：
//计算所要寻找的元素在ziplist中的偏移量
if (forward) {
/* forward = normal head-to-tail offset. */
entry->offset = index - accum;
} else {
/* reverse = need negative offset for tail-to-head, so undo
* the result of the original if (index < 0) above. */
entry->offset = (-index) - 1 + accum;
}由于是逆序遍历，故entry->offset的值为-9-1+5=-5，即第二个节点的倒数第5个entry是我们想要找到的元素。
- （5）最后从ziplist中获取entry即可。
三、llen
返回列表 key 的长度。
如果 key 不存在，则 key 被解释为一个空列表，返回 0 。
void llenCommand(client *c) {
robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
addReplyLongLong(c,listTypeLength(o));
}llen命令实现比较简单，主要是调用的是listTypeLength()方法。
unsigned long listTypeLength(const robj *subject) {
if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
return quicklistCount(subject->ptr);
} else {
serverPanic("Unknown list encoding");
}
}listTypeLength()方法首先判断是否是OBJ_ENCODING_QUICKLIST类型的对象，如果是则调用quicklistCount()方法，直接返回list的长度。
/* Return cached quicklist count */
unsigned long quicklistCount(const quicklist *ql) { return ql->count; }四、linsert
linsert可以在列表的指定位置插入一个元素，用法为LINSERT key BEFORE|AFTER pivot value
将值 value 插入到列表 key 当中，位于值 pivot 之前或之后。
当 pivot 不存在于列表 key 时，不执行任何操作。
当 key 不存在时， key 被视为空列表，不执行任何操作。
如果 key 不是列表类型，返回一个错误。
linsert的源码如下：
void linsertCommand(client *c) {
int where;
robj *subject;
listTypeIterator *iter;
listTypeEntry entry;
int inserted = 0;
//根据第二个参数设置where变量
if (strcasecmp(c->argv[2]->ptr,"after") == 0) {
where = LIST_TAIL;
} else if (strcasecmp(c->argv[2]->ptr,"before") == 0) {
where = LIST_HEAD;
} else {
addReply(c,shared.syntaxerr);
return;
}
//获取并检查key所对应的列表
if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
checkType(c,subject,OBJ_LIST)) return;
/* Seek pivot from head to tail */
//获取list的迭代器
iter = listTypeInitIterator(subject,0,LIST_TAIL);
//开始遍历list
while (listTypeNext(iter,&entry)) {
//寻找指定的目标元素
if (listTypeEqual(&entry,c->argv[3])) {
//若找到，则根据where变量指定的位置进行插入
listTypeInsert(&entry,c->argv[4],where);
inserted = 1;
break;
}
}
//释放迭代器
listTypeReleaseIterator(iter);
//如果插入成功，则发送键空间事件通知，否则返回
if (inserted) {
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_LIST,"linsert", c->argv[1],c->db->id);
server.dirty++;
} else {
/* Notify client of a failed insert */
addReply(c,shared.cnegone);
return;
}
//若插入成功，返回插入后列表的长度
addReplyLongLong(c,listTypeLength(subject));
}linsertCommand()函数的重点是调用listTypeInsert()函数进行列表的插入。而listTypeInsert()函数会根据前插还是后插的不同调用quicklistInsertBefore()和quicklistInsertAfter()
void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
if (entry->li->encoding == OBJ_ENCODING_QUICKLIST) {
value = getDecodedObject(value);
sds str = value->ptr;
size_t len = sdslen(str);
if (where == LIST_TAIL) {
quicklistInsertAfter((quicklist *)entry->entry.quicklist, &entry->entry, str, len);
} else if (where == LIST_HEAD) {
quicklistInsertBefore((quicklist *)entry->entry.quicklist, &entry->entry, str, len);
}
decrRefCount(value);
} else {
serverPanic("Unknown list encoding");
}
}而quicklistInsertAfter()和quicklistInsertBefore()都是调用的_quicklistInsert()方法。以quicklistInsertAfter()为例，介绍quicklist的插入过程。
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *entry,
void *value, const size_t sz) {
_quicklistInsert(quicklist, entry, value, sz, 1);
}
_quicklistInsert()的代码如下，我们称命令中的pivot为目标元素，称value为待插入元素：
/* Insert a new entry before or after existing entry 'entry'.
*
* If after==1, the new value is inserted after 'entry', otherwise
* the new value is inserted before 'entry'. */
REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry, void *value, const size_t sz, int after) {
int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
int fill = quicklist->fill;
quicklistNode *node = entry->node;
quicklistNode *new_node = NULL;
//若目标元素所在的quicklist节点为空，则直接创建一个新的quicklist节点，插入到quicklist中，并将待插入元素插入到新创建的节点中。
if (!node) {
/* we have no reference node, so let's create only node in the list */
D("No node given!");
new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
__quicklistInsertNode(quicklist, NULL, new_node, after);
new_node->count++;
quicklist->count++;
return;
}
/* Populate accounting flags for easier boolean checks later */
//判断目标元素所在的quicklist节点的ziplist是否可以插入，若ziplist的大小超过了redis设置的ziplist的最大值，则不能再插入，需要设置full参数为1
if (!_quicklistNodeAllowInsert(node, fill, sz)) {
D("Current node is full with count %d with requested fill %lu", node->count, fill);
full = 1;
}
//如果待插入元素是要插入到目标元素的后面，且目标元素是当前quicklist节点的最后一个entry，则需要判断当前quicklist节点的下一个节点是否可以已经满了
if (after && (entry->offset == node->count)) {
D("At Tail of current ziplist");
at_tail = 1;
if (!_quicklistNodeAllowInsert(node->next, fill, sz)) {
D("Next node is full too.");
full_next = 1;
}
}
//如果待插入元素是要插入到目标元素的前面，且目标元素是当前quicklist节点的第一个entry，则需要判断当前quicklist节点的前一个节点是否可以已经满了
if (!after && (entry->offset == 0)) {
D("At Head");
at_head = 1;
if (!_quicklistNodeAllowInsert(node->prev, fill, sz)) {
D("Prev node is full too.");
full_prev = 1;
}
}
/* Now determine where and how to insert the new element */
//分情况开始插入
if (!full && after) {//如果当前的quicklist节点的ziplist没有满，且待插入元素是要插入到目标元素的后面
D("Not full, inserting after current position.");
quicklistDecompressNodeForUse(node);
unsigned char *next = ziplistNext(node->zl, entry->zi);
//目标元素的下一个entry是否为空，若为空则调用ziplistPush，若不为空则调用ziplistInsert
if (next == NULL) {
node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL);
} else {
node->zl = ziplistInsert(node->zl, next, value, sz);
}
node->count++;
quicklistNodeUpdateSz(node);
quicklistRecompressOnly(quicklist, node);
} else if (!full && !after) {//如果当前的quicklist节点的ziplist没有满，且待插入元素是要插入到目标元素的前面
D("Not full, inserting before current position.");
quicklistDecompressNodeForUse(node);
node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
node->count++;
quicklistNodeUpdateSz(node);
quicklistRecompressOnly(quicklist, node);
} else if (full && at_tail && node->next && !full_next && after) {
/* If we are: at tail, next has free space, and inserting after:
* - insert entry at head of next node. */
//如果后插的时候当前的quicklist节点的ziplist满了，但是当前的quicklist节点的下一个节点没有满，且目标元素是当前quicklist节点的最后一个entry，
//则将待插入元素插入到当前的quicklist节点的下一个节点的第一个entry的前面
D("Full and tail, but next isn't full; inserting next node head");
new_node = node->next;
quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD);
new_node->count++;
quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node);
} else if (full && at_head && node->prev && !full_prev && !after) {
/* If we are: at head, previous has free space, and inserting before:
* - insert entry at tail of previous node. */
//如果前插的时候当前的quicklist节点的ziplist满了，但是当前的quicklist节点的前一个节点没有满，且目标元素是当前quicklist节点的第一个entry，
//则将待插入元素插入到当前的quicklist节点的前一个节点的最后一个entry的后面
D("Full and head, but prev isn't full, inserting prev node tail");
new_node = node->prev;
quicklistDecompressNodeForUse(new_node);
new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL);
new_node->count++;
quicklistNodeUpdateSz(new_node);
quicklistRecompressOnly(quicklist, new_node);
} else if (full && ((at_tail && node->next && full_next && after) ||
(at_head && node->prev && full_prev && !after))) {
/* If we are: full, and our prev/next is full, then:
* - create new node and attach to quicklist */
//如果前插的时候目标元素是当前quicklist节点的第一个entry，但是当前节点和当前节点的前一个节点都满了，则需要创建一个新的quicklist节点，
//将待插入元素插入到新的quicklist节点中，并将新的quicklist节点插入到当前节点和前一个quicklist节点之间；
//如果后插的时候目标元素是当前quicklist节点的最后一个entry，但是当前节点和当前节点的下一个节点都满了，则需要创建一个新的quicklist节点，
//将待插入元素插入到新的quicklist节点中，并将新的quicklist节点插入到当前节点和下一个quicklist节点之间；
D("\tprovisioning new node...");
new_node = quicklistCreateNode();
new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
new_node->count++;
quicklistNodeUpdateSz(new_node);
__quicklistInsertNode(quicklist, node, new_node, after);
} else if (full) {
/* else, node is full we need to split it. */
/* covers both after and !after cases */
//如果当前节点满了，且前插的时候目标节点不是当前quicklist节点的第一个entry，或者后插的时候目标节点不是当前quicklist节点的最后一个entry，则需要将当前节点拆分为两个quicklist节点
D("\tsplitting node...");
quicklistDecompressNodeForUse(node);
new_node = _quicklistSplitNode(node, entry->offset, after);
new_node->zl = ziplistPush(new_node->zl, value, sz,
after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
new_node->count++;
quicklistNodeUpdateSz(new_node);
__quicklistInsertNode(quicklist, node, new_node, after);
_quicklistMergeNodes(quicklist, node);
}
quicklist->count++;
}插入过程的流程图如下图所示
五、rpush
rpush命令可以将一个或多个值 value 插入到列表 key 的表尾(最右边)。rpush命令会调用pushGenericCommand()方法来执行push操作。
void pushGenericCommand(client *c, int where) {
int j, pushed = 0;
//根据key获取列表
robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
//对列表进行判断，如果不是OBJ_LIST类型的对象，则返回
if (lobj && lobj->type != OBJ_LIST) {
addReply(c,shared.wrongtypeerr);
return;
}
//开始遍历命令中需要push的值
for (j = 2; j < c->argc; j++) {
//若列表为空，则创建一个quicklist对象，并与key建立关系
if (!lobj) {
lobj = createQuicklistObject();
quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
server.list_compress_depth);
dbAdd(c->db,c->argv[1],lobj);
}
//将需要push的值push到列表中
listTypePush(lobj,c->argv[j],where);
pushed++;
}
//返回列表的长度
addReplyLongLong(c, (lobj ? listTypeLength(lobj) : 0));
//若进行了push操作，则进行键空间事件通知
if (pushed) {
char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
signalModifiedKey(c->db,c->argv[1]);
notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
}
server.dirty += pushed;
}rpush命令比较简单，这里需要注意的是，如果之前有客户端执行了brpop命令，当列表为空的时候，会让客户端阻塞，这是如果其他客户端执行了rpush操作，redis是如何通知执行了brpop命令而阻塞的客户端它所关注的列表有值了呢？答案就在pushGenericCommand()中。在执行pushGenericCommand()时，如果列表为空，会执行dbAdd()方法：
void dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr);
int retval = dictAdd(db->dict, copy, val);
serverAssertWithInfo(NULL,key,retval == DICT_OK);
if (val->type == OBJ_LIST) signalListAsReady(db, key);
if (server.cluster_enabled) slotToKeyAdd(key);
}dbAdd()函数会执行signalListAsReady()函数，该函数会将命令中的key保存在db->ready_keys字典中。
在命令执行函数processCommand()中，会通知由于执行了brpop而阻塞的客户端。
int processCommand(client *c) {
......
/* Exec the command */
if (c->flags & CLIENT_MULTI &&
c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
{
queueMultiCommand(c);
addReply(c,shared.queued);
} else {
call(c,CMD_CALL_FULL);
c->woff = server.master_repl_offset;
if (listLength(server.ready_keys))
handleClientsBlockedOnLists();
}
return C_OK;
}调用完call()后，会判断ready_keys的长度是否为0，若不为0，则会执行handleClientsBlockedOnLists()函数，通知阻塞的客户端。
六、命令的发布订阅消息
redis在执行完命令后，通常会发生键空间的通知。而发送通知是通过notifyKeyspaceEvent()实现的。
void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) {
sds chan;
robj *chanobj, *eventobj;
int len = -1;
char buf[24];
/* If notifications for this class of events are off, return ASAP. */
if (!(server.notify_keyspace_events & type)) return;
//创建事件对象
eventobj = createStringObject(event,strlen(event));
/* __keyspace@<db>__:<key> <event> notifications. */
//键空间通知，即某个键执行了什么命令
if (server.notify_keyspace_events & NOTIFY_KEYSPACE) {
chan = sdsnewlen("__keyspace@",11);
len = ll2string(buf,sizeof(buf),dbid);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, key->ptr);
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(chanobj, eventobj);
decrRefCount(chanobj);
}
/* __keyevent@<db>__:<event> <key> notifications. */
//键事件通知，即某个命令被什么键执行了
if (server.notify_keyspace_events & NOTIFY_KEYEVENT) {
chan = sdsnewlen("__keyevent@",11);
if (len == -1) len = ll2string(buf,sizeof(buf),dbid);
chan = sdscatlen(chan, buf, len);
chan = sdscatlen(chan, "__:", 3);
chan = sdscatsds(chan, eventobj->ptr);
chanobj = createObject(OBJ_STRING, chan);
pubsubPublishMessage(chanobj, key);
decrRefCount(chanobj);
}
decrRefCount(eventobj);
}而通知的功能是靠函数pubsubPublishMessage()实现的。
在订阅消息时，有两种订阅途径，一种是订阅频道，另一个是订阅一个模式，即订阅所有与该模式想匹配的频道。
当一个客户端订阅频道后，redis服务器会将订阅信息存储在server.pubsub_channels中。如下图所示：
当一个客户端订阅频道后，redis服务器会将订阅信息存储在server.pubsub_patterns中。如下图所示：
发送消息时，分别需要遍历频道字典和模式列表
int pubsubPublishMessage(robj *channel, robj *message) {
int receivers = 0;
dictEntry *de;
listNode *ln;
listIter li;
/* Send to clients listening for that channel */
//发送频道订阅消息
de = dictFind(server.pubsub_channels,channel);
if (de) {
list *list = dictGetVal(de);
listNode *ln;
listIter li;
listRewind(list,&li);
while ((ln = listNext(&li)) != NULL) {
client *c = ln->value;
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.messagebulk);
addReplyBulk(c,channel);
addReplyBulk(c,message);
receivers++;
}
}
/* Send to clients listening to matching channels */
//发送模式订阅消息
if (listLength(server.pubsub_patterns)) {
listRewind(server.pubsub_patterns,&li);
channel = getDecodedObject(channel);
while ((ln = listNext(&li)) != NULL) {
pubsubPattern *pat = ln->value;
if (stringmatchlen((char*)pat->pattern->ptr,
sdslen(pat->pattern->ptr),
(char*)channel->ptr,
sdslen(channel->ptr),0)) {
addReply(pat->client,shared.mbulkhdr[4]);
addReply(pat->client,shared.pmessagebulk);
addReplyBulk(pat->client,pat->pattern);
addReplyBulk(pat->client,channel);
addReplyBulk(pat->client,message);
receivers++;
}
}
decrRefCount(channel);
}
return receivers;
}