系列

redis數(shù)據(jù)淘汰原理
redis過(guò)期數(shù)據(jù)刪除策略
redis server事件模型
redis cluster mget 引發(fā)的討論
redis 3.x windows 集群搭建
redis 命令執(zhí)行過(guò)程
redis string底層數(shù)據(jù)結(jié)構(gòu)
redis list底層數(shù)據(jù)結(jié)構(gòu)
redis hash底層數(shù)據(jù)結(jié)構(gòu)
redis set底層數(shù)據(jù)結(jié)構(gòu)
redis zset底層數(shù)據(jù)結(jié)構(gòu)
redis 客戶端管理
redis 主從同步-slave端
redis 主從同步-master端
redis 主從超時(shí)檢測(cè)
redis aof持久化
redis rdb持久化
redis 數(shù)據(jù)恢復(fù)過(guò)程
redis TTL實(shí)現(xiàn)原理
redis cluster集群建立
redis cluster集群選主

redis 主從命令

?redis的主從同步始于命令SLAVEOF host port，通過(guò)這個(gè)命令能夠建立主從關(guān)系。

• SLAVEOF 命令用于在 Redis 運(yùn)行時(shí)動(dòng)態(tài)地修改復(fù)制(replication)功能的行為。

• 通過(guò)執(zhí)行 SLAVEOF host port 命令，可以將當(dāng)前服務(wù)器轉(zhuǎn)變?yōu)橹付ǚ?wù)器的從屬服務(wù)器(slave server)。

• 如果當(dāng)前服務(wù)器已經(jīng)是某個(gè)主服務(wù)器(master server)的從屬服務(wù)器，那么執(zhí)行 SLAVEOF host port 將使當(dāng)前服務(wù)器停止對(duì)舊主服務(wù)器的同步，丟棄舊數(shù)據(jù)集，轉(zhuǎn)而開(kāi)始對(duì)新主服務(wù)器進(jìn)行同步。

• 另外，對(duì)一個(gè)從屬服務(wù)器執(zhí)行命令 SLAVEOF NO ONE 將使得這個(gè)從屬服務(wù)器關(guān)閉復(fù)制功能，并從從屬服務(wù)器轉(zhuǎn)變回主服務(wù)器，原來(lái)同步所得的數(shù)據(jù)集不會(huì)被丟棄。

• 利用『 SLAVEOF NO ONE 不會(huì)丟棄同步所得數(shù)據(jù)集』這個(gè)特性，可以在主服務(wù)器失敗的時(shí)候，將從屬服務(wù)器用作新的主服務(wù)器，從而實(shí)現(xiàn)無(wú)間斷運(yùn)行。

redis 主從源碼分析

?主從連接實(shí)現(xiàn)邏輯分為兩步走

• 將redis的節(jié)點(diǎn)設(shè)置為REDIS_REPL_CONNECT狀態(tài)
• 由定時(shí)任務(wù)根據(jù)狀態(tài)實(shí)現(xiàn)主從連接并實(shí)現(xiàn)數(shù)據(jù)同步

redis 設(shè)置主從同步狀態(tài)

?slaveofCommand主要處理兩種情況：去主從關(guān)系和添加主從關(guān)系。

• slaveof no one 表示去除主從同步關(guān)系
• slaveof ip port 表示添加主從同步關(guān)系

?去除主從同步關(guān)系做的事情包括：

• 停止server.master的socket的連接，用于全量同步使用的socket
• 停止server.cached_master的socket連接，用于部分同步使用的socket

?添加主從同步關(guān)系做的事情包括：

• 判斷ip:port是否跟已建立的連接相同那么就忽略
• 如果ip:port是新就進(jìn)入重建和主節(jié)點(diǎn)的連接replicationSetMaster。

void slaveofCommand(redisClient *c) {
   
    // 不允許在集群模式中使用
    if (server.cluster_enabled) {
        addReplyError(c,"SLAVEOF not allowed in cluster mode.");
        return;
    }

    // SLAVEOF NO ONE 讓從服務(wù)器轉(zhuǎn)為主服務(wù)器
    if (!strcasecmp(c->argv[1]->ptr,"no") &&
        !strcasecmp(c->argv[2]->ptr,"one")) {
        if (server.masterhost) {
            // 讓服務(wù)器取消復(fù)制，成為主服務(wù)器
            replicationUnsetMaster();
            redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
        }
    } else {
        long port;

        // 獲取端口參數(shù)
        if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != REDIS_OK))
            return;

        // 檢查輸入的 host 和 port 是否服務(wù)器目前的主服務(wù)器
        // 如果是的話，向客戶端返回 +OK ，不做其他動(dòng)作
        if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
            && server.masterport == port) {
            redisLog(REDIS_NOTICE,"SLAVE OF would result into synchronization with the master we are already connected with.");
            addReplySds(c,sdsnew("+OK Already connected to specified master\r\n"));
            return;
        }

        // 沒(méi)有前任主服務(wù)器，或者客戶端指定了新的主服務(wù)器
        // 開(kāi)始執(zhí)行復(fù)制操作
        replicationSetMaster(c->argv[1]->ptr, port);
        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
            server.masterhost, server.masterport);
    }
    addReply(c,shared.ok);
}

?釋放和主節(jié)點(diǎn)建立的socket連接，這部分是指全量同步的client連接。

// 取消復(fù)制，將服務(wù)器設(shè)置為主服務(wù)器
void replicationUnsetMaster(void) {

    if (server.masterhost == NULL) return; /* Nothing to do. */

    sdsfree(server.masterhost);
    server.masterhost = NULL;

    if (server.master) {
        if (listLength(server.slaves) == 0) {
            server.master_repl_offset = server.master->reploff;
            freeReplicationBacklog();
        }
        freeClient(server.master);
    }

    replicationDiscardCachedMaster();

    cancelReplicationHandshake();

    server.repl_state = REDIS_REPL_NONE;
}

?釋放backlog相關(guān)的緩存，邏輯簡(jiǎn)單明了不多解釋。

// 釋放 backlog
void freeReplicationBacklog(void) {
    redisAssert(listLength(server.slaves) == 0);
    zfree(server.repl_backlog);
    server.repl_backlog = NULL;
}

?釋放和主節(jié)點(diǎn)建立的socket連接，這部分是指部分同步的client連接。

/* 
 * 清空 master 緩存，在條件已經(jīng)不可能執(zhí)行 partial resync 時(shí)執(zhí)行
 */
void replicationDiscardCachedMaster(void) {

    if (server.cached_master == NULL) return;

    redisLog(REDIS_NOTICE,"Discarding previously cached master state.");
    server.cached_master->flags &= ~REDIS_MASTER;
    freeClient(server.cached_master);
    server.cached_master = NULL;
}

?如果復(fù)制在握手階段那么就取消主從復(fù)制。

/* 
 * 如果有正在進(jìn)行的非阻塞復(fù)制在進(jìn)行，那么取消它。
 * 如果復(fù)制在握手階段被取消，那么返回 1 ，
 * 并且 server.repl_state 被設(shè)置為 REDIS_REPL_CONNECT 。
 * 否則返回 0 ，并且不執(zhí)行任何操作。
 */
int cancelReplicationHandshake(void) {
    if (server.repl_state == REDIS_REPL_TRANSFER) {
        replicationAbortSyncTransfer();
    } else if (server.repl_state == REDIS_REPL_CONNECTING ||
             server.repl_state == REDIS_REPL_RECEIVE_PONG)
    {
        undoConnectWithMaster();
    } else {
        return 0;
    }
    return 1;
}

?釋放舊的master信息，并設(shè)置新的master信息，在這里主要是把server.repl_state = REDIS_REPL_CONNECT。這里設(shè)置了狀態(tài)，通過(guò)定時(shí)任務(wù)實(shí)現(xiàn)slave->master的主從連接

// 將服務(wù)器設(shè)為指定地址的從服務(wù)器
void replicationSetMaster(char *ip, int port) {

    // 清除原有的主服務(wù)器地址（如果有的話）
    sdsfree(server.masterhost);

    // IP
    server.masterhost = sdsnew(ip);

    // 端口
    server.masterport = port;

    // 清除原來(lái)可能有的主服務(wù)器信息。。。

    // 如果之前有其他地址，那么釋放它
    if (server.master) freeClient(server.master);
    // 斷開(kāi)所有從服務(wù)器的連接，強(qiáng)制所有從服務(wù)器執(zhí)行重同步
    disconnectSlaves(); /* Force our slaves to resync with us as well. */
    // 清空可能有的 master 緩存，因?yàn)橐呀?jīng)不會(huì)執(zhí)行 PSYNC 了
    replicationDiscardCachedMaster(); /* Don't try a PSYNC. */
    // 釋放 backlog ，同理， PSYNC 目前已經(jīng)不會(huì)執(zhí)行了
    freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
    // 取消之前的復(fù)制進(jìn)程（如果有的話）
    cancelReplicationHandshake();

    // 進(jìn)入連接狀態(tài)（重點(diǎn)）
    server.repl_state = REDIS_REPL_CONNECT;
    server.master_repl_offset = 0;
    server.repl_down_since = 0;
}

redis 執(zhí)行主從同步

?在執(zhí)行同步過(guò)程中我們首先排除其他異常情況，重新建立主從連接，執(zhí)行數(shù)據(jù)拷貝，重新加載rdb數(shù)據(jù)。整個(gè)步驟如下：

• 處理主從連接異常情況
• 處理rdb文件傳送超時(shí)情況
• 處理已連接的client但是超時(shí)情況
• 處理主從重新建立連接，主要在connectWithMaster函數(shù)
• 向該節(jié)點(diǎn)的slave節(jié)點(diǎn)發(fā)送ping的心跳命令,slave也可以掛slave節(jié)點(diǎn)
• 需要強(qiáng)調(diào)的是這個(gè)ping命令只要發(fā)送一個(gè)"\n"命令

/* --------------------------- REPLICATION CRON  ---------------------------- */
// 復(fù)制 cron 函數(shù)，每秒調(diào)用一次
void replicationCron(void) {

    // 嘗試連接到主服務(wù)器，但超時(shí)
    if (server.masterhost &&
        (server.repl_state == REDIS_REPL_CONNECTING ||
         server.repl_state == REDIS_REPL_RECEIVE_PONG) &&
        (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
    {
        redisLog(REDIS_WARNING,"Timeout connecting to the MASTER...");
        // 取消連接
        undoConnectWithMaster();
    }

    // RDB 文件的傳送已超時(shí)？
    if (server.masterhost && server.repl_state == REDIS_REPL_TRANSFER &&
        (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
    {
        redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER... If the problem persists try to set the 'repl-timeout' parameter in redis.conf to a larger value.");
        // 停止傳送，并刪除臨時(shí)文件
        replicationAbortSyncTransfer();
    }

    // 從服務(wù)器曾經(jīng)連接上主服務(wù)器，但現(xiàn)在超時(shí)
    if (server.masterhost && server.repl_state == REDIS_REPL_CONNECTED &&
        (time(NULL)-server.master->lastinteraction) > server.repl_timeout)
    {
        redisLog(REDIS_WARNING,"MASTER timeout: no data nor PING received...");
        // 釋放主服務(wù)器
        freeClient(server.master);
    }

    // 嘗試連接主服務(wù)器
    if (server.repl_state == REDIS_REPL_CONNECT) {
        redisLog(REDIS_NOTICE,"Connecting to MASTER %s:%d",
            server.masterhost, server.masterport);
        if (connectWithMaster() == REDIS_OK) {
            redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started");
        }
    }

    // 定期向主服務(wù)器發(fā)送 ACK 命令
    // 不過(guò)如果主服務(wù)器帶有 REDIS_PRE_PSYNC 的話就不發(fā)送
    // 因?yàn)閹в性摌?biāo)識(shí)的版本為 < 2.8 的版本，這些版本不支持 ACK 命令
    if (server.masterhost && server.master &&
        !(server.master->flags & REDIS_PRE_PSYNC))
        replicationSendAck();
    
    /* 
     * 如果服務(wù)器有從服務(wù)器，定時(shí)向它們發(fā)送 PING 。
     *
     * 這樣從服務(wù)器就可以實(shí)現(xiàn)顯式的 master 超時(shí)判斷機(jī)制，
     * 即使 TCP 連接未斷開(kāi)也是如此。
     */
    if (!(server.cronloops % (server.repl_ping_slave_period * server.hz))) {
        listIter li;
        listNode *ln;
        robj *ping_argv[1];

        // 向所有已連接 slave （狀態(tài)為 ONLINE）發(fā)送 PING
        ping_argv[0] = createStringObject("PING",4);
        replicationFeedSlaves(server.slaves, server.slaveseldb, ping_argv, 1);
        decrRefCount(ping_argv[0]);

        /*
         * 向那些正在等待 RDB 文件的從服務(wù)器（狀態(tài)為 BGSAVE_START 或 BGSAVE_END）
         * 發(fā)送 "\n"
         *
         *
         * 這個(gè) "\n" 會(huì)被從服務(wù)器忽略，
         * 它的作用就是用來(lái)防止主服務(wù)器因?yàn)殚L(zhǎng)期不發(fā)送信息而被從服務(wù)器誤判為超時(shí)
         */
        listRewind(server.slaves,&li);
        while((ln = listNext(&li))) {
            redisClient *slave = ln->value;

            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START ||
                slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
                if (write(slave->fd, "\n", 1) == -1) {
                    /* Don't worry, it's just a ping. */
                }
            }
        }
    }

    /* Disconnect timedout slaves. */
    // 斷開(kāi)超時(shí)從服務(wù)器
    if (listLength(server.slaves)) {
        listIter li;
        listNode *ln;

        // 遍歷所有從服務(wù)器
        listRewind(server.slaves,&li);
        while((ln = listNext(&li))) {
            redisClient *slave = ln->value;

            // 略過(guò)未 ONLINE 的從服務(wù)器
            if (slave->replstate != REDIS_REPL_ONLINE) continue;

            // 不檢查舊版的從服務(wù)器
            if (slave->flags & REDIS_PRE_PSYNC) continue;

            // 釋放超時(shí)從服務(wù)器
            if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout)
            {
                char ip[REDIS_IP_STR_LEN];
                int port;

                if (anetPeerToString(slave->fd,ip,sizeof(ip),&port) != -1) {
                    redisLog(REDIS_WARNING,
                        "Disconnecting timedout slave: %s:%d",
                        ip, slave->slave_listening_port);
                }
                
                // 釋放
                freeClient(slave);
            }
        }
    }

    // 在沒(méi)有任何從服務(wù)器的 N 秒之后，釋放 backlog
    if (listLength(server.slaves) == 0 && server.repl_backlog_time_limit &&
        server.repl_backlog)
    {
        time_t idle = server.unixtime - server.repl_no_slaves_since;

        if (idle > server.repl_backlog_time_limit) {
            // 釋放
            freeReplicationBacklog();
            redisLog(REDIS_NOTICE,
                "Replication backlog freed after %d seconds "
                "without connected slaves.",
                (int) server.repl_backlog_time_limit);
        }
    }

    // 在沒(méi)有任何從服務(wù)器，AOF 關(guān)閉的情況下，清空 script 緩存
    // 因?yàn)橐呀?jīng)沒(méi)有傳播 EVALSHA 的必要了
    if (listLength(server.slaves) == 0 &&
        server.aof_state == REDIS_AOF_OFF &&
        listLength(server.repl_scriptcache_fifo) != 0)
    {
        replicationScriptCacheFlush();
    }

    // 更新符合給定延遲值的從服務(wù)器的數(shù)量
    refreshGoodSlavesCount();
}

?創(chuàng)建連接主服務(wù)的socket并綁定syncWithMaster函數(shù)到eventLoop當(dāng)中，這個(gè)時(shí)候會(huì)將整個(gè)連接狀態(tài)設(shè)置為REDIS_REPL_CONNECTING。
?slave的socket連接到master成功后會(huì)首先觸發(fā)的寫(xiě)事件，執(zhí)行寫(xiě)事件對(duì)應(yīng)的函數(shù)syncWithMaster。

// 以非阻塞方式連接主服務(wù)器
int connectWithMaster(void) {
    int fd;

    // 連接主服務(wù)器
    fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport);
    if (fd == -1) {
        redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
            strerror(errno));
        return REDIS_ERR;
    }

    // 監(jiān)聽(tīng)主服務(wù)器 fd 的讀和寫(xiě)事件，并綁定文件事件處理器
    if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) ==
            AE_ERR)
    {
        close(fd);
        redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
        return REDIS_ERR;
    }

    // 初始化統(tǒng)計(jì)變量
    server.repl_transfer_lastio = server.unixtime;
    server.repl_transfer_s = fd;

    // 將狀態(tài)改為已連接
    server.repl_state = REDIS_REPL_CONNECTING;

    return REDIS_OK;
}

?整個(gè)同步過(guò)程其實(shí)類似TCP通信的幾次握手事件，這里先用文字表述一下，后面再以交互圖的形式呈現(xiàn)出來(lái)，整個(gè)過(guò)程如下：

• slave發(fā)送ping命令到master
• master發(fā)送pong命令到slave
• slave發(fā)送auth命令到master
• slave發(fā)送listen port到master便于顯示
• slave發(fā)送PSYNC到master
• master發(fā)送+FULLRESYNC（支持全量同步）或者+CONTINUE（支持部分同步）
• slave發(fā)送sync命令道m(xù)aster
• slave創(chuàng)建本地的temp的rdb文件
• slave綁定fd的讀事件處理函數(shù)為readSyncBulkPayload,所有讀事件由該函數(shù)處理

// 從服務(wù)器用于同步主服務(wù)器的回調(diào)函數(shù)
void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
    char tmpfile[256], *err;
    int dfd, maxtries = 5;
    int sockerr = 0, psync_result;
    socklen_t errlen = sizeof(sockerr);
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(privdata);
    REDIS_NOTUSED(mask);

    // 如果處于 SLAVEOF NO ONE 模式，那么關(guān)閉 fd
    if (server.repl_state == REDIS_REPL_NONE) {
        close(fd);
        return;
    }

    // 檢查套接字錯(cuò)誤
    if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) == -1)
        sockerr = errno;
    if (sockerr) {
        aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
        redisLog(REDIS_WARNING,"Error condition on socket for SYNC: %s",
            strerror(sockerr));
        goto error;
    }

    // 如果狀態(tài)為 CONNECTING ，那么在進(jìn)行初次同步之前，
    // 向主服務(wù)器發(fā)送一個(gè)非阻塞的 PONG 
    // 因?yàn)榻酉聛?lái)的 RDB 文件發(fā)送非常耗時(shí)，所以我們想確認(rèn)主服務(wù)器真的能訪問(wèn)
    if (server.repl_state == REDIS_REPL_CONNECTING) {
        redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
        /* Delete the writable event so that the readable event remains
         * registered and we can wait for the PONG reply. */
        // 手動(dòng)發(fā)送同步 PING ，暫時(shí)取消監(jiān)聽(tīng)寫(xiě)事件
        aeDeleteFileEvent(server.el,fd,AE_WRITABLE);
        // 更新?tīng)顟B(tài)
        server.repl_state = REDIS_REPL_RECEIVE_PONG;
        /* Send the PING, don't check for errors at all, we have the timeout
         * that will take care about this. */
        // 同步發(fā)送 PING
        syncWrite(fd,"PING\r\n",6,100);

        // 返回，等待 PONG 到達(dá)
        return;
    }

    // 接收 PONG 命令
    if (server.repl_state == REDIS_REPL_RECEIVE_PONG) {
        char buf[1024];

        // 手動(dòng)同步接收 PONG ，暫時(shí)取消監(jiān)聽(tīng)讀事件
        aeDeleteFileEvent(server.el,fd,AE_READABLE);

        // 嘗試在指定時(shí)間限制內(nèi)讀取 PONG
        buf[0] = '\0';
        // 同步接收 PONG
        if (syncReadLine(fd,buf,sizeof(buf),
            server.repl_syncio_timeout*1000) == -1)
        {
            redisLog(REDIS_WARNING,
                "I/O error reading PING reply from master: %s",
                strerror(errno));
            goto error;
        }

        // 接收到的數(shù)據(jù)只有兩種可能：
        // 第一種是 +PONG ，第二種是因?yàn)槲打?yàn)證而出現(xiàn)的 -NOAUTH 錯(cuò)誤
        if (buf[0] != '+' &&
            strncmp(buf,"-NOAUTH",7) != 0 &&
            strncmp(buf,"-ERR operation not permitted",28) != 0)
        {
            // 接收到未驗(yàn)證錯(cuò)誤
            redisLog(REDIS_WARNING,"Error reply to PING from master: '%s'",buf);
            goto error;
        } else {
            // 接收到 PONG
            redisLog(REDIS_NOTICE,
                "Master replied to PING, replication can continue...");
        }
    }

    // 進(jìn)行身份驗(yàn)證
    if(server.masterauth) {
        err = sendSynchronousCommand(fd,"AUTH",server.masterauth,NULL);
        if (err[0] == '-') {
            redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",err);
            sdsfree(err);
            goto error;
        }
        sdsfree(err);
    }

    // 將從服務(wù)器的端口發(fā)送給主服務(wù)器，
    // 使得主服務(wù)器的 INFO 命令可以顯示從服務(wù)器正在監(jiān)聽(tīng)的端口
    {
        sds port = sdsfromlonglong(server.port);
        err = sendSynchronousCommand(fd,"REPLCONF","listening-port",port,
                                         NULL);
        sdsfree(port);
        /* Ignore the error if any, not all the Redis versions support
         * REPLCONF listening-port. */
        if (err[0] == '-') {
            redisLog(REDIS_NOTICE,"(Non critical) Master does not understand REPLCONF listening-port: %s", err);
        }
        sdsfree(err);
    }

    // 根據(jù)返回的結(jié)果決定是執(zhí)行部分 resync ，還是 full-resync
    psync_result = slaveTryPartialResynchronization(fd);

    // 可以執(zhí)行部分 resync
    if (psync_result == PSYNC_CONTINUE) {
        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Master accepted a Partial Resynchronization.");
        // 返回
        return;
    }

    // 主服務(wù)器不支持 PSYNC ，發(fā)送 SYNC
    if (psync_result == PSYNC_NOT_SUPPORTED) {
        redisLog(REDIS_NOTICE,"Retrying with SYNC...");
        // 向主服務(wù)器發(fā)送 SYNC 命令
        if (syncWrite(fd,"SYNC\r\n",6,server.repl_syncio_timeout*1000) == -1) {
            redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
                strerror(errno));
            goto error;
        }
    }

    // 如果執(zhí)行到這里，
    // 那么 psync_result == PSYNC_FULLRESYNC 或 PSYNC_NOT_SUPPORTED

    // 打開(kāi)一個(gè)臨時(shí)文件，用于寫(xiě)入和保存接下來(lái)從主服務(wù)器傳來(lái)的 RDB 文件數(shù)據(jù)
    while(maxtries--) {
        snprintf(tmpfile,256,
            "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
        dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
        if (dfd != -1) break;
        sleep(1);
    }
    if (dfd == -1) {
        redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
        goto error;
    }

    // 設(shè)置一個(gè)讀事件處理器，來(lái)讀取主服務(wù)器的 RDB 文件
    if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
            == AE_ERR)
    {
        redisLog(REDIS_WARNING,
            "Can't create readable event for SYNC: %s (fd=%d)",
            strerror(errno),fd);
        goto error;
    }

    // 設(shè)置狀態(tài)
    server.repl_state = REDIS_REPL_TRANSFER;

    // 更新統(tǒng)計(jì)信息
    server.repl_transfer_size = -1;
    server.repl_transfer_read = 0;
    server.repl_transfer_last_fsync_off = 0;
    server.repl_transfer_fd = dfd;
    server.repl_transfer_lastio = server.unixtime;
    server.repl_transfer_tmpfile = zstrdup(tmpfile);

    return;

error:
    close(fd);
    server.repl_transfer_s = -1;
    server.repl_state = REDIS_REPL_CONNECT;
    return;
}

?主從同步的數(shù)據(jù)同步的核心函數(shù)在這里，這里需要說(shuō)明一下的是我們重新綁定了讀事件對(duì)應(yīng)的處理函數(shù)readSyncBulkPayload，整個(gè)核心邏輯如下：

• slave接收master發(fā)送的ping命令就更新server.repl_transfer_lastio字段
• slave讀取master發(fā)送rdb文件數(shù)據(jù)并寫(xiě)入到本地的臨時(shí)rdb文件當(dāng)中
• slave加載rdb文件
• 若開(kāi)啟AOF數(shù)據(jù)持久化,那么將加載的數(shù)據(jù)寫(xiě)入aof文件

// 異步 RDB 文件讀取函數(shù)
#define REPL_MAX_WRITTEN_BEFORE_FSYNC (1024*1024*8) /* 8 MB */
void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[4096];
    ssize_t nread, readlen;
    off_t left;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(privdata);
    REDIS_NOTUSED(mask);

    // 讀取 RDB 文件的大小
    if (server.repl_transfer_size == -1) {

        // 調(diào)用讀函數(shù)
        if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout*1000) == -1) {
            redisLog(REDIS_WARNING,
                "I/O error reading bulk count from MASTER: %s",
                strerror(errno));
            goto error;
        }

        // 出錯(cuò)？
        if (buf[0] == '-') {
            redisLog(REDIS_WARNING,
                "MASTER aborted replication with an error: %s",
                buf+1);
            goto error;
        } else if (buf[0] == '\0') {
            /* At this stage just a newline works as a PING in order to take
             * the connection live. So we refresh our last interaction
             * timestamp. */
            // 只接到了一個(gè)作用和 PING 一樣的 '\0'
            // 更新最后互動(dòng)時(shí)間
            server.repl_transfer_lastio = server.unixtime;
            return;
        } else if (buf[0] != '$') {
            // 讀入的內(nèi)容出錯(cuò)，和協(xié)議格式不符
            redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$' (we received '%s'), are you sure the host and port are right?", buf);
            goto error;
        }

        // 分析 RDB 文件大小
        server.repl_transfer_size = strtol(buf+1,NULL,10);

        redisLog(REDIS_NOTICE,
            "MASTER <-> SLAVE sync: receiving %lld bytes from master",
            (long long) server.repl_transfer_size);
        return;
    }

    // 讀數(shù)據(jù)

    // 還有多少字節(jié)要讀？
    left = server.repl_transfer_size - server.repl_transfer_read;
    readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
    // 讀取
    nread = read(fd,buf,readlen);
    if (nread <= 0) {
        redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
            (nread == -1) ? strerror(errno) : "connection lost");
        replicationAbortSyncTransfer();
        return;
    }
    // 更新最后 RDB 產(chǎn)生的 IO 時(shí)間
    server.repl_transfer_lastio = server.unixtime;
    if (write(server.repl_transfer_fd,buf,nread) != nread) {
        redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", strerror(errno));
        goto error;
    }
    // 加上剛讀取好的字節(jié)數(shù)
    server.repl_transfer_read += nread;

    // 定期將讀入的文件 fsync 到磁盤，以免 buffer 太多，一下子寫(xiě)入時(shí)撐爆 IO
    if (server.repl_transfer_read >=
        server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
    {
        off_t sync_size = server.repl_transfer_read -
                          server.repl_transfer_last_fsync_off;
        rdb_fsync_range(server.repl_transfer_fd,
            server.repl_transfer_last_fsync_off, sync_size);
        server.repl_transfer_last_fsync_off += sync_size;
    }

    // 檢查 RDB 是否已經(jīng)傳送完畢
    if (server.repl_transfer_read == server.repl_transfer_size) {

        // 完畢，將臨時(shí)文件改名為 dump.rdb
        if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
            redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
            replicationAbortSyncTransfer();
            return;
        }

        // 先清空舊數(shù)據(jù)庫(kù)
        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Flushing old data");
        signalFlushedDb(-1);
        emptyDb(replicationEmptyDbCallback);

        // 先刪除主服務(wù)器的讀事件監(jiān)聽(tīng)，因?yàn)?rdbLoad() 函數(shù)也會(huì)監(jiān)聽(tīng)讀事件
        aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);

        // 載入 RDB
        if (rdbLoad(server.rdb_filename) != REDIS_OK) {
            redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
            replicationAbortSyncTransfer();
            return;
        }

        /* Final setup of the connected slave <- master link */
        // 關(guān)閉臨時(shí)文件
        zfree(server.repl_transfer_tmpfile);
        close(server.repl_transfer_fd);

        // 將主服務(wù)器設(shè)置成一個(gè) redis client
        // 注意 createClient 會(huì)為主服務(wù)器綁定事件，為接下來(lái)接收命令做好準(zhǔn)備
        server.master = createClient(server.repl_transfer_s);
        // 標(biāo)記這個(gè)客戶端為主服務(wù)器
        server.master->flags |= REDIS_MASTER;
        // 標(biāo)記它為已驗(yàn)證身份
        server.master->authenticated = 1;
        // 更新復(fù)制狀態(tài)
        server.repl_state = REDIS_REPL_CONNECTED;
        // 設(shè)置主服務(wù)器的復(fù)制偏移量
        server.master->reploff = server.repl_master_initial_offset;
        // 保存主服務(wù)器的 RUN ID
        memcpy(server.master->replrunid, server.repl_master_runid,
            sizeof(server.repl_master_runid));

        // 如果 offset 被設(shè)置為 -1 ，那么表示主服務(wù)器的版本低于 2.8 
        // 無(wú)法使用 PSYNC ，所以需要設(shè)置相應(yīng)的標(biāo)識(shí)值
        if (server.master->reploff == -1)
            server.master->flags |= REDIS_PRE_PSYNC;
        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");

        // 如果有開(kāi)啟 AOF 持久化，那么重啟 AOF 功能，并強(qiáng)制生成新數(shù)據(jù)庫(kù)的 AOF 文件
        if (server.aof_state != REDIS_AOF_OFF) {
            int retry = 10;

            // 關(guān)閉
            stopAppendOnly();
            // 再重啟
            while (retry-- && startAppendOnly() == REDIS_ERR) {
                redisLog(REDIS_WARNING,"Failed enabling the AOF after successful master synchronization! Trying it again in one second.");
                sleep(1);
            }
            if (!retry) {
                redisLog(REDIS_WARNING,"FATAL: this slave instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now.");
                exit(1);
            }
        }
    }

    return;

error:
    replicationAbortSyncTransfer();
    return;
}

redis 執(zhí)行主從同步報(bào)文交互

主從報(bào)文交互