Redis中的阻塞性命令可以让客户端被阻塞等待特定事件发生,常用在需要等待可用数据、等待结果等场景。

阻塞命令概述

等待数据

当数据不存在时,阻塞执行client直到数据符合条件或者超时。包括以下命令:

• BRPOP/BLPOP: 从列表左右两端阻塞弹出元素,直到有元素为止。

• BRPOPLPUSH: 阻塞式列表元素移动。

• BLMOVE: 阻塞式的列表元素移动。

• XREAD: 阻塞读取Stream数据。

• XREADGROUP: 阻塞读取Stream的消费者组数据。

• BZPOPMIN/BZPOPMAX: 阻塞并返回有序集合弹出的最小/大元素。

实现原理

• 客户端向服务器发送一个阻塞命令,如BLPOP key 1。

• 服务器将这个命令和客户端信息加入到一个字典中,键为阻塞的键,值为一个链表结构,链表中的每个元素记录一个客户端的信息。

• 当字典中有多个客户端阻塞在同一个键上时,它们会被添加到同一个链表中。

• 同时服务器会将客户端添加到TimeoutTable, 以便在beforesleep 函数中定时检测客户端阻塞是否超时

• 目标键有新操作时,服务器会获取等待key对应的客户端列表, 并向客户端发送回复。

• beforesleep 函数中定时检测到客户端阻塞超时,向客户端发送超时回复包

通过字典和链表结构,服务器可以高效管理多个客户端在同一个键上的阻塞,并在合适的时候解除阻塞

等待从节点数据同步

当还没有足够数量的从节点执行完client最新的写命令时，阻塞当前client直到有足够副本执行命令或者超时。

• WAIT: 阻塞等待指定键的修改。

实现原理

• 客户端向服务器发送wait 命令。

• 服务器在处理wait命令时判断 是否已有存在足够数量的从节点已经执行完当前客户端最后一条写命令，如果足够直接返回。

• 如果数量不足够,将本客户端 添加到server.clientswaitingacks 列表中, 并且设置客户端block 标记，添加到TimeoutTable; 设置 server.getackfrom_slaves以便在beforesleep 中向各个从节点发送请求 获取执行副本复制log 的offset

• 在beforesleep 遍历server.clientswaitingacks 客户端列表; 对每个客户端判断是否已有存在足够数量的从节点已经执行完当前客户端最后一条写命令,如果是取消客户端阻塞。

• beforesleep 函数中定时检测到客户端阻塞超时,向客户端发送超时回复包

暂停客户端执行

执行client PAUSE/ 或者failover 命令时, 会让服务器使所有客户端(不管是已经存在的还是后面新建立链接的)进入挂起状态, 直到超时。

• client PAUSE milliseconds 让服务器使所有客户端(不管存在还是新建立链接)进入挂起状态,直到milliseconds 时间后

• client UNPAUSE 会让服务器取消pause客户端并且唤醒所有被暂停的客户端。

• failover option 哨兵提供了 failover 命令来手动触发主从切换。执行切换时先pause 执行写命令的客户端; 等待从节点执行完所有主节点写命令; 然后主把自己变为从节点, 原来的某个从节点切换成主; 最后unpause 执行写命令客户端

实现原理

• 客户端向服务器发送client PAUSE 命令, 服务器设置全局pause标记(server.clientpausetype)并且记录 取消阻塞的 时间(server.clientpauseend_time)。

• 服务器收到任意客户端命令,准备执行前判断是否需要阻塞客户端, 如果不需要阻塞即执行命令。

• 如果需要阻塞全部客户端, 或者阻塞写命令并且此客户端的命令时写命令 阻塞此客户端, 将客户端添加到server.paused_clients 列表中;

• 客户端向服务器发送client UNPAUSE 命令, 取消全局pause标记; 唤醒已经被阻塞的所有客户端。

• beforesleep 函数中定时检测是否已经到了server.clientpauseend_time时间, 如果是取消全局pause标记 ; 唤醒已经被阻塞的所有客户端。

源码剖析

redis源码版本：6.2.5 本文从以下几个方面带着大家走读代码：

• 阻塞等待数据的BLPOP

• 等待从节点数据同步

• 暂停客户端执行(client PAUSE/UNPAUSE )

主从同步 相关变量

//path:src/server.h
/* Client block type (btype field in client structure)
 * if CLIENT_BLOCKED flag is set. */
// 客户端block 类型
#define BLOCKED_NONE 0    /* Not blocked, no CLIENT_BLOCKED flag set. */
// list block 命令 (数据类型block)
#define BLOCKED_LIST 1    /* BLPOP & co. */
// 从节点同步执行命令(wait命令)
#define BLOCKED_WAIT 2    /* WAIT for synchronous replication. */
// modlue相关的block
#define BLOCKED_MODULE 3  /* Blocked by a loadable module. */
// stream 类型xread命令 (数据类型block)
#define BLOCKED_STREAM 4  /* XREAD. */
// zset bzpop 命令 (数据类型block)
#define BLOCKED_ZSET 5    /* BZPOP et al. */
// 暂停block, 用于主动故障转移或者屏蔽redis服务读写
#define BLOCKED_PAUSE 6   /* Blocked by CLIENT PAUSE */
#define BLOCKED_NUM 7     /* Number of blocked states. */
// 阻塞客户端flags 会打上标记
// 打上标记后processInputBuffer不解析客户端字节流处理命令
// unblock 时取消此标记
#define CLIENT_BLOCKED (1<<4) /* The client is waiting in a blocking operation */
// unblocked 时打上此标记, 并且放置到server.unblocked_clients列表中
// 在beforesleep调用processUnblockedClients 解析客户端缓冲区字节流，并且处理命令
#define CLIENT_UNBLOCKED (1<<7) /* This client was unblocked and is stored in
/* This structure holds the blocking operation state for a client.
 * The fields used depend on client->btype. */
// 用于client 记录阻塞(block)的具体信息
typedef struct blockingState {
    /* Generic fields. */
    // 通用字段, block超时时间
    mstime_t timeout;       /* Blocking operation timeout. If UNIX current time
                             * is > timeout then the operation timed out. */
    // 用于 list,zset,stream(因等待数据block类型)
    /* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM */
    dict *keys;             /* The keys we are waiting to terminate a blocking
                             * operation such as BLPOP or XREAD. Or NULL. */
    // BLMOVE 的目标列表
    robj *target;           /* The key that should receive the element,
                             * for BLMOVE. */
    // 用于记录从列表 开始 pop/push
    // 用于 BLPOP, BRPOP and BLMOVE
    struct listPos {
        int wherefrom;      /* Where to pop from */
        int whereto;        /* Where to push to */
    } listpos;              /* The positions in the src/dst lists
                             * where we want to pop/push an element
                             * for BLPOP, BRPOP and BLMOVE. */
    /* BLOCK_STREAM */
    size_t xread_count;     /* XREAD COUNT option. */
    robj *xread_group;      /* XREADGROUP group name. */
    robj *xread_consumer;   /* XREADGROUP consumer name. */
    int xread_group_noack;
    /* BLOCKED_WAIT */
    // wait命令使用(wait命令用于block到副本同步完成)
    // 等待多少个副本同步完成
    int numreplicas;        /* Number of replicas we are waiting for ACK. */
    // 同步 副本复杂log 的offset
    long long reploffset;   /* Replication offset to reach. */
    /* BLOCKED_MODULE */
    void *module_blocked_handle; /* RedisModuleBlockedClient structure.
                                    which is opaque for the Redis core, only
                                    handled in module.c. */
} blockingState;
// 每个客户端链接对应一个client数据结构
typedef struct client {
// ...
// 标记位
uint64_t flags; /* Client flags: CLIENT_* macros. */
// 用于client 记录阻塞(block)的具体信息
blockingState bpop; /* blocking state */
// ...
}
struct redisServer {
// ...
    // 用于阻塞客户端超时处理
    rax *clients_timeout_table; /* Radix tree for blocked clients timeouts. */
// ...
}
// path: src/networking.c
/* This function is called every time, in the client structure 'c', there is
 * more query buffer to process, because we read more data from the socket
 * or because a client was blocked and later reactivated, so there could be
 * pending query buffer, already representing a full command, to process. */
// 解析client buf协议包
// 根据是否在io线程上下文决定是否执行命令
// 如果在io上下文,解析到第一个完整协议包后保存return
// 否则一直解析协议处理命令，直到解析不出完整协议包
// 设置了 CLIENT_BLOCKED 后processInputBuffer不解析客户端字节流处理命令
void processInputBuffer(client *c) {
    /* Keep processing while there is something in the input buffer */
    while(c->qb_pos < sdslen(c->querybuf)) {
        /* Immediately abort if the client is in the middle of something. */
        // 设置了 CLIENT_BLOCKED 直接不解析客户端协议处理,直接返回
        if (c->flags & CLIENT_BLOCKED) break;
        /* Don't process more buffers from clients that have already pending
         * commands to execute in c->argv. */
        // CLIENT_PENDING_COMMAND 代表已经解析到可执行命令但是还不能真正执行(比如在io多线程读取上下文中)
        // 因此直接退出
        if (c->flags & CLIENT_PENDING_COMMAND) break;
        /* Don't process input from the master while there is a busy script
         * condition on the slave. We want just to accumulate the replication
         * stream (instead of replying -BUSY like we do with other clients) and
         * later resume the processing. */
        if (server.lua_timedout && c->flags & CLIENT_MASTER) break;
        /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
         * written to the client. Make sure to not let the reply grow after
         * this flag has been set (i.e. don't process more commands).
         *
         * The same applies for clients we want to terminate ASAP. */
        if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break;
        /* Determine request type when unknown. */
        if (!c->reqtype) {
            if (c->querybuf[c->qb_pos] == '*') {
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                c->reqtype = PROTO_REQ_INLINE;
            }
        }
        // 从query_buf 根据redis协议解包
        if (c->reqtype == PROTO_REQ_INLINE) {
            if (processInlineBuffer(c) != C_OK) break;
            /* If the Gopher mode and we got zero or one argument, process
             * the request in Gopher mode. To avoid data race, Redis won't
             * support Gopher if enable io threads to read queries. */
            if (server.gopher_enabled && !server.io_threads_do_reads &&
                ((c->argc == 1 && ((char*)(c->argv[0]->ptr))[0] == '/') ||
                  c->argc == 0))
            {
                processGopherRequest(c);
                resetClient(c);
                c->flags |= CLIENT_CLOSE_AFTER_REPLY;
                break;
            }
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }
        /* Multibulk processing could see a <= 0 length. */
        if (c->argc == 0) {
            resetClient(c);
        } else {
            // 解包成功进入这里
            /* If we are in the context of an I/O thread, we can't really
             * execute the command here. All we can do is to flag the client
             * as one that needs to process the command. */
            // CLIENT_PENDING_READ被设置，代表io线程进行读取和解析
            // 目前处于解析中, 因此处在io多线程中，不能处理命令
            // 打上CLIENT_PENDING_COMMAND(有命令待处理) 返回
            if (c->flags & CLIENT_PENDING_READ) {
                c->flags |= CLIENT_PENDING_COMMAND;
                break;
            }
            // 走到这里说明可以直接处理命令
            /* We are finally ready to execute the command. */
            if (processCommandAndResetClient(c) == C_ERR) {
                /* If the client is no longer valid, we avoid exiting this
                 * loop and trimming the client buffer later. So we return
                 * ASAP in that case. */
                return;
            }
        }
    }
    /* Trim to pos */
    if (c->qb_pos) {
        sdsrange(c->querybuf,c->qb_pos,-1);
        c->qb_pos = 0;
    }
}
// path :src/blocked.c
/* This function is called in the beforeSleep() function of the event loop
 * in order to process the pending input buffer of clients that were
 * unblocked after a blocking operation. */
// 处理unblock 列表
// 在beforesleep调用
void processUnblockedClients(void) {
    listNode *ln;
    client *c;
    // 遍历server.unblocked_clients 列表
    while (listLength(server.unblocked_clients)) {
        ln = listFirst(server.unblocked_clients);
        serverAssert(ln != NULL);
        c = ln->value;
        listDelNode(server.unblocked_clients,ln);
        // 去掉unblock标记
        c->flags &= ~CLIENT_UNBLOCKED;
        /* Process remaining data in the input buffer, unless the client
         * is blocked again. Actually processInputBuffer() checks that the
         * client is not blocked before to proceed, but things may change and
         * the code is conceptually more correct this way. */
        if (!(c->flags & CLIENT_BLOCKED)) {
            /* If we have a queued command, execute it now. */
            // 执行pending 的命令
            // 之前解析出来当时没有执行的命令
            if (processPendingCommandsAndResetClient(c) == C_ERR) {
                continue;
            }
            /* Then process client if it has more data in it's buffer. */
            // 之前设置CLIENT_BLOCKED, tcp字节流被读取到缓存区中但是没有解析协议处理命令
            // 解析缓冲区buf(读), 处理更多命令
            if (c->querybuf && sdslen(c->querybuf) > 0) {
                processInputBuffer(c);
            }
        }
    }
}

阻塞等待数据的BLPOP

BLPOP命令处理

// path:src/t_list.c
/* BLPOP <key> [<key> ...] <timeout> */
// BLPOP 命令处理函数
void blpopCommand(client *c) {
    blockingPopGenericCommand(c,LIST_HEAD);
}
/* Blocking RPOP/LPOP */
// 处理 RPOP/LPOP命令
void blockingPopGenericCommand(client *c, int where) {
    robj *o;
    mstime_t timeout;
    int j;
    // 获取超时时间
    if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout,UNIT_SECONDS)
        != C_OK) return;
    for (j = 1; j < c->argc-1; j++) {
        // 循环遍历key,获取val
        o = lookupKeyWrite(c->db,c->argv[j]);
        if (o != NULL) {
            if (checkType(c,o,OBJ_LIST)) {
                // 判断不是LIST 类型,返回(失败了)
                return;
            } else {
                // LIST类型
                if (listTypeLength(o) != 0) {
                    // 判断LIST有元素,不用阻塞，直接取元素返回
                    /* Non empty list, this is like a normal [LR]POP. */
                    robj *value = listTypePop(o,where);
                    serverAssert(value != NULL);
                    addReplyArrayLen(c,2);
                    addReplyBulk(c,c->argv[j]);
                    addReplyBulk(c,value);
                    decrRefCount(value);
                    listElementsRemoved(c,c->argv[j],where,o,1);
                    /* Replicate it as an [LR]POP instead of B[LR]POP. */
                    rewriteClientCommandVector(c,2,
                        (where == LIST_HEAD) ? shared.lpop : shared.rpop,
                        c->argv[j]);
                    return;
                }
            }
        }
    }
    /* If we are not allowed to block the client, the only thing
     * we can do is treating it as a timeout (even with timeout 0). */
    // 客户端标识为不可阻塞,直接当成超时返回
    if (c->flags & CLIENT_DENY_BLOCKING) {
        addReplyNullArray(c);
        return;
    }
    /* If the keys do not exist we must block */
    struct listPos pos = {where};
    // block client(等待某些key)
    // 用于 BLOCKED_LIST, BLOCKED_ZSET or BLOCKED_STREAM
    blockForKeys(c,BLOCKED_LIST,c->argv + 1,c->argc - 2,timeout,NULL,&pos,NULL);
}
// path:src/blocked.c
// block client(等待某些key)
// 用于 BLOCKED_LIST, BLOCKED_ZSET or BLOCKED_STREAM
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, robj *target, struct listPos *listpos, streamID *ids) {
    dictEntry *de;
    list *l;
    int j;
    // 记录block的timeout等信息
    c->bpop.timeout = timeout;
    // target用于记录BLMOVE的目标list
    c->bpop.target = target;
    if (listpos != NULL) c->bpop.listpos = *listpos;
    if (target != NULL) incrRefCount(target);
    for (j = 0; 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. */
        // client 记录需要等待 key 而block
        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 */
        // 在全局c->db->blocking_keys中记录那些客户端因为那个key block
        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);
        }
        // 将client 添加到key 对应的list中
        listAddNodeTail(l,c);
        bki->listnode = listLast(l);
    }
    // 设置CLIENT_BLOCKE标记 并且添加到TimeoutTable中
    blockClient(c,btype);
}
/* Block a client for the specific operation type. Once the CLIENT_BLOCKED
 * flag is set client query buffer is not longer processed, but accumulated,
 * and will be processed when the client is unblocked. */
// 设置CLIENT_BLOCKE标记 并且添加到TimeoutTable中
void blockClient(client *c, int btype) {
    // 设置CLIENT_BLOCKE标记
    c->flags |= CLIENT_BLOCKED;
    c->btype = btype;
    server.blocked_clients++;
    server.blocked_clients_by_type[btype]++;
    // 添加到TimeoutTable中
    addClientToTimeoutTable(c);
    if (btype == BLOCKED_PAUSE) {
        // 如果是pause, 还添加到server.paused_client中
        // 并且添加CLIENT_PENDING_COMMAN标记
        listAddNodeTail(server.paused_clients, c);
        c->paused_list_node = listLast(server.paused_clients);
        /* Mark this client to execute its command */
        c->flags |= CLIENT_PENDING_COMMAND;
    }
}
/* Add the specified client id / timeout as a key in the radix tree we use
 * to handle blocked clients timeouts. The client is not added to the list
 * if its timeout is zero (block forever). */
// 添加到TimeoutTable中
void addClientToTimeoutTable(client *c) {
    if (c->bpop.timeout == 0) return;
    uint64_t timeout = c->bpop.timeout;
    unsigned char buf[CLIENT_ST_KEYLEN];
    encodeTimeoutKey(buf,timeout,c);
    // 添加到TimeoutTable中
    if (raxTryInsert(server.clients_timeout_table,buf,sizeof(buf),NULL,NULL))
        c->flags |= CLIENT_IN_TO_TABLE;
}

调用链

void blpopCommand(client *c) // BLPOP命令处理函数
    - void blockingPopGenericCommand(client *c, int where)  // 真正处理 BRPOP/BLPOP命令 逻辑
        - 遍历命令中的key列表
            - key对应的value 不是LIST类型，报错返回
            - key对应的LIST有元素，取数据返回
        - 客户端标识为不可阻塞,直接当成超时返回
        - void blockForKeys(client *c, ...) // block client(等待某些key)
            - - 将block的具体全局信息保存在  c->bpop
            - 遍历命令中的key列表
                - 在c->bpop.keys保存client 等待那些key数据
                - 在db的全局dict,即c->db->blocking_keys中对应key的列表中保存当前client,代表client等待这个key的数据
            - void blockClient(client *c, int btype) // 设置CLIENT_BLOCKE标记 并且添加到TimeoutTable中
                - c->flags |= CLIENT_BLOCKED; // 设置CLIENT_BLOCKE标记
                - void addClientToTimeoutTable(client *c) // 添加到TimeoutTable中
                    - raxTryInsert(server.clients_timeout_table,buf,sizeof(buf),NULL,NULL))

综上可知:

• 当LIST 有数据时直接返回数据

• LIST不存在(没数据时)，设置client block标记，并且将client等待key的信息 保存在 c->bpop.keys 和 c->db->blocking_keys中

• 为了超时解除客户端阻塞, 将client 保存在server.clientstimeouttable 的数据结构中

LIST添加数据解除阻塞

// path:src/t_list.c
/* LPUSH <key> <element> [<element> ...] */
// LPUSH 命令处理函数
void lpushCommand(client *c) {
    pushGenericCommand(c,LIST_HEAD,0);
}
/*-----------------------------------------------------------------------------
 * List Commands
 *----------------------------------------------------------------------------*/
/* Implements LPUSH/RPUSH/LPUSHX/RPUSHX. 
 * 'xx': push if key exists. */
// 处理list push 元素
void pushGenericCommand(client *c, int where, int xx) {
    int j;
    // 找到对应的value
    robj *lobj = lookupKeyWrite(c->db, c->argv[1]);
    // 检查是否是LIST类型
    if (checkType(c,lobj,OBJ_LIST)) return;
    if (!lobj) {
        // 不存在
        if (xx) {
            addReply(c, shared.czero);
            return;
        }
        // 创建LIST OBJECT
        lobj = createQuicklistObject();
        quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
                            server.list_compress_depth);
        //  添加key value
        // 当LIST  不存在时才会block客户端
        // 存在表示至少有一个元素，不会block客户端
        // 所以在dbAdd 调用signalKeyAsReady 即可保证LIST 添加元素时及时唤醒客户端
        dbAdd(c->db,c->argv[1],lobj);
    }
    // 添加元素到LIST 对象中
    for (j = 2; j < c->argc; j++) {
        listTypePush(lobj,c->argv[j],where);
        server.dirty++;
    }
    // 添加回复包给客户端
    addReplyLongLong(c, listTypeLength(lobj));
    char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
    signalModifiedKey(c,c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
}
// path:src/db.c
/* Add the key to the DB. It's up to the caller to increment the reference
 * counter of the value if needed.
 *
 * The program is aborted if the key already exists. */
// 添加key value
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);
    // 添加kv 后, 调用signalKeyAsReady将key 加入到server.ready_keys
    // 后续在beforeSleep 处理server.ready_keys unblock相关客户端
    // 当LIST 或者ZSET 不存在时才会block客户端
    // 存在表示至少有一个元素，不会block客户端
    // 所以在dbAdd 调用signalKeyAsReady 即可保证LIST 或者ZSET 添加元素时及时唤醒客户端
    signalKeyAsReady(db, key, val->type);
    if (server.cluster_enabled) slotToKeyAdd(key->ptr);
}
// path:src/blocked.c
/* 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 handleClientsBlockedOnKeys() */
void signalKeyAsReady(redisDb *db, robj *key, int type) {
    readyList *rl;
    /* Quick returns. */
    int btype = getBlockedTypeByType(type);
    // OBJ_LIST, OBJ_ZSET, OBJ_STREAM, OBJ_MODULE 才可能block 
    if (btype == BLOCKED_NONE) {
        // 如果key对应的value 不会引起任何block,直接返回
        /* The type can never block. */
        return;
    }
    if (!server.blocked_clients_by_type[btype] &&
        !server.blocked_clients_by_type[BLOCKED_MODULE]) {
        // 不存在被block住的客户端,直接返回
        /* No clients block on this type. Note: Blocked modules are represented
         * by BLOCKED_MODULE, even if the intention is to wake up by normal
         * types (list, zset, stream), so we need to check that there are no
         * blocked modules before we do a quick return here. */
        return;
    }
    /* No clients blocking for this key? No need to queue it. */
    // 没有client因为此key block住,直接返回
    if (dictFind(db->blocking_keys,key) == NULL) return;
    /* Key was already signaled? No need to queue it again. */
    // 已经在db->ready_keys 中，直接返回
    if (dictFind(db->ready_keys,key) != NULL) return;
    /* Ok, we need to queue this key into server.ready_keys. */
    // 添加到 server.ready_keys中, beforeSleep时统一处理
    rl = zmalloc(sizeof(*rl));
    rl->key = key;
    rl->db = db;
    incrRefCount(key);
    // 添加到 server.ready_keys中, beforeSleep时统一处理
    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);
    serverAssert(dictAdd(db->ready_keys,key,NULL) == DICT_OK);
}
// beforesleep 中调用
// 遍历server.ready_keys 处理因为 key(也即数据)block的客户端
void handleClientsBlockedOnKeys(void) {
    // server.ready_keys保存所有此次epoll_wait之后执行命令后有变动的key
    // server.ready_keys链表元素大于0
    while(listLength(server.ready_keys) != 0) {
        list *l;
        /* Point server.ready_keys to a fresh list and save the current one
         * locally. This way as we run the old list we are free to call
         * signalKeyAsReady() that may push new elements in server.ready_keys
         * when handling clients blocked into BLMOVE. */
        l = server.ready_keys;
        server.ready_keys = listCreate();
         // 遍历server.ready_keys链表
        while(listLength(l) != 0) {
            listNode *ln = listFirst(l);
            readyList *rl = ln->value;
            /* First of all remove this key from db->ready_keys so that
             * we can safely call signalKeyAsReady() against this key. */
            // 将当前处理的key 从db->ready_keys 删除
            dictDelete(rl->db->ready_keys,rl->key);
            /* Even if we are not inside call(), increment the call depth
             * in order to make sure that keys are expired against a fixed
             * reference time, and not against the wallclock time. This
             * way we can lookup an object multiple times (BLMOVE does
             * that) without the risk of it being freed in the second
             * lookup, invalidating the first one.
             * See https://github.com/redis/redis/pull/6554. */
            server.fixed_time_expire++;
            updateCachedTime(0);
            /* Serve clients blocked on the key. */
            // 获取key对应的val
            robj *o = lookupKeyWrite(rl->db,rl->key);
            if (o != NULL) {
                // 根据val 类型做具体处理
                if (o->type == OBJ_LIST)
                    // LIST
                    serveClientsBlockedOnListKey(o,rl);
                else if (o->type == OBJ_ZSET)
                    // ZSET
                    serveClientsBlockedOnSortedSetKey(o,rl);
                else if (o->type == OBJ_STREAM)
                    // OBJ_STREAM
                    serveClientsBlockedOnStreamKey(o,rl);
                /* We want to serve clients blocked on module keys
                 * regardless of the object type: we don't know what the
                 * module is trying to accomplish right now. */
                serveClientsBlockedOnKeyByModule(rl);
            }
            server.fixed_time_expire--;
            /* Free this item. */
            decrRefCount(rl->key);
            zfree(rl);
            listDelNode(l,ln);
        }
        listRelease(l); /* We have the new list on place at this point. */
    }
}
/* Helper function for handleClientsBlockedOnKeys(). This function is called
 * when there may be clients blocked on a list key, and there may be new
 * data to fetch (the key is ready). */
// 用于处理 LIST添加了新元素时, 因为LIST阻塞的 client 
void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
    /* We serve clients in the same order they blocked for
     * this key, from the first blocked to the last. */
    // 获取key 对应的 block 客户端列表
    dictEntry *de = dictFind(rl->db->blocking_keys,rl->key);
    if (de) {
        list *clients = dictGetVal(de);
        int numclients = listLength(clients);
        // 遍历block client 列表
        while(numclients--) {
            listNode *clientnode = listFirst(clients);
            client *receiver = clientnode->value;
            // 不是 BLOCKED_LIST) 跳过
            if (receiver->btype != BLOCKED_LIST) {
                /* Put at the tail, so that at the next call
                 * we'll not run into it again. */
                listRotateHeadToTail(clients);
                continue;
            }
            robj *dstkey = receiver->bpop.target;
            int wherefrom = receiver->bpop.listpos.wherefrom;
            int whereto = receiver->bpop.listpos.whereto;
            // 从列表获取对应元素
            robj *value = listTypePop(o, wherefrom);
            if (value) {
                /* Protect receiver->bpop.target, that will be
                 * freed by the next unblockClient()
                 * call. */
                if (dstkey) incrRefCount(dstkey);
                monotime replyTimer;
                elapsedStart(&replyTimer);
                // 处理具体LIST block 命令
                // BLMOVE BLPOP BRPOP
                if (serveClientBlockedOnList(receiver,
                    rl->key,dstkey,rl->db,value,
                    wherefrom, whereto) == C_ERR)
                {
                    /* If we failed serving the client we need
                     * to also undo the POP operation. */
                    // 失败，value放回列表
                    listTypePush(o,value,wherefrom);
                }
                updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
                // unblock 客户端
                unblockClient(receiver);
                if (dstkey) decrRefCount(dstkey);
                decrRefCount(value);
            } else {
                // LIST 没有元素了,break
                break;
            }
        }
    }
    // LIST 为空，直接删除KV
    if (listTypeLength(o) == 0) {
        dbDelete(rl->db,rl->key);
        notifyKeyspaceEvent(NOTIFY_GENERIC,"del",rl->key,rl->db->id);
    }
    /* We don't call signalModifiedKey() as it was already called
     * when an element was pushed on the list. */
}
// 当LIST 有变动(新增元素)时 处理 BLPOP 或者 BRPOR 或者 BLMOVE命令
// 向某个被block的客户端 重新处理命令,发送数据回复包
int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto)
{
    robj *argv[5];
    if (dstkey == NULL) {
        // dstkey为空，即为BLPOP 或者 BRPOR命令
        /* Propagate the [LR]POP operation. */
        argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop :
                                             shared.rpop;
        argv[1] = key;
        propagate((wherefrom == LIST_HEAD) ?
            server.lpopCommand : server.rpopCommand,
            db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL);
        /* BRPOP/BLPOP */
        // POP出来的元素值发送给客户端
        addReplyArrayLen(receiver,2);
        addReplyBulk(receiver,key);
        addReplyBulk(receiver,value);
        /* Notify event. */
        char *event = (wherefrom == LIST_HEAD) ? "lpop" : "rpop";
        notifyKeyspaceEvent(NOTIFY_LIST,event,key,receiver->db->id);
    } else {
        /* BLMOVE */
        // 将POP出来的元素压到另外一个列表
        // 先获取dstkey 对应LIST Value
        robj *dstobj =
            lookupKeyWrite(receiver->db,dstkey);
        if (!(dstobj &&
             checkType(receiver,dstobj,OBJ_LIST)))
        {
            // 存在 并且是 LIST 类型
            lmoveHandlePush(receiver,dstkey,dstobj,value,whereto);
            /* Propagate the LMOVE/RPOPLPUSH operation. */
            int isbrpoplpush = (receiver->lastcmd->proc == brpoplpushCommand);
            argv[0] = isbrpoplpush ? shared.rpoplpush : shared.lmove;
            argv[1] = key;
            argv[2] = dstkey;
            argv[3] = getStringObjectFromListPosition(wherefrom);
            argv[4] = getStringObjectFromListPosition(whereto);
            propagate(isbrpoplpush ? server.rpoplpushCommand : server.lmoveCommand,
                db->id,argv,(isbrpoplpush ? 3 : 5),
                PROPAGATE_AOF|
                PROPAGATE_REPL);
            /* Notify event ("lpush" or "rpush" was notified by lmoveHandlePush). */
            notifyKeyspaceEvent(NOTIFY_LIST,wherefrom == LIST_TAIL ? "rpop" : "lpop",
                                key,receiver->db->id);
        } else {
            // 不是LIST 类型, 只能报错
            /* BLMOVE failed because of wrong
             * destination type. */
            return C_ERR;
        }
    }
    return C_OK;
}
/* Unblock a client calling the right function depending on the kind
 * of operation the client is blocking for. */
// 取消客户端block
// 根据客户端不同的block类型(原因)调用不同函数进行处理
void unblockClient(client *c) {
    if (c->btype == BLOCKED_LIST ||
        c->btype == BLOCKED_ZSET ||
        c->btype == BLOCKED_STREAM) {
        // 等待数据类型的 block
        unblockClientWaitingData(c);
    } else if (c->btype == BLOCKED_WAIT) {
        // 等待副本同步的block
        unblockClientWaitingReplicas(c);
    } else if (c->btype == BLOCKED_MODULE) {
        // 等待module状态相关的block
        if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
        unblockClientFromModule(c);
    } else if (c->btype == BLOCKED_PAUSE) {
        // 暂停类型的block
        listDelNode(server.paused_clients,c->paused_list_node);
        c->paused_list_node = NULL;
    } else {
        serverPanic("Unknown btype in unblockClient().");
    }
    /* Reset the client for a new query since, for blocking commands
     * we do not do it immediately after the command returns (when the
     * client got blocked) in order to be still able to access the argument
     * vector from module callbacks and updateStatsOnUnblock. */
    if (c->btype != BLOCKED_PAUSE) {
        // 命令类型block(即非pause类型)
        // 此处需要释放或者reset相关变量
        freeClientOriginalArgv(c);
        resetClient(c);
    }
    /* Clear the flags, and put the client in the unblocked list so that
     * we'll process new commands in its query buffer ASAP. */
    server.blocked_clients--;
    server.blocked_clients_by_type[c->btype]--;
    // 去掉blocked 标记
    c->flags &= ~CLIENT_BLOCKED;
    c->btype = BLOCKED_NONE;
    // 从block超时相关的dict中删除客户端
    removeClientFromTimeoutTable(c);
    // 对client添加unblock 标记，并且添加到erver.unblocked_clients
    // 之后在beforesleep 进行处理(unblock时或者之后的命令处理)
    queueClientForReprocessing(c);
}
/* Unblock a client that's waiting in a blocking operation such as BLPOP.
 * You should never call this function directly, but unblockClient() instead. */
// 通过调用unblockClient() 来unblock
// 遍历c->bpop.keys, 从c->db->blocking_keys key对应列表中删除对应client
// 删除c->bpop.keys列表
void unblockClientWaitingData(client *c) {
    dictEntry *de;
    dictIterator *di;
    list *l;
    serverAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
    di = dictGetIterator(c->bpop.keys);
    /* The client may wait for multiple keys, so unblock it for every key. */
    // 一个客户端可能等待多个key 就绪, 因此遍历所有等待的key进行unblock
    // 遍历 c->bpop.keys
    while((de = dictNext(di)) != NULL) {
        robj *key = dictGetKey(de);
        bkinfo *bki = dictGetVal(de);
        /* Remove this client from the list of clients waiting for this key. */
        // 从c->db->blocking_keys 找到key对应列表
        l = dictFetchValue(c->db->blocking_keys,key);
        serverAssertWithInfo(c,key,l != NULL);
        // 从列表中删除本client的 node
        listDelNode(l,bki->listnode);
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
            // 如果列表为空,从c->db->blocking_keys删除对应key val
            dictDelete(c->db->blocking_keys,key);
    }
    dictReleaseIterator(di);
    /* Cleanup the client structure */
    // 清理client结构中有关block相关的变量
    dictEmpty(c->bpop.keys,NULL);
    if (c->bpop.target) {
        decrRefCount(c->bpop.target);
        c->bpop.target = NULL;
    }
    if (c->bpop.xread_group) {
        decrRefCount(c->bpop.xread_group);
        decrRefCount(c->bpop.xread_consumer);
        c->bpop.xread_group = NULL;
        c->bpop.xread_consumer = NULL;
    }
}
// 对client添加unblock 标记，并且添加到erver.unblocked_clients
// 在beforesleep 进行1处理(unblock时或者之后的命令处理)
void queueClientForReprocessing(client *c) {
    /* The client may already be into the unblocked list because of a previous
     * blocking operation, don't add back it into the list multiple times. */
    if (!(c->flags & CLIENT_UNBLOCKED)) {
        c->flags |= CLIENT_UNBLOCKED;
        listAddNodeTail(server.unblocked_clients,c);
    }
}
// path:src/server.c
// 处理完所有客户端事件(命令)
// 进入epoll_wait之前
void beforeSleep(struct aeEventLoop *eventLoop) {
    // ...
    /* Try to process blocked clients every once in while. Example: A module
     * calls RM_SignalKeyAsReady from within a timer callback (So we don't
     * visit processCommand() at all). */
    // 处理被某些key block的客户端列表
    handleClientsBlockedOnKeys();
    // ...
}
// path:src/timeout.c
/* Remove the client from the table when it is unblocked for reasons
 * different than timing out. */
// 将client 从TimeoutTable中删除
void removeClientFromTimeoutTable(client *c) {
    if (!(c->flags & CLIENT_IN_TO_TABLE)) return;
    c->flags &= ~CLIENT_IN_TO_TABLE;
    uint64_t timeout = c->bpop.timeout;
    unsigned char buf[CLIENT_ST_KEYLEN];
    encodeTimeoutKey(buf,timeout,c);
    raxRemove(server.clients_timeout_table,buf,sizeof(buf),NULL);
}