参数详解复制进程_Redis源码分析13——主从复制

struct redisServer{
/* Replication (master) */char replid[CONFIG_RUN_ID_SIZE&＃43;1]; /* My current replication ID. */char replid2[CONFIG_RUN_ID_SIZE&＃43;1]; /* replid inherited from master*/
}


#define REPL_STATE_NONE 0 /* No active replication */
#define REPL_STATE_CONNECT 1 /* Must connect to master */
#define REPL_STATE_CONNECTING 2 /* Connecting to master */
/* --- Handshake states, must be ordered --- */
#define REPL_STATE_RECEIVE_PONG 3 /* Wait for PING reply */
#define REPL_STATE_SEND_AUTH 4 /* Send AUTH to master */
#define REPL_STATE_RECEIVE_AUTH 5 /* Wait for AUTH reply */
#define REPL_STATE_SEND_PORT 6 /* Send REPLCONF listening-port */
#define REPL_STATE_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_IP 8 /* Send REPLCONF ip-address */
#define REPL_STATE_RECEIVE_IP 9 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_CAPA 10 /* Send REPLCONF capa */
#define REPL_STATE_RECEIVE_CAPA 11 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_PSYNC 12 /* Send PSYNC */
#define REPL_STATE_RECEIVE_PSYNC 13 /* Wait for PSYNC reply */
/* --- End of handshake states --- */
#define REPL_STATE_TRANSFER 14 /* Receiving .rdb from master */
#define REPL_STATE_CONNECTED 15 /* Connected to master */


void replicaofCommand(client *c) {/* The special host/port combination "NO" "ONE" turns the instance* into a master. Otherwise the new master address is set. */if (!strcasecmp(c->argv[1]->ptr,"no") &&!strcasecmp(c->argv[2]->ptr,"one")) {if (server.masterhost) {replicationUnsetMaster();sds client &＃61; catClientInfoString(sdsempty(),c);serverLog(LL_NOTICE,"MASTER MODE enabled (user request from &＃39;%s&＃39;)",client);sdsfree(client);}} else {...}addReply(c,shared.ok);
}/* 取消主从复制&＃xff0c;将他自己设置为主&＃xff0c;恢复最初状态. */
void replicationUnsetMaster(void) {if (server.masterhost &＃61;&＃61; NULL) return; /* Nothing to do. */sdsfree(server.masterhost);server.masterhost &＃61; NULL;shiftReplicationId();if (server.master) freeClient(server.master);replicationDiscardCachedMaster();cancelReplicationHandshake();disconnectSlaves();server.repl_state &＃61; REPL_STATE_NONE;server.slaveseldb &＃61; -1;server.repl_no_slaves_since &＃61; server.unixtime;
}


struct redisServer
{...char *masterhost; /* Hostname of master */int masterport; /* Port of master */...
}


/* Set replication to the specified master address and port. */
void replicationSetMaster(char *ip, int port) {int was_master &＃61; server.masterhost &＃61;&＃61; NULL;//记录新的主节点sdsfree(server.masterhost);server.masterhost &＃61; sdsnew(ip);server.masterport &＃61; port;if (server.master) {freeClient(server.master);}disconnectAllBlockedClients(); /* Clients blocked in master, now slave. *//* Force our slaves to resync with us as well. They may hopefully be able* to partially resync with us, but we can notify the replid change. */disconnectSlaves();cancelReplicationHandshake();//这里缓存主节点if (was_master) replicationCacheMasterUsingMyself();server.repl_state &＃61; REPL_STATE_CONNECT;
}


struct{。。。
int slave_listening_port; /* As configured with: SLAVECONF listening-port */char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */。。。
} client;


#define SLAVE_CAPA_NONE 0
#define SLAVE_CAPA_EOF (1<<0) /* Can parse the RDB EOF streaming format. */
#define SLAVE_CAPA_PSYNC2 (1<<1) /* Supports PSYNC2 protocol. */


struct client{...int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */...
}


run_with_period(1000) replicationCron();


replicationCron---->connectWithMaster---->syncWithMasterint connectWithMaster(void) {//注册可写事件if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) &＃61;&＃61;AE_ERR){close(fd);serverLog(LL_WARNING,"Can&＃39;t create readable event for SYNC");return C_ERR;}//更新状态server.repl_transfer_lastio &＃61; server.unixtime;server.repl_transfer_s &＃61; fd;server.repl_state &＃61; REPL_STATE_CONNECTING;return C_OK;
}/* This handler fires when the non blocking connect was able to* establish a connection with the master. */
void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {char tmpfile[256], *err &＃61; NULL;int dfd &＃61; -1, maxtries &＃61; 5;int sockerr &＃61; 0, psync_result;socklen_t errlen &＃61; sizeof(sockerr);UNUSED(el);UNUSED(privdata);UNUSED(mask);/* If this event fired after the user turned the instance into a master* with SLAVEOF NO ONE we must just return ASAP. */if (server.repl_state &＃61;&＃61; REPL_STATE_NONE) {close(fd);return;}/* Check for errors in the socket: after a non blocking connect() we* may find that the socket is in error state. */if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) &＃61;&＃61; -1)sockerr &＃61; errno;if (sockerr) {serverLog(LL_WARNING,"Error condition on socket for SYNC: %s",strerror(sockerr));goto error;}/* Send a PING to check the master is able to reply without errors. */if (server.repl_state &＃61;&＃61; REPL_STATE_CONNECTING) {serverLog(LL_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 &＃61; REPL_STATE_RECEIVE_PONG;/* Send the PING, don&＃39;t check for errors at all, we have the timeout* that will take care about this. */err &＃61; sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PING",NULL);if (err) goto write_error;return;}/* Receive the PONG command. */if (server.repl_state &＃61;&＃61; REPL_STATE_RECEIVE_PONG) {err &＃61; sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);/* We accept only two replies as valid, a positive &＃43;PONG reply* (we just check for "&＃43;") or an authentication error.* Note that older versions of Redis replied with "operation not* permitted" instead of using a proper error code, so we test* both. */if (err[0] !&＃61; &＃39;&＃43;&＃39; &&strncmp(err,"-NOAUTH",7) !&＃61; 0 &&strncmp(err,"-ERR operation not permitted",28) !&＃61; 0){serverLog(LL_WARNING,"Error reply to PING from master: &＃39;%s&＃39;",err);sdsfree(err);goto error;} else {serverLog(LL_NOTICE,"Master replied to PING, replication can continue...");}sdsfree(err);server.repl_state &＃61; REPL_STATE_SEND_AUTH;}/* AUTH with the master if required. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_AUTH) {if (server.masterauth) {err &＃61; sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",server.masterauth,NULL);if (err) goto write_error;server.repl_state &＃61; REPL_STATE_RECEIVE_AUTH;return;} else {server.repl_state &＃61; REPL_STATE_SEND_PORT;}}/* Receive AUTH reply. */if (server.repl_state &＃61;&＃61; REPL_STATE_RECEIVE_AUTH) {err &＃61; sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);if (err[0] &＃61;&＃61; &＃39;-&＃39;) {serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);sdsfree(err);goto error;}sdsfree(err);server.repl_state &＃61; REPL_STATE_SEND_PORT;}/* Set the slave port, so that Master&＃39;s INFO command can list the* slave listening port correctly. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_PORT) {sds port &＃61; sdsfromlonglong(server.slave_announce_port ?server.slave_announce_port : server.port);err &＃61; sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF","listening-port",port, NULL);sdsfree(port);if (err) goto write_error;sdsfree(err);server.repl_state &＃61; REPL_STATE_RECEIVE_PORT;return;}/* Receive REPLCONF listening-port reply. */if (server.repl_state &＃61;&＃61; REPL_STATE_RECEIVE_PORT) {err &＃61; sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);/* Ignore the error if any, not all the Redis versions support* REPLCONF listening-port. */if (err[0] &＃61;&＃61; &＃39;-&＃39;) {serverLog(LL_NOTICE,"(Non critical) Master does not understand ""REPLCONF listening-port: %s", err);}sdsfree(err);server.repl_state &＃61; REPL_STATE_SEND_IP;}/* Skip REPLCONF ip-address if there is no slave-announce-ip option set. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_IP &&server.slave_announce_ip &＃61;&＃61; NULL){server.repl_state &＃61; REPL_STATE_SEND_CAPA;}/* Set the slave ip, so that Master&＃39;s INFO command can list the* slave IP address port correctly in case of port forwarding or NAT. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_IP) {err &＃61; sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF","ip-address",server.slave_announce_ip, NULL);if (err) goto write_error;sdsfree(err);server.repl_state &＃61; REPL_STATE_RECEIVE_IP;return;}/* Receive REPLCONF ip-address reply. */if (server.repl_state &＃61;&＃61; REPL_STATE_RECEIVE_IP) {err &＃61; sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);/* Ignore the error if any, not all the Redis versions support* REPLCONF listening-port. */if (err[0] &＃61;&＃61; &＃39;-&＃39;) {serverLog(LL_NOTICE,"(Non critical) Master does not understand ""REPLCONF ip-address: %s", err);}sdsfree(err);server.repl_state &＃61; REPL_STATE_SEND_CAPA;}/* Inform the master of our (slave) capabilities.** EOF: supports EOF-style RDB transfer for diskless replication.* PSYNC2: supports PSYNC v2, so understands &＃43;CONTINUE .** The master will ignore capabilities it does not understand. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_CAPA) {err &＃61; sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF","capa","eof","capa","psync2",NULL);if (err) goto write_error;sdsfree(err);server.repl_state &＃61; REPL_STATE_RECEIVE_CAPA;return;}/* Receive CAPA reply. */if (server.repl_state &＃61;&＃61; REPL_STATE_RECEIVE_CAPA) {err &＃61; sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);/* Ignore the error if any, not all the Redis versions support* REPLCONF capa. */if (err[0] &＃61;&＃61; &＃39;-&＃39;) {serverLog(LL_NOTICE,"(Non critical) Master does not understand ""REPLCONF capa: %s", err);}sdsfree(err);server.repl_state &＃61; REPL_STATE_SEND_PSYNC;}/* Try a partial resynchonization. If we don&＃39;t have a cached master* slaveTryPartialResynchronization() will at least try to use PSYNC* to start a full resynchronization so that we get the master run id* and the global offset, to try a partial resync at the next* reconnection attempt. */if (server.repl_state &＃61;&＃61; REPL_STATE_SEND_PSYNC) {if (slaveTryPartialResynchronization(fd,0) &＃61;&＃61; PSYNC_WRITE_ERROR) {err &＃61; sdsnew("Write error sending the PSYNC command.");goto write_error;}server.repl_state &＃61; REPL_STATE_RECEIVE_PSYNC;return;}/* If reached this point, we should be in REPL_STATE_RECEIVE_PSYNC. */if (server.repl_state !&＃61; REPL_STATE_RECEIVE_PSYNC) {serverLog(LL_WARNING,"syncWithMaster(): state machine error, ""state should be RECEIVE_PSYNC but is %d",server.repl_state);goto error;}psync_result &＃61; slaveTryPartialResynchronization(fd,1);if (psync_result &＃61;&＃61; PSYNC_WAIT_REPLY) return; /* Try again later... *//* If the master is in an transient error, we should try to PSYNC* from scratch later, so go to the error path. This happens when* the server is loading the dataset or is not connected with its* master and so forth. */if (psync_result &＃61;&＃61; PSYNC_TRY_LATER) goto error;/* Note: if PSYNC does not return WAIT_REPLY, it will take care of* uninstalling the read handler from the file descriptor. */if (psync_result &＃61;&＃61; PSYNC_CONTINUE) {serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization.");return;}/* PSYNC failed or is not supported: we want our slaves to resync with us* as well, if we have any sub-slaves. The master may transfer us an* entirely different data set and we have no way to incrementally feed* our slaves after that. */disconnectSlaves(); /* Force our slaves to resync with us as well. */freeReplicationBacklog(); /* Don&＃39;t allow our chained slaves to PSYNC. *//* Fall back to SYNC if needed. Otherwise psync_result &＃61;&＃61; PSYNC_FULLRESYNC* and the server.master_replid and master_initial_offset are* already populated. */if (psync_result &＃61;&＃61; PSYNC_NOT_SUPPORTED) {serverLog(LL_NOTICE,"Retrying with SYNC...");if (syncWrite(fd,"SYNCrn",6,server.repl_syncio_timeout*1000) &＃61;&＃61; -1) {serverLog(LL_WARNING,"I/O error writing to MASTER: %s",strerror(errno));goto error;}}/* Prepare a suitable temp file for bulk transfer */while(maxtries--) {snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());dfd &＃61; open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);if (dfd !&＃61; -1) break;sleep(1);}if (dfd &＃61;&＃61; -1) {serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> REPLICA synchronization: %s",strerror(errno));goto error;}/* Setup the non blocking download of the bulk file. */if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)&＃61;&＃61; AE_ERR){serverLog(LL_WARNING,"Can&＃39;t create readable event for SYNC: %s (fd&＃61;%d)",strerror(errno),fd);goto error;}server.repl_state &＃61; REPL_STATE_TRANSFER;server.repl_transfer_size &＃61; -1;server.repl_transfer_read &＃61; 0;server.repl_transfer_last_fsync_off &＃61; 0;server.repl_transfer_fd &＃61; dfd;server.repl_transfer_lastio &＃61; server.unixtime;server.repl_transfer_tmpfile &＃61; zstrdup(tmpfile);return;error:aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);if (dfd !&＃61; -1) close(dfd);close(fd);server.repl_transfer_s &＃61; -1;server.repl_state &＃61; REPL_STATE_CONNECT;return;write_error: /* Handle sendSynchronousCommand(SYNC_CMD_WRITE) errors. */serverLog(LL_WARNING,"Sending command to master in replication handshake: %s", err);sdsfree(err);goto error;
}


/* When a background RDB saving/transfer terminates, call the right handler. */
void backgroundSaveDoneHandler(int exitcode, int bysignal) {switch(server.rdb_child_type) {case RDB_CHILD_TYPE_DISK:backgroundSaveDoneHandlerDisk(exitcode,bysignal);break;case RDB_CHILD_TYPE_SOCKET:backgroundSaveDoneHandlerSocket(exitcode,bysignal);break;default:serverPanic("Unknown RDB child type.");break;}
}


/* SYNC and PSYNC command implemenation. */
void syncCommand(client *c) {..../* Setup the slave as one waiting for BGSAVE to start. The following code* paths will change the state if we handle the slave differently. */c->replstate &＃61; SLAVE_STATE_WAIT_BGSAVE_START;if (server.repl_disable_tcp_nodelay)anetDisableTcpNoDelay(NULL, c->fd); /* Non critical if it fails. */c->repldbfd &＃61; -1;c->flags |&＃61; CLIENT_SLAVE;listAddNodeTail(server.slaves,c);/* Create the replication backlog if needed. */if (listLength(server.slaves) &＃61;&＃61; 1 && server.repl_backlog &＃61;&＃61; NULL) {/* When we create the backlog from scratch, we always use a new* replication ID and clear the ID2, since there is no valid* past history. */changeReplicationId();clearReplicationId2();createReplicationBacklog();}/* CASE 1: BGSAVE is in progress, with disk target. */if (server.rdb_child_pid !&＃61; -1 &&server.rdb_child_type &＃61;&＃61; RDB_CHILD_TYPE_DISK){/* Ok a background save is in progress. Let&＃39;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. */client *slave;listNode *ln;listIter li;listRewind(server.slaves,&li);while((ln &＃61; listNext(&li))) {slave &＃61; ln->value;if (slave->replstate &＃61;&＃61; SLAVE_STATE_WAIT_BGSAVE_END) break;}/* To attach this slave, we check that it has at least all the* capabilities of the slave that triggered the current BGSAVE. */if (ln && ((c->slave_capa & slave->slave_capa) &＃61;&＃61; slave->slave_capa)) {/* Perfect, the server is already registering differences for* another slave. Set the right state, and copy the buffer. */copyClientOutputBuffer(c,slave);replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);serverLog(LL_NOTICE,"Waiting for end of BGSAVE for SYNC");} else {/* No way, we need to wait for the next BGSAVE in order to* register differences. */serverLog(LL_NOTICE,"Can&＃39;t attach the replica to the current BGSAVE. Waiting for next BGSAVE for SYNC");}。。。}


void updateSlavesWaitingBgsave(int bgsaveerr, int type) {listNode *ln;int startbgsave &＃61; 0;int mincapa &＃61; -1;listIter li;listRewind(server.slaves,&li);while((ln &＃61; listNext(&li))) {client *slave &＃61; ln->value;if (slave->replstate &＃61;&＃61; SLAVE_STATE_WAIT_BGSAVE_START) {startbgsave &＃61; 1;mincapa &＃61; (mincapa &＃61;&＃61; -1) ? slave->slave_capa :(mincapa & slave->slave_capa);} else if (slave->replstate &＃61;&＃61; SLAVE_STATE_WAIT_BGSAVE_END) {struct redis_stat buf;/* If this was an RDB on disk save, we have to prepare to send* the RDB from disk to the slave socket. Otherwise if this was* already an RDB -> Slaves socket transfer, used in the case of* diskless replication, our work is trivial, we can just put* the slave online. */if (type &＃61;&＃61; RDB_CHILD_TYPE_SOCKET) {...} else {if (bgsaveerr !&＃61; C_OK) {freeClient(slave);serverLog(LL_WARNING,"SYNC failed. BGSAVE child returned an error");continue;}if ((slave->repldbfd &＃61; open(server.rdb_filename,O_RDONLY)) &＃61;&＃61; -1 ||redis_fstat(slave->repldbfd,&buf) &＃61;&＃61; -1) {freeClient(slave);serverLog(LL_WARNING,"SYNC failed. Can&＃39;t open/stat DB after BGSAVE: %s", strerror(errno));continue;}slave->repldboff &＃61; 0;slave->repldbsize &＃61; buf.st_size;slave->replstate &＃61; SLAVE_STATE_SEND_BULK;slave->replpreamble &＃61; sdscatprintf(sdsempty(),"$%lldrn",(unsigned long long) slave->repldbsize);aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) &＃61;&＃61; AE_ERR) {freeClient(slave);continue;}}}}if (startbgsave) startBgsaveForReplication(mincapa);
}


/* SYNC and PSYNC command implemenation. */
void syncCommand(client *c) {.../* CASE 2: BGSAVE is in progress, with socket target. */} else if (server.rdb_child_pid !&＃61; -1 &&server.rdb_child_type &＃61;&＃61; RDB_CHILD_TYPE_SOCKET){/* There is an RDB child process but it is writing directly to* children sockets. We need to wait for the next BGSAVE* in order to synchronize. */serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC");/* CASE 3: There is no BGSAVE is progress. */} else {if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF)) {/* Diskless replication RDB child is created inside* replicationCron() since we want to delay its start a* few seconds to wait for more slaves to arrive. */if (server.repl_diskless_sync_delay)serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC");} else {//这里if (server.aof_child_pid &＃61;&＃61; -1) {startBgsaveForReplication(c->slave_capa);} else {}}}return;
}


void updateSlavesWaitingBgsave(int bgsaveerr, int type) {listNode *ln;int startbgsave &＃61; 0;int mincapa &＃61; -1;listIter li;listRewind(server.slaves,&li);while((ln &＃61; listNext(&li))) {client *slave &＃61; ln->value;if (slave->replstate &＃61;&＃61; SLAVE_STATE_WAIT_BGSAVE_START) {startbgsave &＃61; 1;mincapa &＃61; (mincapa &＃61;&＃61; -1) ? slave->slave_capa :(mincapa & slave->slave_capa);} else if (slave->replstate &＃61;&＃61; SLAVE_STATE_WAIT_BGSAVE_END) {struct redis_stat buf;if (type &＃61;&＃61; RDB_CHILD_TYPE_SOCKET) {replicationGetSlaveName(slave));//在这里额slave->replstate &＃61; SLAVE_STATE_ONLINE;slave->repl_put_online_on_ack &＃61; 1;slave->repl_ack_time &＃61; server.unixtime; /* Timeout otherwise. */} else {}}}if (startbgsave) startBgsaveForReplication(mincapa);
}


√


/* Setup the non blocking download of the bulk file. */if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)&＃61;&＃61; AE_ERR){serverLog(LL_WARNING,"Can&＃39;t create readable event for SYNC: %s (fd&＃61;%d)",strerror(errno),fd);goto error;}


if(aof_is_enabled) stopAppendOnly();//暂停aof持久化
signalFlushedDb(-1);emptyDb(-1,server.repl_slave_lazy_flush ? EMPTYDB_ASYNC : EMPTYDB_NO_FLAGS,replicationEmptyDbCallback);//清空数据库aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);//删除可读事件serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");rdbSaveInfo rsi &＃61; RDB_SAVE_INFO_INIT;if (rdbLoad(server.rdb_filename,&rsi) !&＃61; C_OK) { //加载RDBserverLog(LL_WARNING,"Failed trying to load the MASTER synchronization DB from disk");cancelReplicationHandshake();//暂停if (aof_is_enabled) restartAOF();return;}/* Final setup of the connected slave <- master link */zfree(server.repl_transfer_tmpfile);close(server.repl_transfer_fd);replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db);server.repl_state &＃61; REPL_STATE_CONNECTED;server.repl_down_since &＃61; 0;/* 记录主节点的runid和偏移量 */memcpy(server.replid,server.master->replid,sizeof(server.replid));server.master_repl_offset &＃61; server.master->reploff;clearReplicationId2();if (server.repl_backlog &＃61;&＃61; NULL) createReplicationBacklog();if (aof_is_enabled) restartAOF();//重启aof