当前位置：首页 > article >正文

【Redis 源码】6AOF持久化

article 2024/10/3 12:49:01

1 AOF功能说明

aof.c 文件是 Redis 中负责 AOF（Append-Only File）持久化的核心文件。AOF 持久化通过记录服务器接收到的每个写命令来实现数据的持久化。这样，在 Redis 重启时，可以通过重放这些命令来恢复数据。

2 AOF相关配置

appendonly yes: 将这一行的值从 no 改为 yes，以启用 AOF 持久化。
appendfilename "appendonly.aof": 设置 AOF 文件的名称，默认是 appendonly.aof。
appendfsync <policy> 设置 fsync 策略，可以选择always、everysec、 no。推荐everyse因为它提供了较好的性能和数据安全性之间的平衡。
- always: 每次写操作后都进行 fsync，最安全但性能较差。
- everysec: 每秒进行一次 fsync，性能较好且数据安全性较高。
- no: 不主动进行 fsync，由操作系统决定何时同步到磁盘，性能最好但数据安全性最低。

3 aof 文件格式

Redis 的 AOF（Append-Only File）文件格式是一种文本格式，记录了 Redis 服务器接收到的所有写命令。AOF 文件中的每条记录都是一个完整的 Redis 命令，可以直接复制到 Redis 客户端中执行。

格式如下：

*<参数数量> CR LF
$<参数1的长度> CR LF
<参数1数据> CR LF
$<参数2的长度> CR LF
<参数2数据> CR LF
...

* 表示参数的数量。
$ 表示参数的长度。
CR 是回车符（ASCII 13），LF 是换行符（ASCII 10）。

例如：

*2
$6
SELECT
$1
0
*3
$3
SET
$5
liyan
$8
liyan223

为执行了2个命令：

SELECT 0
SET liyan liyan223

4 核心数据结构

redisCommand

用于表示一个 Redis 命令。这个结构体包含了命令的各种属性和元数据，使得 Redis 能够正确地解析、执行和管理这些命令。

struct redisCommand {
    //命令的名称，例如 "SET" 或 "GET"。
    char *name;  
    /*
    指向实际处理该命令的函数。这个函数会执行命令的具体逻辑。
	示例: 对于 SET 命令，proc 会指向 setCommand 函数。
    */
    redisCommandProc *proc;
    /*
    命令所需的参数个数。如果命令可以接受可变数量的参数，通常设置为 -N，其中 N 是最小参数个数。
	示例: SET 命令的 arity 为 3，因为 SET key value [EX seconds|PX milliseconds] [NX|XX] 至少需要三个参数（键、值和其他可选参数）。
    */
    int arity;
    /*
    命令标志的字符串表示形式，每个字符代表一个标志。常见的标志包括：
    'w': 写操作
    'r': 读操作
    'm': 可以修改多个键
    'a': 管理命令
    'f': 快速执行
    's': 允许在从服务器上执行
    'R': 随机命令
    'S': 排序命令
    'l': 加载时使用
    't': 不跟踪时间
    'M': 使用模块
    'k': 键空间通知
    'g': 读取全局状态
    'F': 不记录 AOF
    'C': 集群模式下不支持
    'A': ACL 控制
    SET 命令的 sflags 可能是 "wm"，表示写操作且可以修改多个键。
    */
    char *sflags;   /* Flags as string representation, one char per flag. */
    /*
    从 sflags 字符串中解析出的实际标志位。每个标志位对应一个比特位。
	示例: 如果 sflags 是 "wm"，则 flags 将包含 CMD_WRITE 和 CMD_DENYOOM 标志。
    */
    uint64_t flags; /* The actual flags, obtained from the 'sflags' field. */
    /* Use a function to determine keys arguments in a command line.
     * Used for Redis Cluster redirect. */
    /*
    用于确定命令行中的键参数的函数。主要用于 Redis Cluster 中的重定向。
	示例: 对于 MGET 命令，getkeys_proc 会指向 getKeysFromMultiKeyCommand 函数。
    */
    redisGetKeysProc *getkeys_proc;
    /* What keys should be loaded in background when calling this command? */
    /*
     第一个键参数的位置（从 0 开始计数）。如果命令没有键参数，则为 0。
	示例: SET 命令的 firstkey 为 1，因为第一个参数是键名。
    */
    int firstkey; /* The first argument that's a key (0 = no keys) */
    /*
    最后一个键参数的位置（从 0 开始计数）。如果命令没有键参数，则与 firstkey 相同。
	示例: MGET 命令的 lastkey 为 -1，表示最后一个参数也是键名
    */
    int lastkey;  /* The last argument that's a key */
    /*
     键参数之间的步长。如果命令只涉及连续的键参数，则为 1。
	示例: MGET 命令的 keystep 为 1，因为所有参数都是键名。
    */
    int keystep;  /* The step between first and last key */
    /*
    microseconds: 执行该命令所花费的总微秒数。
    calls: 该命令被调用的次数。
    rejected_calls: 该命令被拒绝的次数（例如，由于权限问题）。
    failed_calls: 该命令执行失败的次数。
    */
    long long microseconds, calls, rejected_calls, failed_calls;
    /*
    命令的 ID。这是一个从 0 开始递增的 ID，用于检查访问控制列表 (ACL)。连接能够执行给定命令的前提是连接关联的用户在允许命令的位图中有此命令的位。
    */
    int id;     /* Command ID. This is a progressive ID starting from 0 that
                   is assigned at runtime, and is used in order to check
                   ACLs. A connection is able to execute a given command if
                   the user associated to the connection has this command
                   bit set in the bitmap of allowed commands. */
};

假设有一个 SET 命令的定义如下：

/*
name 为 "set"
proc 指向 setCommandImpl 函数
arity 为 3，表示 SET 命令至少需要三个参数
sflags 为 "wm"，表示写操作且可以修改多个键
flags 由 sflags 解析得到
getkeys_proc 指向 getKeysFromCommand 函数
firstkey 和 lastkey 均为 1，表示第一个参数是键名
keystep 为 1，表示键参数之间没有间隔
统计信息初始为 0
id 初始为 0
*/
struct redisCommand setCommand = {
    "set",
    setCommandImpl,
    3,
    "wm",
    CMD_WRITE | CMD_DENYOOM,
    getKeysFromCommand,
    1,  // firstkey
    1,  // lastkey
    1,  // keystep
    0,  // microseconds
    0,  // calls
    0,  // rejected_calls
    0,  // failed_calls
    0   // id
};

5 核心代码

startAppendOnly

启用 AOF 持久化，并打开或创建 AOF 文件。

startAppendOnly 函数的主要任务是初始化 AOF 文件，处理可能的后台操作冲突，并设置 AOF 状态以便开始记录写命令。该函数确保 AOF 文件正确打开，并在必要时触发 AOF 重写。如果在过程中遇到任何错误，函数会记录日志并返回错误状态。

int startAppendOnly(void) {
    char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */
    int newfd;
	//打开或创建 AOF 文件
    /*
    使用 open 系统调用以只写、追加和创建模式打开 AOF 文件。
	如果文件无法打开（例如权限问题），则记录警告日志并返回错误。
	serverAssert 用于确保 AOF 状态为 AOF_OFF，即 AOF 尚未开启。
    */
    newfd = open(server.aof_filename,O_WRONLY|O_APPEND|O_CREAT,0644);
    serverAssert(server.aof_state == AOF_OFF);
    if (newfd == -1) {
        char *cwdp = getcwd(cwd,MAXPATHLEN);

        serverLog(LL_WARNING,
            "Redis needs to enable the AOF but can't open the "
            "append only file %s (in server root dir %s): %s",
            server.aof_filename,
            cwdp ? cwdp : "unknown",
            strerror(errno));
        return C_ERR;
    }
    // 处理后台进程.
    /*
    如果有其他后台操作且不是 AOF 重写，则标记 AOF 重写待执行。
	如果已经有 AOF 重写在进行中，则停止当前的 AOF 重写并启动新的 AOF 重写。
	调用 rewriteAppendOnlyFileBackground 触发后台 AOF 重写，如果失败则关闭文件描述符并记录错误。
    */
    if (hasActiveChildProcess() && server.child_type != CHILD_TYPE_AOF) {
        server.aof_rewrite_scheduled = 1;
        serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
    } else {
        /* If there is a pending AOF rewrite, we need to switch it off and
         * start a new one: the old one cannot be reused because it is not
         * accumulating the AOF buffer. */
        if (server.child_type == CHILD_TYPE_AOF) {
            serverLog(LL_WARNING,"AOF was enabled but there is already an AOF rewriting in background. Stopping background AOF and starting a rewrite now.");
            killAppendOnlyChild();
        }
        if (rewriteAppendOnlyFileBackground() == C_ERR) {
            close(newfd);
            serverLog(LL_WARNING,"Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.");
            return C_ERR;
        }
    }
    /* We correctly switched on AOF, now wait for the rewrite to be complete
     * in order to append data on disk. */
    /*
    设置 AOF 状态
    将 AOF 状态设置为 AOF_WAIT_REWRITE，表示等待 AOF 重写完成。
	更新最后一次 fsync 的时间戳。
	设置 AOF 文件描述符。
    */
    server.aof_state = AOF_WAIT_REWRITE;
    server.aof_last_fsync = server.unixtime;
    server.aof_fd = newfd;

    /* If AOF fsync error in bio job, we just ignore it and log the event. */
    //理之前的 AOF 错误
    int aof_bio_fsync_status;
    atomicGet(server.aof_bio_fsync_status, aof_bio_fsync_status);
    if (aof_bio_fsync_status == C_ERR) {
        serverLog(LL_WARNING,
            "AOF reopen, just ignore the AOF fsync error in bio job");
        atomicSet(server.aof_bio_fsync_status,C_OK);
    }

    /* If AOF was in error state, we just ignore it and log the event. */
    if (server.aof_last_write_status == C_ERR) {
        serverLog(LL_WARNING,"AOF reopen, just ignore the last error.");
        server.aof_last_write_status = C_OK;
    }
    return C_OK;
}

feedAppendOnlyFile

feedAppendOnlyFile 函数是 Redis 中用于将写命令追加到 AOF 缓冲区的关键函数。这个函数确保所有对数据库的修改操作都被记录到 AOF 文件中，以便在服务器重启时能够恢复数据。

void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
    sds buf = sdsempty();
    /* The DB this command was targeting is not the same as the last command
     * we appended. To issue a SELECT command is needed. */
    /*
    如果当前命令的目标数据库与上一个命令不同，则需要插入一个 SELECT 命令来切换数据库。
	使用 snprintf 和 sdscatprintf 将 SELECT 命令格式化并追加到 buf 中。
	更新 server.aof_selected_db 以反映当前选中的数据库。
	server.aof_selected_db 为文件中没有加载数据库
    */
    if (dictid != server.aof_selected_db) {
        char seldb[64];

        snprintf(seldb,sizeof(seldb),"%d",dictid);
        buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
            (unsigned long)strlen(seldb),seldb);
        server.aof_selected_db = dictid;
    }
	//处理过期命令
    if (cmd->proc == expireCommand || cmd->proc == pexpireCommand ||
        cmd->proc == expireatCommand) {
        /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */
        buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
    //处理包含过期时间的SET命令    
    } else if (cmd->proc == setCommand && argc > 3) {
        robj *pxarg = NULL;
        /* When SET is used with EX/PX argument setGenericCommand propagates them with PX millisecond argument.
         * So since the command arguments are re-written there, we can rely here on the index of PX being 3. */
        if (!strcasecmp(argv[3]->ptr, "px")) {
            pxarg = argv[4];
        }
        /* For AOF we convert SET key value relative time in milliseconds to SET key value absolute time in
         * millisecond. Whenever the condition is true it implies that original SET has been transformed
         * to SET PX with millisecond time argument so we do not need to worry about unit here.*/
        if (pxarg) {
            robj *millisecond = getDecodedObject(pxarg);
            long long when = strtoll(millisecond->ptr,NULL,10);
            when += mstime();

            decrRefCount(millisecond);

            robj *newargs[5];
            newargs[0] = argv[0];
            newargs[1] = argv[1];
            newargs[2] = argv[2];
            newargs[3] = shared.pxat;
            newargs[4] = createStringObjectFromLongLong(when);
            buf = catAppendOnlyGenericCommand(buf,5,newargs);
            decrRefCount(newargs[4]);
        } else {
            buf = catAppendOnlyGenericCommand(buf,argc,argv);
        }
      //其他命令
    } else {
        /* All the other commands don't need translation or need the
         * same translation already operated in the command vector
         * for the replication itself. */
        buf = catAppendOnlyGenericCommand(buf,argc,argv);
    }

    /* Append to the AOF buffer. This will be flushed on disk just before
     * of re-entering the event loop, so before the client will get a
     * positive reply about the operation performed. */
    //追加到 AOF 缓冲区
    if (server.aof_state == AOF_ON)
        server.aof_buf = sdscatlen(server.aof_buf,buf,sdslen(buf));

    /* If a background append only file rewriting is in progress we want to
     * accumulate the differences between the child DB and the current one
     * in a buffer, so that when the child process will do its work we
     * can append the differences to the new append only file. */
    if (server.child_type == CHILD_TYPE_AOF)
        aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
	//释放 buf 占用的内存。
    sdsfree(buf);
}

feedAppendOnlyFile 函数的主要任务是将执行的写命令追加到 AOF 缓冲区中，以便稍后写入磁盘。该函数还处理了数据库选择命令、过期命令和 SET 命令的特殊情况，确保 AOF 文件中的命令能够正确地反映实际的操作。此外，如果后台正在进行 AOF 重写，它还会将命令追加到重写缓冲区中，以便在重写完成后应用这些新命令。

rewriteAppendOnlyFileBackground

rewriteAppendOnlyFileBackground 是 Redis 中用于在后台执行 AOF 重写的函数。AOF 重写是为了减小 AOF 文件的大小，通过创建一个新的 AOF 文件，该文件只包含重建当前数据集所需的最少命令。这个过程是在后台进行的，以避免阻塞主进程。

int rewriteAppendOnlyFileBackground(void) {
    pid_t childpid;
	//检查是否有活动的子进程:
    if (hasActiveChildProcess()) return C_ERR;
    //创建管道 用于父子通信
    if (aofCreatePipes() != C_OK) return C_ERR;
    //使用 redisFork 创建一个子进程。CHILD_TYPE_AOF 表示这是一个 AOF 重写子进程
    if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
        //子线程逻辑
        /*
        设置进程标题为 "redis-aof-rewrite"。
		设置 CPU 亲和性，确保子进程运行在指定的 CPU 上。
		生成临时文件名 tmpfile。
		调用 rewriteAppendOnlyFile 进行 AOF 重写。
		如果重写成功，发送子进程信息并退出，返回状态 0；否则返回状态 1。
        */
        char tmpfile[256];

        /* Child */
        redisSetProcTitle("redis-aof-rewrite");
        redisSetCpuAffinity(server.aof_rewrite_cpulist);
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
        if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
            sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
            exitFromChild(0);
        } else {
            exitFromChild(1);
        }
    } else {
        /* Parent */
        if (childpid == -1) {
            serverLog(LL_WARNING,
                "Can't rewrite append only file in background: fork: %s",
                strerror(errno));
            aofClosePipes();
            return C_ERR;
        }
        serverLog(LL_NOTICE,
            "Background append only file rewriting started by pid %ld",(long) childpid);
        server.aof_rewrite_scheduled = 0;
        server.aof_rewrite_time_start = time(NULL);

        /* We set appendseldb to -1 in order to force the next call to the
         * feedAppendOnlyFile() to issue a SELECT command, so the differences
         * accumulated by the parent into server.aof_rewrite_buf will start
         * with a SELECT statement and it will be safe to merge. */
        server.aof_selected_db = -1;
        replicationScriptCacheFlush();
        return C_OK;
    }
    return C_OK; /* unreached */
}

rewriteAppendOnlyFile 函数是 Redis 中用于重写 AOF 文件的核心函数。该函数负责将当前内存中的数据集以最小的命令集合形式写入一个新的 AOF 文件中。这个过程通常在后台进行，以避免阻塞主进程。

int rewriteAppendOnlyFile(char *filename) {
    rio aof;
    FILE *fp = NULL;
    char tmpfile[256];
    char byte;

    /* Note that we have to use a different temp name here compared to the
     * one used by rewriteAppendOnlyFileBackground() function. */
    /*
    生成一个临时文件名，并尝试打开它。如果打开失败，记录警告日志并返回错误。
    */
    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
    fp = fopen(tmpfile,"w");
    if (!fp) {
        serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
        return C_ERR;
    }
	//初始化 AOF 缓冲区和 Rio（Redis I/O）结构体。
    server.aof_child_diff = sdsempty();
    rioInitWithFile(&aof,fp);
	// 如果配置了增量同步，则设置 Rio 的自动同步大小。
    if (server.aof_rewrite_incremental_fsync)
        rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
	//开始保存操作
    startSaving(RDBFLAGS_AOF_PREAMBLE);
	/*
	如果启用了 RDB 前言（preamble），则使用 rdbSaveRio 将数据集保存为 RDB 格式。
	否则，使用 rewriteAppendOnlyFileRio 逐个键值对地重写 AOF 文件。
	*/
    if (server.aof_use_rdb_preamble) {
        int error;
        if (rdbSaveRio(&aof,&error,RDBFLAGS_AOF_PREAMBLE,NULL) == C_ERR) {
            errno = error;
            goto werr;
        }
    } else {
        if (rewriteAppendOnlyFileRio(&aof) == C_ERR) goto werr;
    }

    /* Do an initial slow fsync here while the parent is still sending
     * data, in order to make the next final fsync faster. */
    //初步同步文件，确保数据被写入磁盘。
    if (fflush(fp) == EOF) goto werr;
    if (fsync(fileno(fp)) == -1) goto werr;

    /* Read again a few times to get more data from the parent.
     * We can't read forever (the server may receive data from clients
     * faster than it is able to send data to the child), so we try to read
     * some more data in a loop as soon as there is a good chance more data
     * will come. If it looks like we are wasting time, we abort (this
     * happens after 20 ms without new data). */
    int nodata = 0;
    mstime_t start = mstime();
    while(mstime()-start < 1000 && nodata < 20) {
        if (aeWait(server.aof_pipe_read_data_from_parent, AE_READABLE, 1) <= 0)
        {
            nodata++;
            continue;
        }
        nodata = 0; /* Start counting from zero, we stop on N *contiguous*
                       timeouts. */
        aofReadDiffFromParent();
    }

    /* Ask the master to stop sending diffs. */
    if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr;
    if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK)
        goto werr;
    /* We read the ACK from the server using a 5 seconds timeout. Normally
     * it should reply ASAP, but just in case we lose its reply, we are sure
     * the child will eventually get terminated. */
    if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 ||
        byte != '!') goto werr;
    serverLog(LL_NOTICE,"Parent agreed to stop sending diffs. Finalizing AOF...");

    /* Read the final diff if any. */
    aofReadDiffFromParent();

    /* Write the received diff to the file. */
    serverLog(LL_NOTICE,
        "Concatenating %.2f MB of AOF diff received from parent.",
        (double) sdslen(server.aof_child_diff) / (1024*1024));

    /* Now we write the entire AOF buffer we received from the parent
     * via the pipe during the life of this fork child.
     * once a second, we'll take a break and send updated COW info to the parent */
    size_t bytes_to_write = sdslen(server.aof_child_diff);
    const char *buf = server.aof_child_diff;
    long long cow_updated_time = mstime();
    long long key_count = dbTotalServerKeyCount();
    while (bytes_to_write) {
        /* We write the AOF buffer in chunk of 8MB so that we can check the time in between them */
        size_t chunk_size = bytes_to_write < (8<<20) ? bytes_to_write : (8<<20);

        if (rioWrite(&aof,buf,chunk_size) == 0)
            goto werr;

        bytes_to_write -= chunk_size;
        buf += chunk_size;

        /* Update COW info */
        long long now = mstime();
        if (now - cow_updated_time >= 1000) {
            sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
            cow_updated_time = now;
        }
    }

    /* Make sure data will not remain on the OS's output buffers */
    if (fflush(fp)) goto werr;
    if (fsync(fileno(fp))) goto werr;
    if (fclose(fp)) { fp = NULL; goto werr; }
    fp = NULL;

    /* Use RENAME to make sure the DB file is changed atomically only
     * if the generate DB file is ok. */
    if (rename(tmpfile,filename) == -1) {
        serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
        unlink(tmpfile);
        stopSaving(0);
        return C_ERR;
    }
    serverLog(LL_NOTICE,"SYNC append only file rewrite performed");
    stopSaving(1);
    return C_OK;

werr:
    serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
    if (fp) fclose(fp);
    unlink(tmpfile);
    stopSaving(0);
    return C_ERR;
}

rewriteAppendOnlyFileRio 函数是 Redis 中用于将当前数据库中的所有键值对重写到 AOF 文件的核心函数。该函数通过遍历每个数据库（db）中的键值对，并生成相应的 Redis 命令，将其写入 rio 结构体中。

int rewriteAppendOnlyFileRio(rio *aof) {
    dictIterator *di = NULL;
    dictEntry *de;
    size_t processed = 0;
    int j;
    long key_count = 0;
    long long updated_time = 0;
	//遍历每个数据库
    for (j = 0; j < server.dbnum; j++) {
        /*
        获取当前数据库。
        如果数据库为空，则跳过。
        获取字典迭代器。
        */
        char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
        redisDb *db = server.db+j;
        dict *d = db->dict;
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);

        /* SELECT the new DB */
        /*
        发送 SELECT 命令来选择当前数据库。
		写入数据库编号。
        */
        if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
        if (rioWriteBulkLongLong(aof,j) == 0) goto werr;

        /* Iterate this DB writing every entry */
        //遍历数据库中的每个键值对
        while((de = dictNext(di)) != NULL) {
            sds keystr;
            robj key, *o;
            long long expiretime;
			/*
			
			获取键和对应的值。
			初始化静态字符串对象 key。
			获取键的过期时间。
			*/
            keystr = dictGetKey(de);
            o = dictGetVal(de);
            initStaticStringObject(key,keystr);

            expiretime = getExpire(db,&key);

            /* Save the key and associated value */
            // 保存键值对
            if (o->type == OBJ_STRING) {
                /* Emit a SET command */
                char cmd[]="*3\r\n$3\r\nSET\r\n";
                if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                /* Key and value */
                if (rioWriteBulkObject(aof,&key) == 0) goto werr;
                if (rioWriteBulkObject(aof,o) == 0) goto werr;
            } else if (o->type == OBJ_LIST) {
                if (rewriteListObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_SET) {
                if (rewriteSetObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_ZSET) {
                if (rewriteSortedSetObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_HASH) {
                if (rewriteHashObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_STREAM) {
                if (rewriteStreamObject(aof,&key,o) == 0) goto werr;
            } else if (o->type == OBJ_MODULE) {
                if (rewriteModuleObject(aof,&key,o) == 0) goto werr;
            } else {
                serverPanic("Unknown object type");
            }
            /* Save the expire time */
            // 如果键有设置过期时间，则发送 PEXPIREAT 命令。
            if (expiretime != -1) {
                char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
                if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                if (rioWriteBulkObject(aof,&key) == 0) goto werr;
                if (rioWriteBulkLongLong(aof,expiretime) == 0) goto werr;
            }
            /* Read some diff from the parent process from time to time. */
            // 定期从父进程读取数据差异，以确保子进程不会遗漏新的写操作。
            if (aof->processed_bytes > processed+AOF_READ_DIFF_INTERVAL_BYTES) {
                processed = aof->processed_bytes;
                aofReadDiffFromParent();
            }

            /* Update info every 1 second (approximately).
             * in order to avoid calling mstime() on each iteration, we will
             * check the diff every 1024 keys */
            if ((key_count++ & 1023) == 0) {
                long long now = mstime();
                if (now - updated_time >= 1000) {
                    sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
                    updated_time = now;
                }
            }
        }
        dictReleaseIterator(di);
        di = NULL;
    }
    return C_OK;

werr:
    if (di) dictReleaseIterator(di);
    return C_ERR;
}