/* 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
[cpp]
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;
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这个函数：
[cpp]
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就是实现第二种主从同步。
[cpp]
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,