redis源码阅读-持久化之aof详解

aof相关配置

aof-rewrite-incremental-fsync yes
# aof 开关，默认是关闭的，改为yes表示开启
appendonly no
# aof的文件名，默认
appendfilename "appendonly.aof"
# aof刷数据的策略,有no/everysec/aways
appendfsync everysec
no-appendfsync-on-rewrite no
# aof超出配置大小的比例，模式是100%，可以理解为阈值 
auto-aof-rewrite-percentage 100
# aof 配置的文件的大小，默认64mb
auto-aof-rewrite-min-size 64mb
aof-load-truncated yes
# rewrite 进行时候，rewrite 文件分两种格式，1. 先 用 rdb 序列化，序列化结果写入aof文件，然后期间积累的差异用追加aof命令格式 ，2 整个文件都是aof的命令追加格式
aof-use-rdb-preamble yes

appendfsync 一共有3种策略

alays 主要有数据改动就把数据刷入磁盘，性能相对最差，但最安全
everysec 每隔1秒刷一次数据，redis默认的，也是redis推荐的
no 不主动刷，什么时候刷数据，取决于操作系统，大多数linux 30秒提交一次

aof写入：

在之前分析redis的流程的时候《redis源码阅读三-终于把主线任务执行搞明白了》里读取客户端信息的时候

在processCommand 函数里，解析出来执行命令，放入了client中

int processCommand(client *c) {
    //解析出来命令
    c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
    //CMD_CALL_FULL 包含着CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_PROPAGATE
    //CMD_CALL_PROPAGATE 包含着CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL
    call(c,CMD_CALL_FULL);
}
//命令执行
void call(client *c, int flags) {
    /**
     * dirty 记录修改次数
     * start记录命令开始执行时间us
     * duration记录命令执行花费时间
     */
    long long dirty;
    ustime_t start, duration;
    int client_old_flags = c->flags;
    struct redisCommand *real_cmd = c->cmd;
    //从server.dirty中获取
    dirty = server.dirty;
    //命令执行，会执行对应的redisCommand
    c->cmd->proc(c);
    //计算执行时间
    duration = ustime()-start;
    //在对应的命令里执行一次，dirty+1
    //这里表示本次命令有多少次修改
    dirty = server.dirty-dirty;
    if (flags & CMD_CALL_PROPAGATE && (c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP){
        if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
            propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags);
    }    
}
/**
 * 传播执行的命令到aof或从节点
 * @param cmd
 * @param dbid
 * @param argv
 * @param argc
 * @param flags
 */
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
               int flags)
{
    if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF)
        //aof追加（主要是生成aof内容，并追加到.aof_buf）
        feedAppendOnlyFile(cmd,dbid,argv,argc);
    if (flags & PROPAGATE_REPL)
        //主从复制处理
        replicationFeedSlaves(server.slaves,dbid,argv,argc);
}

在feedAppendOnlyFile方法里主要是aof内容生成，方法就不具体列了，主要做了三件事

组装刚执行命令的aof内容buf，将过期时间由相对转成绝对（重点）
如果AOF开启的情况，将刚组装的buf放入到server.aof_buf 后
如果正在重写aof，将buf写到 server.aof_rewrite_buf_blocks中（在aofRewriteBufferAppend里）

/**
 * aof重写的情况下调用，将buf写到一个链表里
 * @param s  aof信息
 * @param len
 */
void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
    listNode *ln = listLast(server.aof_rewrite_buf_blocks);
    aofrwblock *block = ln ? ln->value : NULL;
    while(len) {
        //将传递过来的s写入到aof_rewrite_buf_blocks
    }
    //现在是主进程，如果有管道，通过管道server.aof_pipe_write_data_to_child 给子进程
    if (aeGetFileEvents(server.el,server.aof_pipe_write_data_to_child) == 0) {
        aeCreateFileEvent(server.el, server.aof_pipe_write_data_to_child,AE_WRITABLE, aofChildWriteDiffData, NULL);
}
/**
 * 将server.aof_rewrite_buf_blocks 中的数据写入server.aof_pipe_write_data_to_child子进程的管道里
 * 此处增量更新的数据，会由子进程在将虚拟空间里的数据落地后再次读取
 * @param el
 * @param fd
 * @param privdata
 * @param mask
 */
void aofChildWriteDiffData(aeEventLoop *el, int fd, void *privdata, int mask) {
     while(1) {
          ln = listFirst(server.aof_rewrite_buf_blocks);
          block = ln ? ln->value : NULL;
          if (block->used > 0) {
              //通过管道把block->buf里的数据写给子进程
              nwritten = write(server.aof_pipe_write_data_to_child, block->buf,block->used);
              if (nwritten <= 0) return;
              memmove(block->buf,block->buf+nwritten,block->used-nwritten);
              block->used -= nwritten;
              block->free += nwritten;
          }
          if (block->used == 0) listDelNode(server.aof_rewrite_buf_blocks,ln);
     }
}

这里有几个变量

server.aof_buf aof缓冲区，用于存放每次执行命令后的aof信息
server.aof_rewrite_buf_blocks 只要是有aof的子进程，就把新产生的命令添加上去
server.aof_pipe_write_data_to_child aof 子进程的管道，用于将server.aof_rewrite_buf_blocks信息给子进程

在主进程执行命令后，并没有写aof文件，只是将命令拼装成了字符串，放入到了aof缓冲区server.aof_buf 中，如果有aof的子进程，将aof信息放入到server.aof_rewrite_buf_blocks 然后通过管道将该信息给到子进程。

那aof是什么时候写入到文件里呢？，别急，看下图

aof有几个场景的写入：

在主流程的循环体里
- 在循环执行前的beforesleep里
- 在serverCron里
准备停止之前调用一次
通过configSetCommand设置
主从复制

最终都是调用的flushAppendOnlyFile

//在beforeSleep里，没有任何的逻辑判断，直接调用
void beforeSleep(struct aeEventLoop *eventLoop) {
    //将aof缓冲区写入磁盘
    flushAppendOnlyFile(0);    
}
//在serverCron中
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    //设置了刷入延期开始时间，直接调用
    if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);
    //每秒再判断一次，如果上一次执行异常，再刷入一次
    run_with_period(1000) {
        if (server.aof_last_write_status == C_ERR)
            flushAppendOnlyFile(0);
    }
}

/**
  * 刷入文件
  * @param force 1 强制，0不强制
  */
void flushAppendOnlyFile(int force) {
    //每秒刷入的时候，判断一下aof的fsync是否在bio线程里执行了，是sync_in_progress 为true
    if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
        sync_in_progress = aofFsyncInProgress();
    //每秒并且非强制刷入
    if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
        if (sync_in_progress) {
            //刷入延期开始时间为0，表示没有任何的执行
            if (server.aof_flush_postponed_start == 0) {
                //先将刷入延期开始时间置为当前时间
                server.aof_flush_postponed_start = server.unixtime;
                return;
                //以后再来，只要当前时间-开始时间<2直接返回（所以对redis的aof来说，并不是每秒执行，而是至少2秒以上，如果阻塞了，时间更久）
            } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
                return;
            }
            server.aof_delayed_fsync++;
        }
    }
    //这里用了卫语句的方式,将不能触发的在前面拦截
    //写入到aof文件里
    nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
    //写完aof后，将刷入延期开始时间置为0
    server.aof_flush_postponed_start = 0;
    //省略异常的处理
    //记录aof当前的大小
    server.aof_current_size += nwritten;
     /**
     * aof的可用buf比较小了，就清空buf
     */
    if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
        sdsclear(server.aof_buf);
    } else {
        sdsfree(server.aof_buf);
        server.aof_buf = sdsempty();
    }
}

//aof写入方式的枚举
configEnum aof_fsync_enum[] = {
    {"everysec", AOF_FSYNC_EVERYSEC},
    {"always", AOF_FSYNC_ALWAYS},
    {"no", AOF_FSYNC_NO},
    {NULL, 0}
};

aof的写入是在beforeSleep 里，在serverCron 主要是处理延期写入或者处理写入异常
aof通过server.aof_flush_postponed_start来延期写入，第一次将此值赋值为当前时间，写完以后置为0
aof通过write写入文件（获取的是文件fd对应的偏移指针，顺序写）

aof重写

在上一篇《redis源码阅读-持久化之RDB》中

rdb

serverCron在执行的时候，有一个backgroundRewriteDoneHandler方法，当时就有疑问，这个重写是干啥的？

然后搜索了下，发现了这个函数rewriteAppendOnlyFileBackground

我们看下这个函数的触发的时机

先说下触发时机：

在周期性循环里进行
通过命令调用执行aof重写
- 一个是bgrewriteaofCommand
- 一个是configSetCommand
主从复制时触发

在serverCron里有两次调用

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    /**调用①，这次调用是之前有rdb进程的时候，设置为了aof重写等待任务（本该执行的逻辑之前因为rdb没有执行，所以直接触发）*/
    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 && server.aof_rewrite_scheduled){
        rewriteAppendOnlyFileBackground();
    }
    if (server.rdb_child_pid != -1 || server.aof_child_pid != -1 ||ldbPendingChildren()){

    }else {

        /** 调用②
         *  在没有rdb和aof子进程的时候，
         *  先判断rdb是否达到了执行条件
         *     save 900 1  配置的判断
         *  再判断aof是否达到了执行条件
         *    1，先判断aof是否打开
         *    2，如果有rdb或aof子进程了也不再执行（互斥）
         *    3，有触发阈值，
         *    4，aof当前的文件的大小大于设置的最小大小
         */
        if (server.aof_state == AOF_ON && server.rdb_child_pid == -1 && server.aof_child_pid == -1 && server.aof_rewrite_perc &&
            server.aof_current_size > server.aof_rewrite_min_size)
        {
            /**
             * aof_rewrite_base_size 启动默认为0，第一次达到min_size以后设置为1，这时候触发
             * 等待aof重写完以后，会把aof_rewrite_base_size设置为重写后的大小，这个值一致随着aof文件大小自增
             */
            long long base = server.aof_rewrite_base_size ?server.aof_rewrite_base_size : 1;
            /**
              * 这个公式也很有意思，aof的重写会随着文件的变大而逐步变大，而不是64mb
              */
            long long growth = (server.aof_current_size*100/base) - 100;
            if (growth >= server.aof_rewrite_perc) {
                //触发aof重写
                rewriteAppendOnlyFileBackground();
            }
        }
    }
}

① 位置标注的调用是为了延迟补偿
②位置标注的每次在没有rdb和aof子进程的情况下都会进来
redis通过server.aof_rewrite_scheduled来控制aof的延迟执行

/**
 * 后台的方式重写aof，通过fork进程处理
 *
 * @return
 */
int rewriteAppendOnlyFileBackground(void) {
    pid_t childpid;
    long long start;
    //双重校验，只要有aof或rdb的子进程就不再执行，
    if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
    //创建6个管道，分别用来数据传输，父子进程之间的ack响应
    if (aofCreatePipes() != C_OK) return C_ERR;
    //创建执行aof完以后回写的管道
    openChildInfoPipe();
    start = ustime();
    //fork子进程进行重写
    if ((childpid = fork()) == 0) {
        char tmpfile[256];

        /* Child */
        //关闭自己不需要关注的
        closeClildUnusedResourceAfterFork();
        //设置进程名称
        redisSetProcTitle("redis-aof-rewrite");
        //格式化重写aof的文件名
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
        //重写aof文件
        if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
            size_t private_dirty = zmalloc_get_private_dirty(-1);
            server.child_info_data.cow_size = private_dirty;
            sendChildInfo(CHILD_INFO_TYPE_AOF);
            exitFromChild(0);
        } else {
            exitFromChild(1);
        }
    } else {
        /* Parent */
        server.stat_fork_time = ustime()-start;
        server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
        latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
        if (childpid == -1) {
            closeChildInfoPipe();
            aofClosePipes();
            return C_ERR;
        }
        //已经开始处理了，就把等待任务表示写回0
        server.aof_rewrite_scheduled = 0;
        //标记aof重写开始的时间
        server.aof_rewrite_time_start = time(NULL);
        //标记aof的的子进程
        server.aof_child_pid = childpid;
        //禁用hash resize
        updateDictResizePolicy();
        server.aof_selected_db = -1;
        replicationScriptCacheFlush();
        return C_OK;
    }
    return C_OK; /* unreached */
}

/**
 * 创建aof管道
 * @return
 */
int aofCreatePipes(void) {
    int fds[6] = {-1, -1, -1, -1, -1, -1};
    int j;

    if (pipe(fds) == -1) goto error; /* parent -> children data. */
    if (pipe(fds+2) == -1) goto error; /* children -> parent ack. */
    if (pipe(fds+4) == -1) goto error; /* parent -> children ack. */
    /* 设置数据管道为非阻塞 */
    if (anetNonBlock(NULL,fds[0]) != ANET_OK) goto error;
    if (anetNonBlock(NULL,fds[1]) != ANET_OK) goto error;
    /**
     * 注册时间处理，监听 server.aof_pipe_read_ack_from_child 传递过来的ack信息，如果有就 server.aof_stop_sending_diff = 1;
     * 同时删除server.aof_pipe_read_ack_from_child的事件处理器
     */
    if (aeCreateFileEvent(server.el, fds[2], AE_READABLE, aofChildPipeReadable, NULL) == AE_ERR) goto error;
    //父进程往子进程写数据的管道
    server.aof_pipe_write_data_to_child = fds[1];
    //子进程从父进程读数据的管道
    server.aof_pipe_read_data_from_parent = fds[0];
    //子进程发送ack给父进程的管道
    server.aof_pipe_write_ack_to_parent = fds[3];
    //父进程读子进程发送的ack信息的管道
    server.aof_pipe_read_ack_from_child = fds[2];
    //父进程发送ack信息给子进程的管道
    server.aof_pipe_write_ack_to_child = fds[5];
    //子进程读取父进程发送的ack信息的管道
    server.aof_pipe_read_ack_from_parent = fds[4];
    server.aof_stop_sending_diff = 0;
    return C_OK;

error:
    for (j = 0; j < 6; j++) if(fds[j] != -1) close(fds[j]);
    return C_ERR;
}

rof重写

/**
 * 同步重写aof
 * @param filename
 * @return
 */
int rewriteAppendOnlyFile(char *filename) {
    rio aof;
    FILE *fp;
    char tmpfile[256];
    char byte;

    //又创建了一个临时文件
    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
    fp = fopen(tmpfile,"w");
    if (!fp) {
        return C_ERR;
    }

    server.aof_child_diff = sdsempty();
    //初始化一个文件io对象
    rioInitWithFile(&aof,fp);

    if (server.aof_rewrite_incremental_fsync)
        rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
    /**
     * aof_use_rdb_preamble = yes 表示开启了混合模式，先将rdb信息刷入aof文件
     * rdb是比较节省空间
     */
    if (server.aof_use_rdb_preamble) {
        int error;
        if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) {
            errno = error;
            goto werr;
        }
    } else {
        //直接以redis的协议从所有db的全局hash表中读取数据写入aof文件
        if (rewriteAppendOnlyFileRio(&aof) == C_ERR) goto werr;
    }
    //清空缓冲区
    if (fflush(fp) == EOF) goto werr;
    //刷入磁盘，正常都是在操作系统的页缓存里，fsync可以刷入磁盘
    if (fsync(fileno(fp)) == -1) goto werr;

    int nodata = 0;
    mstime_t start = mstime();
    /**
     * 在1秒内，每次等待1毫秒，从管道里获取数据
     */
    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. */
    //向父进程发送ack信息!,让父进程停止追加
    if (write(server.aof_pipe_write_ack_to_parent,"!",1) != 1) goto werr;
    //读取master的ack信息
    if (anetNonBlock(NULL,server.aof_pipe_read_ack_from_parent) != ANET_OK)
        goto werr;
    //同步读取父进程的ack消息，5秒超时
    if (syncRead(server.aof_pipe_read_ack_from_parent,&byte,1,5000) != 1 ||
        byte != '!') goto werr;

    /* Read the final diff if any. */
    //再次读一下，补救ack期间传过来的数据
    aofReadDiffFromParent();

    //将aof开始执行之后，追加的数据写入到aof文件中
    if (rioWrite(&aof,server.aof_child_diff,sdslen(server.aof_child_diff)) == 0)
        goto werr;

    //清空缓冲区
    if (fflush(fp) == EOF) goto werr;
    //刷入磁盘，正常都是在操作系统的页缓存里，fsync可以刷入磁盘
    if (fsync(fileno(fp)) == -1) goto werr;
    if (fclose(fp) == EOF) goto werr;

    //将rof的临时文件改名为temp-rewriteaof-bg-%d.aof
    if (rename(tmpfile,filename) == -1) {
        unlink(tmpfile);
        return C_ERR;
    }
    return C_OK;

werr:
    fclose(fp);
    unlink(tmpfile);
    return C_ERR;
}

通过以上的代码我们可以看到：

aof的重写过程和bgsave方式的rdb差不多的过程差不多
aof 重写有两种模式，一种是先写rdb，再追加aof（混合模式），一种是一路aof格式
aof在将虚拟内存空间里的数据写完以后，会通过wait轮训从管道server.aof_pipe_read_data_from_parent里获取增量更新的数据，增量更新在第一小节的时候，有个aofChildWriteDiffData处理器（通过FileEvent机制写入）
然后aof 子父进程发送ack确认消息
最后再次通过rioWrite 将ack期间追加的数据补救回来
最后将rof的临时文件重命名为temp-rewriteaof-bg-进程id.aof

aof后续处理

其实到这里，aof是写到文件里了，但是并没有再改到配置的文件上?

别急，我们上一篇，或者本篇aof小结那张图。

在serverCron有一个分支处理backgroundSaveDoneHandler 和backgroundRewriteDoneHandler。上一篇，我们讲过了backgroundSaveDoneHandler，另一个方法backgroundRewriteDoneHandler没有讲，翻看代码看下

/**
 * 后台aof重写后的处理
 * @param exitcode 获取子进程的结束代码，exitcode=0 表示rdb成功，1表示失败
 * @param bysignal 子进程信号中断代码，如果有值，表示被信号中断
 */
void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
    if (!bysignal && exitcode == 0) {
        //格式化子进程的文件名
        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
            (int)server.aof_child_pid);
        //以写模式打开
        newfd = open(tmpfile,O_WRONLY|O_APPEND);
        //将写入缓冲区aof_rewrite_buf_blocks的数据追加到aof文件数上
        if (aofRewriteBufferWrite(newfd) == -1) {

        }
         //将临时文件改为真正的aof文件
        if (rename(tmpfile,server.aof_filename) == -1) {}
                if (server.aof_fd == -1) {
            close(newfd);
        } else {
            //把新的aof文件的fd赋值过去，redis可以实时写入，这些操作都是线性的
            oldfd = server.aof_fd;
            server.aof_fd = newfd;
            //根据类型是否直接刷盘
            if (server.aof_fsync == AOF_FSYNC_ALWAYS)
                redis_fsync(newfd);
            else if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
                aof_background_fsync(newfd);
            server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
            aofUpdateCurrentSize();
            //重置aof的base_size
            server.aof_rewrite_base_size = server.aof_current_size;
            //aof的偏移指针
            server.aof_fsync_offset = server.aof_current_size;

            /* Clear regular AOF buffer since its contents was just written to
             * the new AOF from the background rewrite buffer. */
            sdsfree(server.aof_buf);
            server.aof_buf = sdsempty();
        }
        //修改aof最近重写的状态
        server.aof_lastbgrewrite_status = C_OK;
    } else if (!bysignal && exitcode != 0) {
        server.aof_lastbgrewrite_status = C_ERR;
    }
}

在最后