redis数据库之主从复制
jopen
11年前
redis除了基本功能外，还提供了主从复制功能。一个redis服务可以有多个slave服务，而这个slave服务又可以有slave服务。master服务把属于自己的slave服务用链表管理起来，也就是struct redisServer中的slaves成员，slave服务会通过redisServer中的masterhost和masterport来标识它的master服务的ip和port。
redis有两种方式来标识从属于哪个master服务：
1、在 配置文件中配置slaveof masterhost masterport
2、发送slaveof命令。
同样redis也提供了两种方式来同步主从的数据库的。
1、通过定时器来完成同步
2、master服务每次执行的命令都会根据情况发送一份给slave服务。
首先来讲述下定时器完成同步的实现：
redis服务的定时功能都是通过serverCron完成，而在serverCron中会调用replicationCron，这个函数就完成了不同的功能。
void replicationCron(void) { ....... if (server.repl_state == REDIS_REPL_CONNECT) { redisLog(REDIS_NOTICE,"Connecting to MASTER..."); if (connectWithMaster() == REDIS_OK) { redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started"); } } ....... }replicationCron检查一些超时情况做一些超时的处理，然后会调用connectWithMaster去连接master服务。 int connectWithMaster(void) { fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport); ...... if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) == AE_ERR) ...... }connectWithMaster首先会向master服务发起连接，然后创建一个读写事件并把设置server.repl_state = REDIS_REPL_CONNECTING;void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) { ...... 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; } ...... if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL) == AE_ERR) ...... server.repl_state = REDIS_REPL_TRANSFER; ...... }syncWithMaster会发送SYNC命令给master服务，然后设置可读事件的handler，并把slave的状态设置为传输状态。下面来看下master服务接收到sync命令的处理： void syncCommand(redisClient *c) { ...... // 检查是否已经有 BGSAVE 在执行，否则就创建一个新的 BGSAVE 任务 if (server.rdb_child_pid != -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 */ // 已有 BGSAVE 在执行，检查它能否用于当前客户端的 SYNC 操作 redisClient *slave; listNode *ln; listIter li; // 检查是否有其他客户端在等待 SYNC 进行 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. */ // 找到一个同样在等到 SYNC 的客户端 // 设置当前客户端的状态，并复制 buffer 。 copyClientOutputBuffer(c,slave); 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 */ // 没有客户端在等待 SYNC ，当前客户端只能等待下次 BGSAVE 进行 c->replstate = REDIS_REPL_WAIT_BGSAVE_START; redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC"); } } else { // 没有 BGSAVE 在进行，自己启动一个。 /* Ok we don't have a BGSAVE in progress, let's start one */ redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC"); if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) { redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE"); addReplyError(c,"Unable to perform background save"); return; } // 等待 BGSAVE 结束 c->replstate = REDIS_REPL_WAIT_BGSAVE_END; } c->repldbfd = -1; c->flags |= REDIS_SLAVE; c->slaveseldb = 0; listAddNodeTail(server.slaves,c); ...... }这里并不是真正处理同步的，而是把slave插入到master中slaves链表中等待真正同步的操作。那什么时候才是真正同步的操作呢？请看updateSlavesWaitingBgsave void updateSlavesWaitingBgsave(int bgsaveerr) { ...... listRewind(server.slaves,&li); while((ln = listNext(&li))) { redisClient *slave = ln->value; if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) { // 告诉那些这次不能同步的客户端，可以等待下次 BGSAVE 了。 startbgsave = 1; slave->replstate = REDIS_REPL_WAIT_BGSAVE_END; } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) { // 这些是本次可以同步的客户端 struct redis_stat buf; // 如果 BGSAVE 失败，释放 slave 节点 if (bgsaveerr != REDIS_OK) { freeClient(slave); redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error"); continue; } // 打开 .rdb 文件 if ((slave->repldbfd = open(server.rdb_filename,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; // 数据库大小（.rdb 文件的大小） slave->repldbsize = buf.st_size; // 状态 slave->replstate = REDIS_REPL_SEND_BULK; // 清除 slave->fd 的写事件 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); // 创建一个将 .rdb 文件内容发送到附属节点的写事件 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) { freeClient(slave); continue; } } } ...... }这个函数会每个slave创建一个可写的事件，并从rdb文件中把数据读出来，通过sendBulkToSlave发送给slave。master发送完后，slave接受数据并进行处理，上面已经看到slave给读事件设置了handler（readSyncBulkPayload） 以上就是定时器实现主从同步，第二种实现主从同步的情况比较简单。
每次master接收到客户端指令都会调用call这个函数：
void call(redisClient *c, int flags) { ...... if (flags & REDIS_CALL_PROPAGATE) { int flags = REDIS_PROPAGATE_NONE; if (c->cmd->flags & REDIS_CMD_FORCE_REPLICATION) flags |= REDIS_PROPAGATE_REPL; if (dirty) flags |= (REDIS_PROPAGATE_REPL | REDIS_PROPAGATE_AOF); if (flags != REDIS_PROPAGATE_NONE) propagate(c->cmd,c->db->id,c->argv,c->argc,flags); } ...... }propagate就是实现第二种主从同步。 void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int flags) { if (server.aof_state != REDIS_AOF_OFF && flags & REDIS_PROPAGATE_AOF) feedAppendOnlyFile(cmd,dbid,argv,argc); if (flags & REDIS_PROPAGATE_REPL && listLength(server.slaves)) replicationFeedSlaves(server.slaves,dbid,argv,argc); }从函数代码中可以看出，reiplicationFeedSlaves就是真正实现主从同步第二种方式的地方，这个函数也比较简单，这里就列举出来啦。 这里要提出一个问题：master服务的命令会同步给slave，但是如果slave服务发生变化，master并不会得到同步，这种情况怎么办？还是slave只允许读操作，而不进行写操作，但是slave服务也可能是别的redis服务的master服务，这样就感觉不合理了。为什么要让slave服务又称为master服务呢？