首页 > 代码库 > REDIS 主从复制
REDIS 主从复制
REDIS目前给出了一个异步的主从复制版本系统。在redis里 提供了几种方式来完成这个工作。 主从复制主要对应在redis/replication.c这个文件里。源码框架里 分为3部分： Master部分/SLAVE部分/复制核心部分
其实主从复制我个人觉得比较难的点就是在于每次重启之后 master/slave传递数据的模式方式
首先对于slave来讲 是主动连接他的master
int connectWithMaster(void) { int fd; fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport); if (fd == -1) { redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", strerror(errno)); return REDIS_ERR; } 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; } server.repl_transfer_lastio = server.unixtime; server.repl_transfer_s = fd; server.repl_state = REDIS_REPL_CONNECTING; return REDIS_OK;}
aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) 就是注册一个可读和可写的事件 注意处理事件函数是syncwithMaster.
rep_state状态就变成了REDIS_PREPL_CONNECTING.相关的server的replication做出相应的调整。
现在我们就进入syncwithMaster进去看看：
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. */ aeDeleteFileEvent(server.el,fd,AE_WRITABLE); 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. */ syncWrite(fd,"PING\r\n",6,100); return; }
首先客户端slave 发送一个PING给server master 这个是带有超时的一个回应， 状态就改成了REDIS_REPL_RECEIVE_PONG 按理来说Master收到了会做出相应的动作。对于slave端而言 下一步就是REDIS_REPL_RECEIVE_PONG这个状态了。其实就是准备接受某个值了
buf[0] = ‘\0‘; 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; }
这是PONG状态下的做的核心事情：读出来 然后判断是否是有相应的内容。
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)
给MASETER发送一个SYNC 然后就进入可读状态 注册了一个readSyncBulkPayload。等下来看看这个事件函数 后面要做的事情就是设置相应的位了：
server.repl_state = REDIS_REPL_TRANSFER; 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);
repl_transfer_size设置-1 表示从master收到的文件大小为-1 。状态变成了REPL_TRANSFER。 现在进入readSyncBulkPayload看看这个函数是怎么接受的：
server.repl_transfer_size = strtol(buf+1,NULL,10);
首先确定了对方要发送多大的文件 然后读到buf 在写到rdb相应的文件里面。
left = server.repl_transfer_size - server.repl_transfer_read; readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf); nread = read(fd,buf,readlen);.............................................................................................................................................................................................write(server.repl_transfer_fd,buf,nread) != nread) server.repl_transfer_read += nread;/* Check if the transfer is now complete */ if (server.repl_transfer_read == server.repl_transfer_size) { 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; } redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory"); signalFlushedDb(-1); emptyDb(); /* Before loading the DB into memory we need to delete the readable * handler, otherwise it will get called recursively since * rdbLoad() will call the event loop to process events from time to * time for non blocking loading. */ aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); if (rdbLoad(server.rdb_filename) != REDIS_OK) { redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk"); replicationAbortSyncTransfer(); return; }
如果2者是相等的 read和transfer_size相等 首先替换名字 替换成rdb_filename名字 然后清空db emptyDb()销毁可读事件 最后调用rdbLoad在本地重新构建一个key_value数据库副本。【master的】 所有的动作的操作都已经完成了 。这里的发送大小双方应该会有一个限定。我们可以从master部分来找到相应的事件出来：
对于主服务器来讲 除了客户端发送了一个PING之后期望得到主机的一个回复之外 真正对这个主从复制有用的应该是从服务器这个操作：
write(fd,“SYNC\r\n”,buf).这个动作一发出： master就会调用SYNCcommand()来完成相应的拷贝动作： 首先进入SYNCcommand()函数进去看看是一个什么情况：
要完成复制 首先要在一个合适的时机：master进入了一个bgsave操作。要保证rdb文件是一个最新的文件。 对于master而言 先看看rdb_pid!=-1如果条件满足 表明正在做这个操作 master只需要等完成了才做相应的动作 而如果不是sync就会触发一个bgsave操作。然后对于主进程而言： 都会设置状态为：WAIT_BGSAVE_END.这个时候syncCommand就完成了 而复制操作还没有开始 进行往下面看
而做bgsaveCommand操作时 都会调用一个function：updateSlavesWaitingBgsave 这样就不会出现同步等待现象了。
if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 ||
打开相应的repldbfd,准备复制文件了：
aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR)
注册了一个sendBulkToSlave.发给salve 这个部分要注意好：
需要注意的点： 1） 发送缓冲区大小怎么设置 和slave是不是设置一样 2）发送是一个同步的过程还是异步的过程
REDIS_IOBUF_LEN: 1024*16 这个变量就是一次读的rdb量 还是进入sendBulkToSlave()看看：
{ 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 practice 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; } redisLog(REDIS_NOTICE,"Synchronization with slave succeeded"); }}
如果repldoff==0 表明是第一次初始化 也就是会发送一个应该发送的长度数据给对方slave.这是第一次发送。注意这里调用write如果成功之后会继续。lseek(slave->repldbfd,slave->repldboff,SEEK_SET); 每次会定位到相应的位置，这个非常恼火 调用磁盘的一个随机操作，比较耗时 如果文件很大 对性能影响比较大。 buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN); 然后读到内存中 然后在做write操作。因为没有关闭事件模型，所以EPOLL轮询时都会认为这个事件还需要执行：还是准备好的，所以继续调用这个函数，从本质上来讲 可以算是一个异步的操作。所以不会出现服务的一个中断现象，但是lseek是比较耗时的，在复制完成了 关闭fd的可读状态 并且把replstate状态标记成REPL_ONLINE，这个状态就是命令传播状态。注册了一个一个新的函数sendReplyToClient，当然把之前的函数事件del掉。所以每次Server端给的buf是比slave端小很多.主从复制核心就是这里了。
REDIS 主从复制