首页 / REDIS / redis源代码分析19–主从复制

redis源代码分析19–主从复制

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了redis源代码分析19–主从复制，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含10503字，纯文字阅读大概需要16分钟。

内容图文

先说下Redis主从复制的特点。

官方文档ReplicationHowto中提到以下特点：
1. 一个master支持多个slave
2. slave可以接受其他slave的连接，作为其他slave的master，从而形成一个master-slave的多级结构
3. 复制在master端是非阻塞的，也就是master在向client复制时可处理其他client的命令，而slave在第一次同步时是阻塞的
4. 复制被利用来提供可扩展性，比如可以将slave端用作数据冗余，也可以将耗时的命令（比如sort）发往某些slave从而避免master的阻塞，另外也可以用slave做持久化，这只需要将master的配置文件中的save指令注释掉。

client可以在一开始时作为slave连接master，也可以在运行后发布sync命令，从而跟master建立主从关系。

接下来我们分别从slave和master的视角概述下redis的主从复制的运行机制。

如果redis作为slave运行，则全局变量server.replstate的状态有REDIS_REPL_NONE（不处于复制状态）、 REDIS_REPL_CONNECT（需要跟master建立连接）、REDIS_REPL_CONNECTED（已跟master建立连接）三种。在读入slaveof配置或者发布slaveof命令后，server.replstate取值为REDIS_REPL_CONNECT，然后在syncWithMaster跟master执行第一次同步后，取值变为REDIS_REPL_CONNECTED。

如果redis作为master运行，则对应某个客户端连接的变量slave.replstate的状态有REDIS_REPL_WAIT_BGSAVE_START（等待bgsave运行）、REDIS_REPL_WAIT_BGSAVE_END（bgsave已dump db，该bulk传输了）、REDIS_REPL_SEND_BULK（正在bulk传输）、REDIS_REPL_ONLINE（已完成开始的bulk传输，以后只需发送更新了）。对于slave客户端（发布sync命令），一开始slave.replstate都处于REDIS_REPL_WAIT_BGSAVE_START状态（后面详解syncCommand函数），然后在后台dump db后（backgroundSaveDoneHandler函数），处于REDIS_REPL_WAIT_BGSAVE_END 状态，然后updateSlavesWaitingBgsave会将状态置为REDIS_REPL_SEND_BULK，并设置write事件的函数 sendBulkToSlave，在sendBulkToSlave运行后，状态就变为REDIS_REPL_ONLINE了，此后master会一直调用replicationFeedSlaves给处于REDIS_REPL_ONLINE状态的slave发送新命令。

我们先看处于master端的redis会执行的代码。

slave端都是通过发布sync命令来跟master同步的，sync命令的处理函数syncCommand如下所示。

该函数中的注释足够明了。如果slave的client设置了REDIS_SLAVE标志，说明master已用syncCommand处理了该 slave。如果master还有对这个client的reply没有发送，则返回出错信息。此后若server.bgsavechildpid != -1且有slave处于REDIS_REPL_WAIT_BGSAVE_END状态，则说明dump db的后台进程刚结束，此时新的slave可直接用保存的rdb进行bulk传输（注意复制reply参数，因为master是非阻塞的，此时可能执行了一些命令，call函数会调用replicationFeedSlaves函数将命令参数保存到slave的reply参数中）。如果没有slave处于REDIS_REPL_WAIT_BGSAVE_END状态，但server.bgsavechildpid != -1，则说明bgsave后台进程没有运行完，需要等待其结束（bgsave后台进程结束后会处理等待的slave）。如果server.bgsavechildpid 等于 -1，则需要启动一个后台进程来dump db了。最后将当前client加到master的slaves链表中。

static void syncCommand(redisClient *c) {
    /* ignore SYNC if aleady slave or in monitor mode */
    if (c->flags & REDIS_SLAVE) return;
    /* SYNC can't be issued when the server has pending data to send to
     * the client about already issued commands. We need a fresh reply
     * buffer registering the differences between the BGSAVE and the current
     * dataset, so that we can copy to other slaves if needed. */
    if (listLength(c->reply) != 0) {
        addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
        return;
    }
    redisLog(REDIS_NOTICE,"Slave ask for synchronization");
    /* Here we need to check if there is a background saving operation
     * in progress, or if it is required to start one */
    if (server.bgsavechildpid != -1) {
        /* Ok a background save is in progress. Let's check if it is a good
         * one for replication, i.e. if there is another slave that is
         * registering differences since the server forked to save */
        redisClient *slave;
        listNode *ln;
        listIter li;
        listRewind(server.slaves,&li);
        while((ln = listNext(&li))) {
            slave = ln->value;
            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
        }
        if (ln) {
            /* Perfect, the server is already registering differences for
             * another slave. Set the right state, and copy the buffer. */
            listRelease(c->reply);
            c->reply = listDup(slave->reply);
            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
            redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
        } else {
            /* No way, we need to wait for the next BGSAVE in order to
             * register differences */
            c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
            redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
        }
    } else {
        /* Ok we don't have a BGSAVE in progress, let's start one */
        redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
        if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
            redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
            addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
            return;
        }
        c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
    }
    c->repldbfd = -1;
    c->flags |= REDIS_SLAVE;
    c->slaveseldb = 0;
    listAddNodeTail(server.slaves,c);
    return;
}

此后slave无论处于REDIS_REPL_WAIT_BGSAVE_START还是REDIS_REPL_WAIT_BGSAVE_END，都只能等 dump db的后台进程运行结束后才会被处理。该进程结束后会执行backgroundSaveDoneHandler函数，而该函数调用 updateSlavesWaitingBgsave来处理slaves。

updateSlavesWaitingBgsave和syncCommand一样，涉及到slave的几个状态变换。对于等待dump db的slave，master都会将其放入server.slaves 链表中。此时，若slave->replstate == REDIS_REPL_WAIT_BGSAVE_START，说明当前dump db不是该slave需要的，redis需要重新启动后台进程来dump db。若slave->replstate == REDIS_REPL_WAIT_BGSAVE_END，则说明当前dump db正是该slave所需要的，此时设置slave的write事件的处理函数sendBulkToSlave。

static void updateSlavesWaitingBgsave(int bgsaveerr) {
    listNode *ln;
    int startbgsave = 0;
    listIter li;
    listRewind(server.slaves,&li);
    while((ln = listNext(&li))) {
        redisClient *slave = ln->value;
        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
            startbgsave = 1;
            slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
        } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
            struct redis_stat buf;
            if (bgsaveerr != REDIS_OK) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
                continue;
            }
            if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
                redis_fstat(slave->repldbfd,&buf) == -1) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
                continue;
            }
            slave->repldboff = 0;
            slave->repldbsize = buf.st_size;
            slave->replstate = REDIS_REPL_SEND_BULK;
            aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
            if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
                freeClient(slave);
                continue;
            }
        }
    }
    if (startbgsave) {
        if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
            listIter li;
            listRewind(server.slaves,&li);
            redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
            while((ln = listNext(&li))) {
                redisClient *slave = ln->value;
                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
                    freeClient(slave);
            }
        }
    }
}

sendBulkToSlave 的逻辑不复杂。它根据slave->repldbfd指向的db，先从dump后的rdb文件中读入db数据，然后发送。发送完后会删除write 事件，设置slave->replstate状态为REDIS_REPL_ONLINE，此后master就会在收到命令后调用call函数，然后使用replicationFeedSlaves同步更新该slave了。replicationFeedSlaves也是遍历slave链表，对处于REDIS_REPL_ONLINE状态的slave，发送当前命令及其参数。

 static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
    redisClient *slave = privdata;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    char buf[REDIS_IOBUF_LEN];
    ssize_t nwritten, buflen;
    if (slave->repldboff == 0) {
        /* Write the bulk write count before to transfer the DB. In theory here
         * we don't know how much room there is in the output buffer of the
         * socket, but in pratice SO_SNDLOWAT (the minimum count for output
         * operations) will never be smaller than the few bytes we need. */
        sds bulkcount;
        bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
            slave->repldbsize);
        if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
        {
            sdsfree(bulkcount);
            freeClient(slave);
            return;
        }
        sdsfree(bulkcount);
    }
    lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
    buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
    if (buflen <= 0) {
        redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
            (buflen == 0) ? "premature EOF" : strerror(errno));
        freeClient(slave);
        return;
    }
    if ((nwritten = write(fd,buf,buflen)) == -1) {
        redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
            strerror(errno));
        freeClient(slave);
        return;
    }
    slave->repldboff += nwritten;
    if (slave->repldboff == slave->repldbsize) {
        close(slave->repldbfd);
        slave->repldbfd = -1;
        aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
        slave->replstate = REDIS_REPL_ONLINE;
        if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
            sendReplyToClient, slave) == AE_ERR) {
            freeClient(slave);
            return;
        }
        addReplySds(slave,sdsempty());
        redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
    }
}

接下来我们看看redis作为slave是如何运行的。

redis 作为slave（当然也可以使用普通的client作为slave端，这样则跟具体client的实现有关了）时，需要在配置文件中指明master的位置，在loadServerConfig读取配置参数时，会将server.replstate设置为REDIS_REPL_CONNECT状态。处于此状态的redis需要运行到serverCron后才能使用syncWithMaster来和master进行初始同步。查看syncWithMaster的代码可知，其实也向master发布sync命令来建立主从关系的，另外，该函数接收、发送数据时使用的是syncRead、syncWrite函数，而这些函数是阻塞的，因此，redis作为slave运行时，建立最初的主从关系时也是阻塞的。

 /* Check if we should connect to a MASTER */
    if (server.replstate == REDIS_REPL_CONNECT && !(loops % 10)) {
        redisLog(REDIS_NOTICE,"Connecting to MASTER...");
        if (syncWithMaster() == REDIS_OK) {
            redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
            if (server.appendonly) rewriteAppendOnlyFileBackground();
        }
    }

另外跟主从复制有关的一个命令就是slaveof命令。此命令是redis主从状态的转换函数，通过前面的分析可知，这只需要更改几个状态即可。

static void slaveofCommand(redisClient *c) {
    if (!strcasecmp(c->argv[1]->ptr,"no") &&
        !strcasecmp(c->argv[2]->ptr,"one")) {
        if (server.masterhost) {
            sdsfree(server.masterhost);
            server.masterhost = NULL;
            if (server.master) freeClient(server.master);
            server.replstate = REDIS_REPL_NONE;
            redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
        }
    } else {
        sdsfree(server.masterhost);
        server.masterhost = sdsdup(c->argv[1]->ptr);
        server.masterport = atoi(c->argv[2]->ptr);
        if (server.master) freeClient(server.master);
        server.replstate = REDIS_REPL_CONNECT;
        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
            server.masterhost, server.masterport);
    }
    addReply(c,shared.ok);
}

原文地址：redis源代码分析19–主从复制, 感谢原作者分享。

内容总结

以上是互联网集市为您收集整理的redis源代码分析19–主从复制全部内容，希望文章能够帮你解决redis源代码分析19–主从复制所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/557362.html

来源：【匿名】

【上一篇】redis源代码分析24–VM（中）【下一篇】redis可以做什么？

更多 ►

【redis源代码分析19–主从复制】教程文章相关的互联网学习教程文章

Redis源代码分析之sds, 动态数组【图】

Redis是用C语言编写的。C语言处理字符串一向是个难点。很容易出现内存越界问题。其它高级语言很容易实现的字符串拼接，在C这里却是百般艰难。因为需要实现计算出字符串所占内存的大小。即不能过大（浪费内存），也不能太小（越界）。甚至在某个用C语言实现的项目中出现了这样的代码即，先计算出字符串的大小。然后申请内存，再拼接字符串。这样的操作几乎是无法忍受的。特别是当我们的字符串构成比较复杂，或者字符串经常需要发生...

scrapy-redis源代码分析【图】

原创文章，链接：http://blog.csdn.net/u012150179/article/details/38226253 + (I) connection.py 负责依据setting中配置实例化redis连接。被dupefilter和scheduler调用。总之涉及到redis存取的都要使用到这个模块。(II) dupefilter.py 负责运行requst的去重。实现的非常有技巧性，使用redis的set数据结构。可是注意scheduler并不使用当中用于在这个模块中实现的dupefilter键做request的调度。而是使用queue.py模块中实现的queu...

redis 源代码分析(一) 内存管理【代码】

一，redis内存管理介绍 redis是一个基于内存的key-value的数据库，其内存管理是很重要的，为了屏蔽不同平台之间的差异，以及统计内存占用量等，redis对内存分配函数进行了一层封装，程序中统一使用zmalloc，zfree一系列函数，其相应的源代码在src/zmalloc.h和src/zmalloc.c两个文件里，源代码点这里。二，redis内存管理源代码分析redis封装是为了屏蔽底层平台的差异，同一时候方便自己实现相关的函数，我们能够通过src/zmalloc.h ...

Redis源代码分析（二十七）--- rio制I/O包裹

I/O每个操作系统，它的一个组成部分。和I/O业务质量，在一定程度上也影响了系统的效率。今天，我在了解了Redis中间I/O的，相同的，Redis在他自己的系统中。也封装了一个I/O层。简称RIO。得先看看RIO中有什么东西喽:struct _rio {/* Backend functions.* Since this functions do not tolerate short writes or reads the return* value is simplified to: zero on error, non zero on complete success. *//* 数据流的读方法 */siz...

redis源代码分析20–发布/订阅

redis的发布/订阅（publish/subscribe）功能类似于传统的消息路由功能，发布者发布消息，订阅者接收消息，沟通发布者和订阅者之间的桥梁是订阅的channel或者pattern。发布者向指定的publish或者pattern发布消息，订阅者阻塞在订阅的channel或者pattern。可以redis的发布/订阅（publish/subscribe）功能类似于传统的消息路由功能，发布者发布消息，订阅者接收消息，沟通发布者和订阅者之间的桥梁是订阅的channel或者pattern。发布者...

redis源代码分析18–持久化之aof

Redis的aof功能的目的是在性能和持久化粒度上对持久化机制提供更好的支持。快照方式持久化的粒度有时间（秒）和改变的key数两种，如果持久化的粒度较小，对性能会有较大的影响，因为每次都是dump整个db；如果持久化的粒度较大，则在指定时间内指定数目的数Redis的aof功能的目的是在性能和持久化粒度上对持久化机制提供更好的支持。快照方式持久化的粒度有时间（秒）和改变的key数两种，如果持久化的粒度较小，对性能会有较大的影...

redis源代码分析17–持久化之快照

redis的持久化支持快照方式。快照方式会将整个db dump到磁盘上。 client 可以发布save/bgsave命令让server将db dump到磁盘上。其中bgsave会执行后台dump（新建子进程执行dump），而save是阻塞式的dump db，会影响其他client的命令执行。除了发布命令执行快照redis的持久化支持快照方式。快照方式会将整个db dump到磁盘上。 client 可以发布save/bgsave命令让server将db dump到磁盘上。其中bgsave会执行后台dump（新建子进程执行dum...

redis源代码分析23–VM（上）

VM是Redis2.0新增的一个功能。在没有VM之前，redis会把db中的所有数据放在内存中。随着redis的不断运行，所使用的内存会越来越大。但同时，client对某些数据的访问频度明显会比其他数据高。redis引入VM功能来试图解决这个问题。简言之，VM使得redis会把很少VM是Redis2.0新增的一个功能。在没有VM之前，redis会把db中的所有数据放在内存中。随着redis的不断运行，所使用的内存会越来越大。但同时，client对某些数据的访问频度明显会...

redis源代码分析21–事务

redis的事务较简单，并不具备事务的acid的全部特征。主要原因之一是redis事务中的命令并不是立即执行的，会一直排队到发布exec命令才执行所有的命令；另一个主要原因是它不支持回滚，事务中的命令可以部分成功，部分失败，命令失败时跟不在事务上下文执行时redis的事务较简单，并不具备事务的acid的全部特征。主要原因之一是redis事务中的命令并不是立即执行的，会一直排队到发布exec命令才执行所有的命令；另一个主要原因是它不支...

redis源代码分析22–协议

redis默认使用tcp协议的6379端口，其协议是文本行格式的而不是二进制格式的，每一行都以\r\n结尾，非常容易理解。参考ProtocolSpecification.html就知道，发布到redis的命令有如下几种返回格式（对于不存在的值，会返回-1，此时client library应返回合适的nredis默认使用tcp协议的6379端口，其协议是文本行格式的而不是二进制格式的，每一行都以”\r\n”结尾，非常容易理解。参考ProtocolSpecification.html就知道，发布到redis的...

redis源代码分析16–阻塞式命令

redis现在只支持对list的阻塞式操作，相关的两个命令是brpop和blpop。这两个命令在list中有元素时，跟普通的pop没有区别，弹出list的一个元素，然后返回。但在list没有元素时，会为redisClient设置REDIS_BLOCKED标志，然后client阻塞（设置REDIS_BLOCKED标redis现在只支持对list的阻塞式操作，相关的两个命令是brpop和blpop。这两个命令在list中有元素时，跟普通的pop没有区别，弹出list的一个元素，然后返回。但在list没有元素时...

redis源代码分析25–VM（下）

这一节介绍下redis中的多线程机制。先看看多线程换出的机制。 serverCron函数中调用 vmSwapOneObjectThreaded开始多线程方式换出value，vmSwapOneObjectThreaded会调用 vmSwapOneObject（参看上一节的解释），而vmSwapOneObject最终会调用vmSwapObjectThrea这一节介绍下redis中的多线程机制。先看看多线程换出的机制。 serverCron函数中调用 vmSwapOneObjectThreaded开始多线程方式换出value，vmSwapOneObjectThreaded会调用 vmS...