Redis 監(jiān)視器

Redis 的監(jiān)視機制允許某一個客戶端監(jiān)視 Redis 服務(wù)器的行為，這種服務(wù)對于測試來說比較有幫助。

監(jiān)視機制通過 monitor 這個命令來實現(xiàn)。來看看它的實現(xiàn)：Redis 在這里只是簡單講這個客戶端加到一個 redis.monitors 鏈表中，接著就回復(fù) ok 給客戶端。

void monitorCommand(redisClient *c) {
    /* ignore MONITOR if already slave or in monitor mode */
    if (c->flags & REDIS_SLAVE) return;
        c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
        listAddNodeTail(server.monitors,c);
        addReply(c,shared.ok);
}

這里可以想象，當這個 Redis 服務(wù)器處理其他命令的時候，會向這個鏈表中的所有客戶端發(fā)送通知。我們找到執(zhí)行命令的核心函數(shù) call()，可以發(fā)現(xiàn)確實是這么做的：

// call() 函數(shù)是執(zhí)行命令的核心函數(shù)，真正執(zhí)行命令的地方
/* Call() is the core of Redis execution of a command */
void call(redisClient *c, int flags) {
    long long dirty, start = ustime(), duration;
    int client_old_flags = c->flags;
    /* Sent the command to clients in MONITOR mode, only if the commands are
    * not generated from reading an AOF. */
    if (listLength(server.monitors) &&
        !server.loading &&
        !(c->cmd->flags & REDIS_CMD_SKIP_MONITOR))
        {
        replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
    }
    ......
}

replicationFeedMonitors() 的實現(xiàn)實際上就是將命令打包好，發(fā)送給每個監(jiān)視器：

// 向監(jiān)視器發(fā)送數(shù)據(jù)
void replicationFeedMonitors(redisClient *c, list *monitors, int dictid,
    robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int j;
    sds cmdrepr = sdsnew("+");
    robj *cmdobj;
    char peerid[REDIS_PEER_ID_LEN];
    struct timeval tv;
    // 時間
    gettimeofday(&tv,NULL);
    cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
    // 各種不同的客戶端
    if (c->flags & REDIS_LUA_CLIENT) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
    } else if (c->flags & REDIS_UNIX_SOCKET) {
        cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket);
    } else {
        getClientPeerId(c,peerid,sizeof(peerid));
        cmdrepr = sdscatprintf(cmdrepr,"[%d %s] ",dictid,peerid);
    }
    for (j = 0; j < argc; j++) {
        if (argv[j]->encoding == REDIS_ENCODING_INT) {
            cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
        } else {
            cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
            sdslen(argv[j]->ptr));
        }
        if (j != argc-1)
            cmdrepr = sdscatlen(cmdrepr," ",1);
        }
        cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
        cmdobj = createObject(REDIS_STRING,cmdrepr);
        // 發(fā)送
        listRewind(monitors,&li);
    while((ln = listNext(&li))) {
        redisClient *monitor = ln->value;
        addReply(monitor,cmdobj);
    }
    decrRefCount(cmdobj);
}