redis源码学习持久化机制：AOF
Posted 看，未来
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了redis源码学习持久化机制：AOF相关的知识，希望对你有一定的参考价值。
AOF简介
上一篇讲的RDB是批量的插入数据，但是如果单纯使用就会有一个隐患：突然断电的时候势必要失去一部分数据，因为还没来得及轮上。
那AOF是个什么情况呢？AOF在redis运行期间，不断的将执行过的命令写入到文件中，如果redis被重启了，只要将这些命令重复执行一遍即可。
那如果单纯的使用AOF呢？占用内存就会很大了，毕竟保存一整条命令和单纯保存一块数据，孰轻孰重还是分得清的。
所以这时候怎么办？RDB做大批量的持久化，在下次RDB来之前，用AOF顶上。
此外，redis 还提供了AOF重写功能，如果一个键有多次写入，只保存最后一次写入命令。
这两个方法我们都能想到，不过是怎么实现的呢？
AOF不是不丢数据，但是由于它的时间间隔很小，所以丢失风险相较于RDB大大降低了。
AOF定时逻辑
redis 通过 serverCron 函数实现定时生成 aof 文件的任务，在上一篇RDB里面一并出现了，我就直接放相应部分源码吧：
...
/* Start a scheduled AOF rewrite if this was requested by the user while
* a BGSAVE was in progress. */
if (!hasActiveChildProcess() &&
server.aof_rewrite_scheduled)
rewriteAppendOnlyFileBackground();
/* Check if a background saving or AOF rewrite in progress terminated. */
if (hasActiveChildProcess() || ldbPendingChildren())
checkChildrenDone();
else
......
//执行AOF的条件：
/*
1、服务器开启了AOF功能
2、当前AOF文件大小大于aof_rewrite_min_size
3、对于上次重写后的AOF文件大小，当前AOF文件增加的空间大小所占比例已经超过了aof_rewrite_perc 配置
*/
if (server.aof_state == AOF_ON &&
!hasActiveChildProcess() &&
server.aof_rewrite_perc &&
server.aof_current_size > server.aof_rewrite_min_size)
long long base = server.aof_rewrite_base_size ?
server.aof_rewrite_base_size : 1;
long long growth = (server.aof_current_size*100/base) - 100;
if (growth >= server.aof_rewrite_perc)
serverLog(LL_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);
rewriteAppendOnlyFileBackground();
/* AOF postponed flush: Try at every cron cycle if the slow fsync
* completed. */
if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);
/* AOF write errors: in this case we have a buffer to flush as well and
* clear the AOF error in case of success to make the DB writable again,
* however to try every second is enough in case of 'hz' is set to
* an higher frequency. */
run_with_period(1000)
if (server.aof_last_write_status == C_ERR)
flushAppendOnlyFile(0);
/* Clear the paused clients flag if needed. */
clientsArePaused(); /* Don't check return value, just use the side effect.*/
/* Replication cron function -- used to reconnect to master,
* detect transfer failures, start background RDB transfers and so forth. */
run_with_period(1000) replicationCron();
/* Run the Redis Cluster cron. */
run_with_period(100)
if (server.cluster_enabled) clusterCron();
...
flushAppendOnlyFile 函数用于刷新AOF缓冲区的内容，写入文件中，只有在 AOF 缓冲区操作被延迟或者写入出错时才触发。
rewriteAppendOnlyFileBackground 函数负责重写AOF文件，在beforesleep函数中触发。
AOF 持久化过程
三部曲：
命令传播、
刷新AOF缓冲区、
同步磁盘
让我们一个一个来看：
命令传播
feedAppendOnlyFile 函数负责将命令传播到 AOF 缓冲区：
void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc)
sds buf = sdsempty();
robj *tmpargv[3];
/* The DB this command was targeting is not the same as the last command
* we appended. To issue a SELECT command is needed. */
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]);
else if (cmd->proc == setexCommand || cmd->proc == psetexCommand)
/* Translate SETEX/PSETEX to SET and PEXPIREAT */
tmpargv[0] = createStringObject("SET",3);
tmpargv[1] = argv[1];
tmpargv[2] = argv[3];
buf = catAppendOnlyGenericCommand(buf,3,tmpargv);
decrRefCount(tmpargv[0]);
buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]);
else if (cmd->proc == setCommand && argc > 3)
int i;
robj *exarg = NULL, *pxarg = NULL;
for (i = 3; i < argc; i ++)
if (!strcasecmp(argv[i]->ptr, "ex")) exarg = argv[i+1];
if (!strcasecmp(argv[i]->ptr, "px")) pxarg = argv[i+1];
serverAssert(!(exarg && pxarg));
if (exarg || pxarg)
/* Translate SET [EX seconds][PX milliseconds] to SET and PEXPIREAT */
buf = catAppendOnlyGenericCommand(buf,3,argv);
if (exarg)
buf = catAppendOnlyExpireAtCommand(buf,server.expireCommand,argv[1],
exarg);
if (pxarg)
buf = catAppendOnlyExpireAtCommand(buf,server.pexpireCommand,argv[1],
pxarg);
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. */
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.aof_child_pid != -1)
aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
sdsfree(buf);
对于一些带有过期时间的命令，需要进行转换。
对于其他命令，直接写入buf暂存区。
如果服务器开启AOF功能，则将buf暂存区内容写入AOF缓冲区。
如果当前存在AOF进程执行AOF重写操作，则将buf暂存区的数据写入AOF重写缓冲区。
刷新缓冲区 && 同步磁盘
void flushAppendOnlyFile(int force)
ssize_t nwritten;
int sync_in_progress = 0;
mstime_t latency;
if (sdslen(server.aof_buf) == 0)
/* Check if we need to do fsync even the aof buffer is empty,
* because previously in AOF_FSYNC_EVERYSEC mode, fsync is
* called only when aof buffer is not empty, so if users
* stop write commands before fsync called in one second,
* the data in page cache cannot be flushed in time. */
if (server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.aof_fsync_offset != server.aof_current_size &&
server.unixtime > server.aof_last_fsync &&
!(sync_in_progress = aofFsyncInProgress()))
goto try_fsync;
else
return;
//检查当前是否存在后台线程正在同步磁盘
if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
sync_in_progress = aofFsyncInProgress();
//如果存在，如果磁盘同步策略为每秒同步，则延迟该AOF缓冲区刷新操作，退出函数；
//如果已延迟时间超过2s，则强制刷新AOF缓冲区，函数继续向下执行。
if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force)
/* With this append fsync policy we do background fsyncing.
* If the fsync is still in progress we can try to delay
* the write for a couple of seconds. */
if (sync_in_progress)
if (server.aof_flush_postponed_start == 0)
/* No previous write postponing, remember that we are
* postponing the flush and return. */
server.aof_flush_postponed_start = server.unixtime;
return;
else if (server.unixtime - server.aof_flush_postponed_start < 2)
/* We were already waiting for fsync to finish, but for less
* than two seconds this is still ok. Postpone again. */
return;
/* Otherwise fall trough, and go write since we can't wait
* over two seconds. */
server.aof_delayed_fsync++;
serverLog(LL_NOTICE,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis.");
/* We want to perform a single write. This should be guaranteed atomic
* at least if the filesystem we are writing is a real physical one.
* While this will save us against the server being killed I don't think
* there is much to do about the whole server stopping for power problems
* or alike */
if (server.aof_flush_sleep && sdslen(server.aof_buf))
usleep(server.aof_flush_sleep);
latencyStartMonitor(latency);
nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
latencyEndMonitor(latency);
/* We want to capture different events for delayed writes:
* when the delay happens with a pending fsync, or with a saving child
* active, and when the above two conditions are missing.
* We also use an additional event name to save all samples which is
* useful for graphing / monitoring purposes. */
if (sync_in_progress)
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
else if (hasActiveChildProcess())
latencyAddSampleIfNeeded("aof-write-active-child",latency);
else
latencyAddSampleIfNeeded("aof-write-alone",latency);
latencyAddSampleIfNeeded("aof-write",latency);
/* We performed the write so reset the postponed flush sentinel to zero. */
server.aof_flush_postponed_start = 0;
if (nwritten != (ssize_t)sdslen(server.aof_buf))
static time_t last_write_error_log = 0;
int can_log = 0;
/* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE)
can_log = 1;
last_write_error_log = server.unixtime;
/* Log the AOF write error and record the error code. */
if (nwritten == -1)
if (can_log)
serverLog(LL_WARNING,"Error writing to the AOF file: %s",
strerror(errno));
server.aof_last_write_errno = errno;
else
if (can_log)
serverLog(LL_WARNING,"Short write while writing to "
"the AOF file: (nwritten=%lld, "
"expected=%lld)",
(long long)nwritten,
(long long)sdslen(server.aof_buf));
if (ftruncate(server.aof_fd, server.aof_current_size) == -1)
if (can_log)
serverLog(LL_WARNING, "Could not remove short write "
"from the append-only file. Redis may refuse "
"to load the AOF the next time it starts. "
"ftruncate: %s", strerror(errno));
else
/* If the ftruncate() succeeded we can set nwritten to
* -1 since there is no longer partial data into the AOF. */
nwritten = -1;
server.aof_last_write_errno = ENOSPC;
/* Handle the AOF write error. */
if (server.aof_fsync == AOF_FSYNC_ALWAYS)
/* We can't recover when the fsync policy is ALWAYS since the
* reply for the client is already in the output buffers, and we
* have the contract with the user that on acknowledged write data
* is synced on disk. */
serverLog(LL_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
exit(1);
else
/* Recover from failed write leaving data into the buffer. However
* set an error to stop accepting writes as long as the error
* condition is not cleared. */
server.aof_last_write_status = C_ERR;
/* Trim the sds buffer if there was a partial write, and there
* was no way to undo it with ftruncate(2). */
if (nwritten > 0)
server.aof_current_size += nwritten;
sdsrange(server.aof_buf,nwritten,-1);
return; /* We'll try again on the next call... */
else
/* Successful write(2). If AOF was in error state, restore the
* OK state and log the event. */
if (server.aof_last_write_status == C_ERR)
serverLog(LL_WARNING,
"AOF write error looks solved, Redis can write again.");
server.aof_last_write_status = C_OK;
server.aof_current_size += nwritten;
/* Re-use AOF buffer when it is small enough. The maximum comes from the
* arena size of 4k minus some overhead (but is otherwise arbitrary). */
if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000)
sdsclear(server.aof_buf);
else
sdsfree(server.aof_buf);
server.aof_buf = sdsempty();
//下面属于同步磁盘范畴
try_fsync:
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
* children doing I/O in the background. */
if (server.aof_no_fsync_on_rewrite && hasActiveChildProcess())
return;
/* Perform the fsync if needed. */
if (server.aof_fsync == AOF_FSYNC_ALWAYS)
/* redis_fsync is defined as fdatasync() for Linux in order to avoid
* flushing metadata. */
latencyStartMonitor(latency);
redis_fsync(server.aof_fd); /* Let's try to get this data on the disk */
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-fsync-always",latency);
server.aof_fsync_offset = server.aof_current_size;
server.aof_last_fsync = server.unixtime;
else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.unixtime > server.aof_last_fsync))
if (!sync_in_progress)
aof_background_fsync(server.aof_fd);
server.aof_fsync_offset = server.aof_current_size;
server.aof_last_fsync = server.unixtime;
以上是关于redis源码学习持久化机制：AOF的主要内容，如果未能解决你的问题，请参考以下文章