Redis系列（八）：資料結構List雙向連結串列中阻塞版本之BLPOP、BRPOP和LINDEX、LINSERT、LRANGE命令詳解
阿新 • • 發佈：2020-06-30
1.BRPOP、BLPOP
BLPOP：
BLPOP是阻塞式列表的彈出原語。 它是命令LPOP的阻塞版本，這是因為當給定列表內沒有任何元素可供彈出的時候，
連線將被BLPOP命令阻塞。 當給定多個 key 引數時，按引數 key 的先後順序依次檢查各個列表，彈出第一個非空列表的頭元素。
BRPOP:
BRPOP是一個阻塞的列表彈出原語。 它是RPOP的阻塞版本，因為這個命令會在給定list無法彈出任何元素的時候阻塞連線。 該命令會按照給出的 key 順序檢視 list，並在找到的第一個非空 list 的尾部彈出一個元素。
請在BLPOP檔案 中檢視該命令的準確語義，因為BRPOP和BLPOP基本是完全一樣的，除了它們一個是從尾部彈出元素，而另一個是從頭部彈出元素。
時間複雜度 :O(1)
127.0.0.1:> rpush order
(integer)
127.0.0.1:> rpush order
(integer)
127.0.0.1:> rpush order
(integer)
127.0.0.1:> rpush order
(integer)
127.0.0.1:> rpush order
(integer)
127.0.0.1:> rpush order
(integer)
127.0.0.1:>
127.0.0.1:> brpop order
) "order"
) ""
127.0.0.1:> brpop order) "order"
) ""
127.0.0.1:> brpop order
) "order"
) ""
127.0.0.1:> brpop order
) "order"
) ""
127.0.0.1:> brpop order
) "order"
) ""
127.0.0.1:> brpop order
) "order"
) ""
(.54s)
127.0.0.1:>
原始碼解析
void blpopCommand(client *c) {
blockingPopGenericCommand(c,LIST_HEAD);
}
void brpopCommand(client *c) {
}
blockingPopGenericCommand
/* Blocking RPOP/LPOP */
void blockingPopGenericCommand(client *c, int where) {
robj *o;
mstime_t timeout;
int j; if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-],&timeout,UNIT_SECONDS)
!= C_OK) return; for (j = ; j < c->argc-; j++) {
o = lookupKeyWrite(c->db,c->argv[j]);
if (o != NULL) {
if (o->type != OBJ_LIST) {
addReply(c,shared.wrongtypeerr);
return;
} else {
if (listTypeLength(o) != ) {
/* Non empty list, this is like a non normal [LR]POP. */
char *event = (where == LIST_HEAD) ? "lpop" : "rpop";
robj *value = listTypePop(o,where);
serverAssert(value != NULL); addReplyArrayLen(c,);
addReplyBulk(c,c->argv[j]);
addReplyBulk(c,value);
decrRefCount(value);
notifyKeyspaceEvent(NOTIFY_LIST,event,
c->argv[j],c->db->id);
if (listTypeLength(o) == ) {
dbDelete(c->db,c->argv[j]);
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",
c->argv[j],c->db->id);
}
signalModifiedKey(c,c->db,c->argv[j]);
server.dirty++; /* Replicate it as an [LR]POP instead of B[LR]POP. */
rewriteClientCommandVector(c,,
(where == LIST_HEAD) ? shared.lpop : shared.rpop,
c->argv[j]);
return;
}
}
}
} /* If we are inside a MULTI/EXEC and the list is empty the only thing
* we can do is treating it as a timeout (even with timeout 0). */
if (c->flags & CLIENT_MULTI) {
addReplyNullArray(c);
return;
} /* If the list is empty or the key does not exists we must block */
blockForKeys(c,BLOCKED_LIST,c->argv + ,c->argc - ,timeout,NULL,NULL);
}
blockForKeys
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, robj *target, streamID *ids) {
dictEntry *de;
list *l;
int j;
c->bpop.timeout = timeout;
c->bpop.target = target;
if (target != NULL) incrRefCount(target);
for (j = ; j < numkeys; j++) {
/* Allocate our bkinfo structure, associated to each key the client
* is blocked for. */
bkinfo *bki = zmalloc(sizeof(*bki));
if (btype == BLOCKED_STREAM)
bki->stream_id = ids[j];
/* If the key already exists in the dictionary ignore it. */
if (dictAdd(c->bpop.keys,keys[j],bki) != DICT_OK) {
zfree(bki);
continue;
}
incrRefCount(keys[j]);
/* And in the other "side", to map keys -> clients */
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();
retval = dictAdd(c->db->blocking_keys,keys[j],l);
incrRefCount(keys[j]);
serverAssertWithInfo(c,keys[j],retval == DICT_OK);
} else {
l = dictGetVal(de);
}
listAddNodeTail(l,c);
bki->listnode = listLast(l);
}
blockClient(c,btype);
}
blockClient
void blockClient(client *c, int btype) {
c->flags |= CLIENT_BLOCKED;
c->btype = btype;
server.blocked_clients++;
server.blocked_clients_by_type[btype]++;
addClientToTimeoutTable(c);
}
addClientToTimeoutTable
void addClientToTimeoutTable(client *c) {
if (c->bpop.timeout == ) return;
uint64_t timeout = c->bpop.timeout;
unsigned char buf[CLIENT_ST_KEYLEN];
encodeTimeoutKey(buf,timeout,c);
if (raxTryInsert(server.clients_timeout_table,buf,sizeof(buf),NULL,NULL))
c->flags |= CLIENT_IN_TO_TABLE;
}
2.LINDEX
返回列表裡的元素的索引 index 儲存在 key 裡面。 下標是從0開始索引的，所以 0 是表示第一個元素，
1 表示第二個元素，並以此類推。 負數索引用於指定從列表尾部開始索引的元素。在這種方法下，-1 表示最後一個元素，-2 表示倒數第二個元素，並以此往前推。
當 key 位置的值不是一個列表的時候，會返回一個error。
時間複雜度：O(N)
127.0.0.1:> lpush mylist "World" "Hello"
(integer)
127.0.0.1:> lindex mylist
"Hello"
127.0.0.1:> lindex mylist
"World"
127.0.0.1:>
原始碼解析
void lindexCommand(client *c) {
robj *o = lookupKeyReadOrReply(c,c->argv[],shared.null[c->resp]);
if (o == NULL || checkType(c,o,OBJ_LIST)) return;
long index;
robj *value = NULL;
if ((getLongFromObjectOrReply(c, c->argv[], &index, NULL) != C_OK))
return;
if (o->encoding == OBJ_ENCODING_QUICKLIST) {
quicklistEntry entry;
if (quicklistIndex(o->ptr, index, &entry)) {
if (entry.value) {
value = createStringObject((char*)entry.value,entry.sz);
} else {
value = createStringObjectFromLongLong(entry.longval);
}
addReplyBulk(c,value);
decrRefCount(value);
} else {
addReplyNull(c);
}
} else {
serverPanic("Unknown list encoding");
}
}
/* Populate 'entry' with the element at the specified zero-based index
* where 0 is the head, 1 is the element next to head
* and so on. Negative integers are used in order to count
* from the tail, -1 is the last element, -2 the penultimate
* and so on. If the index is out of range 0 is returned.
*
* Returns 1 if element found
* Returns 0 if element not found */
int quicklistIndex(const quicklist *quicklist, const long long idx,
quicklistEntry *entry) {
quicklistNode *n;
unsigned long long accum = ;
unsigned long long index;
int forward = idx < ? : ; /* < 0 -> reverse, 0+ -> forward */ initEntry(entry);
entry->quicklist = quicklist; if (!forward) {
index = (-idx) - ;
n = quicklist->tail;
} else {
index = idx;
n = quicklist->head;
} if (index >= quicklist->count)
return ; 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 ; D("Found node: %p at accum %llu, idx %llu, sub+ %llu, sub- %llu", (void *)n,
accum, index, index - accum, (-index) - + accum); entry->node = n;
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) - + accum;
} quicklistDecompressNodeForUse(entry->node);
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 ;
}
3.LINSERT
把 value 插入存於 key 的列表中在基準值 pivot 的前面或後面。
當 key 不存在時，這個list會被看作是空list，任何操作都不會發生。
當 key 存在，但儲存的不是一個list的時候，會返回error。
時間複雜度：O(N)
127.0.0.1:> linsert mylist before "World" "There"
(integer)
127.0.0.1:> lrange mylist -
) "Hello"
) "There"
) "World"
127.0.0.1:>
4.LRANGE
返回儲存在 key 的列表裡指定範圍內的元素。 start 和 end 偏移量都是基於0的下標，即list的第一個元素下標是0（list的表頭），第二個元素下標是1，以此類推。
偏移量也可以是負數，表示偏移量是從list尾部開始計數。 例如， -1 表示列表的最後一個元素，-2 是倒數第二個，以此類推。
時間複雜度：O(S+N)
127.0.0.1:> lrange mylist -
) "Hello"
) "There"
) "World"
127.0.0.1:> lrange mylist
) "Hello"
) "There"
127.0.0.1:>
原始碼解析
void lrangeCommand(client *c) {
robj *o;
long start, end, llen, rangelen;
if ((getLongFromObjectOrReply(c, c->argv[], &start, NULL) != C_OK) ||
(getLongFromObjectOrReply(c, c->argv[], &end, NULL) != C_OK)) return;
if ((o = lookupKeyReadOrReply(c,c->argv[],shared.emptyarray)) == NULL
|| checkType(c,o,OBJ_LIST)) return;
llen = listTypeLength(o);
/* convert negative indexes */
if (start < ) start = llen+start;
if (end < ) end = llen+end;
if (start < ) start = ;
/* Invariant: start >= 0, so this test will be true when end < 0.
* The range is empty when start > end or start >= length. */
if (start > end || start >= llen) {
addReply(c,shared.emptyarray);
return;
}
if (end >= llen) end = llen-;
rangelen = (end-start)+;
/* Return the result in form of a multi-bulk reply */
addReplyArrayLen(c,rangelen);
if (o->encoding == OBJ_ENCODING_QUICKLIST) {
listTypeIterator *iter = listTypeInitIterator(o, start, LIST_TAIL);
while(rangelen--) {
listTypeEntry entry;
listTypeNext(iter, &entry);
quicklistEntry *qe = &entry.entry;
if (qe->value) {
addReplyBulkCBuffer(c,qe->value,qe->sz);
} else {
addReplyBulkLongLong(c,qe->longval);
}
}
listTypeReleaseIterator(iter);
} else {
serverPanic("List encoding is not QUICKLIST!");
}
}