Golang 实现 Redis(4): AOF 持久化与AOF重写
本文是使用 golang 实现 redis 系列的第四篇文章，将介绍如何使用 golang 实现 Append Only File 持久化及 AOF 文件重写。html
本文完整源代码在做者GithubHDT3213/godisgit
AOF 文件
AOF 持久化是典型的异步任务，主协程(goroutine) 可使用 channel 将数据发送到异步协程由异步协程执行持久化操做。github
在 DB 中定义相关字段:golang
type DB struct {
// 主线程使用此channel将要持久化的命令发送到异步协程
aofChan chan *reply.MultiBulkReply
// append file 文件描述符
aofFile *os.File
// append file 路径
aofFilename string
// aof 重写须要的缓冲区，将在AOF重写一节详细介绍
aofRewriteChan chan *reply.MultiBulkReply
// 在必要的时候使用此字段暂停持久化操做
pausingAof sync.RWMutex
}
在进行持久化时须要注意两个细节:redis
- get 之类的读命令并不须要进行持久化
- expire 命令要用等效的 expireat 命令替换。举例说明，10:00 执行
expire a 3600表示键 a 在 11:00 过时，在 10:30 载入AOF文件时执行expire a 3600就成了 11:30 过时与原数据不符。
咱们在命令处理方法中返回 AOF 须要的额外信息：sql
type extra struct {
// 表示该命令是否须要持久化
toPersist bool
// 如上文所述 expire 之类的命令不能直接持久化
// 若 specialAof == nil 则将命令原样持久化，不然持久化 specialAof 中的指令
specialAof []*reply.MultiBulkReply
}
type CmdFunc func(db *DB, args [][]byte) (redis.Reply, *extra)
以 SET 命令为例:数据库
func Set(db *DB, args [][]byte) (redis.Reply, *extra) {
//....
var result int
switch policy {
case upsertPolicy:
result = db.Put(key, entity)
case insertPolicy:
result = db.PutIfAbsent(key, entity)
case updatePolicy:
result = db.PutIfExists(key, entity)
}
extra := &extra{toPersist: result > 0} // 若实际写入了数据则toPresist=true, 若由于XX或NX选项没有实际写入数据则toPresist=false
if result > 0 {
if ttl != unlimitedTTL { // 使用了 EX 或 NX 选项
expireTime := time.Now().Add(time.Duration(ttl) * time.Millisecond)
db.Expire(key, expireTime)
// 持久化时使用 set key value 和 pexpireat 命令代替 set key value EX ttl 命令
extra.specialAof = []*reply.MultiBulkReply{
reply.MakeMultiBulkReply([][]byte{
[]byte("SET"),
args[0],
args[1],
}),
makeExpireCmd(key, expireTime),
}
} else {
db.Persist(key) // override ttl
}
}
return &reply.OkReply{}, extra
}
var pExpireAtCmd = []byte("PEXPIREAT")
func makeExpireCmd(key string, expireAt time.Time) *reply.MultiBulkReply {
args := make([][]byte, 3)
args[0] = pExpireAtCmd
args[1] = []byte(key)
args[2] = []byte(strconv.FormatInt(expireAt.UnixNano()/1e6, 10))
return reply.MakeMultiBulkReply(args)
}
在处理命令的调度方法中将 aof 命令发送到 channel:缓存
func (db *DB) Exec(c redis.Client, args [][]byte) (result redis.Reply) {
// ....
// normal commands
var extra *extra
cmdFunc, ok := router[cmd] // 找到命令对应的处理函数
if !ok {
return reply.MakeErrReply("ERR unknown command '" + cmd + "'")
}
// 使用处理函数执行命令
if len(args) > 1 {
result, extra = cmdFunc(db, args[1:])
} else {
result, extra = cmdFunc(db, [][]byte{})
}
// AOF 持久化
if config.Properties.AppendOnly {
if extra != nil && extra.toPersist {
// 写入 specialAof
if extra.specialAof != nil && len(extra.specialAof) > 0 {
for _, r := range extra.specialAof {
db.addAof(r)
}
} else {
// 写入原始命令
r := reply.MakeMultiBulkReply(args)
db.addAof(r)
}
}
}
return
}
在异步协程中写入命令：安全
func (db *DB) handleAof() {
for cmd := range db.aofChan {
// 异步协程在持久化以前会尝试获取锁，若其余协程持有锁则会暂停持久化操做
// 锁也保证了每次写入完整的一条指令不会格式错误
db.pausingAof.RLock()
if db.aofRewriteChan != nil {
db.aofRewriteChan <- cmd
}
_, err := db.aofFile.Write(cmd.ToBytes())
if err != nil {
logger.Warn(err)
}
db.pausingAof.RUnlock()
}
}
读取过程与协议解析器一节基本相同，不在正文中赘述：loadAof。app
AOF 重写
若咱们对键a赋值100次会在AOF文件中产生100条指令但只有最后一条指令是有效的，为了减小持久化文件的大小须要进行AOF重写以删除无用的指令。
重写必须在固定不变的数据集上进行，不能直接使用内存中的数据。Redis 重写的实现方式是进行 fork 并在子进程中遍历数据库内的数据从新生成AOF文件。因为 golang 不支持 fork 操做，咱们只能采用读取AOF文件生成副本的方式来代替fork。
在进行AOF重写操做时须要知足两个要求:
- 若 AOF 重写失败或被中断，AOF 文件需保持重写以前的状态不能丢失数据
- 进行 AOF 重写期间执行的命令必须保存到新的AOF文件中, 不能丢失
所以咱们设计了一套比较复杂的流程：
- 暂停AOF写入 -> 更改状态为重写中 -> 复制当前AOF文件 -> 恢复AOF写入
- 在重写过程当中，持久化协程在将命令写入文件的同时也将其写入内存中的重写缓存区
- 重写协程读取AOF副本并将重写到临时文件（tmp.aof）中
- 暂停AOF写入 -> 将重写缓冲区中的命令写入tmp.aof -> 使用临时文件tmp.aof覆盖AOF文件（使用文件系统的mv命令保证安全）-> 清空重写缓冲区 -> 恢复AOF写入
在不阻塞在线服务的同时进行其它操做是一项必需的能力，AOF重写的思路在解决这类问题时具备重要的参考价值。好比Mysql Online DDL: gh-ost采用了相似的策略保证数据一致。
首先准备开始重写操做:
func (db *DB) startRewrite() (*os.File, error) {
// 暂停AOF写入， 数据会在 db.aofChan 中暂时堆积
db.pausingAof.Lock()
defer db.pausingAof.Unlock()
// 建立重写缓冲区
db.aofRewriteChan = make(chan *reply.MultiBulkReply, aofQueueSize)
// 建立临时文件
file, err := ioutil.TempFile("", "aof")
if err != nil {
logger.Warn("tmp file create failed")
return nil, err
}
return file, nil
}
在重写过程当中，持久化协程进行双写：
func (db *DB) handleAof() {
for cmd := range db.aofChan {
db.pausingAof.RLock()
if db.aofRewriteChan != nil {
// 数据写入重写缓冲区
db.aofRewriteChan <- cmd
}
_, err := db.aofFile.Write(cmd.ToBytes())
if err != nil {
logger.Warn(err)
}
db.pausingAof.RUnlock()
}
}
执行重写:
func (db *DB) aofRewrite() {
file, err := db.startRewrite()
if err != nil {
logger.Warn(err)
return
}
// load aof file
tmpDB := &DB{
Data: dict.MakeSimple(),
TTLMap: dict.MakeSimple(),
Locker: lock.Make(lockerSize),
interval: 5 * time.Second,
aofFilename: db.aofFilename,
}
tmpDB.loadAof()
// rewrite aof file
tmpDB.Data.ForEach(func(key string, raw interface{}) bool {
var cmd *reply.MultiBulkReply
entity, _ := raw.(*DataEntity)
switch val := entity.Data.(type) {
case []byte:
cmd = persistString(key, val)
case *List.LinkedList:
cmd = persistList(key, val)
case *set.Set:
cmd = persistSet(key, val)
case dict.Dict:
cmd = persistHash(key, val)
case *SortedSet.SortedSet:
cmd = persistZSet(key, val)
}
if cmd != nil {
_, _ = file.Write(cmd.ToBytes())
}
return true
})
tmpDB.TTLMap.ForEach(func(key string, raw interface{}) bool {
expireTime, _ := raw.(time.Time)
cmd := makeExpireCmd(key, expireTime)
if cmd != nil {
_, _ = file.Write(cmd.ToBytes())
}
return true
})
db.finishRewrite(file)
}
重写完毕后写入缓冲区中的数据并替换正式文件:
func (db *DB) finishRewrite(tmpFile *os.File) {
// 暂停AOF写入
db.pausingAof.Lock()
defer db.pausingAof.Unlock()
// 将重写缓冲区内的数据写入临时文件
// 由于handleAof已被暂停，在遍历期间aofRewriteChan中不会有新数据
loop:
for {
select {
case cmd := <-db.aofRewriteChan:
_, err := tmpFile.Write(cmd.ToBytes())
if err != nil {
logger.Warn(err)
}
default:
// 只有 channel 为空时才会进入此分支
break loop
}
}
// 释放重写缓冲区
close(db.aofRewriteChan)
db.aofRewriteChan = nil
// 使用临时文件代替aof文件
_ = db.aofFile.Close()
_ = os.Rename(tmpFile.Name(), db.aofFilename)
// 从新打开文件描述符以保证正常写入
aofFile, err := os.OpenFile(db.aofFilename, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600)
if err != nil {
panic(err)
}
db.aofFile = aofFile
}
相关文章
- 1. AOF持久化
- 2. Redis 持久化之 AOF 重写
- 3. redis持久化的AOF重写机制
- 4. 【Redis】持久化RDB与AOF
- 5. Redis持久化RDB与AOF
- 6. redis持久化RDB与AOF
- 7. Redis持久化（RDB与AOF）
- 8. redis持久化 rdb与aof
- 9. redis持久化 RDB与AOF
- 10. ＜Redis＞AOF持久化
- 更多相关文章...
- • Redis的两种备份（持久化）方式：RDB和AOF - Redis教程
- • Hibernate持久化对象的状态转换实例 - Hibernate教程
- • Spring Cloud 微服务实战(三) - 服务注册与发现
- • ☆基于Java Instrument的Agent实现
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. AOF持久化
- 2. Redis 持久化之 AOF 重写
- 3. redis持久化的AOF重写机制
- 4. 【Redis】持久化RDB与AOF
- 5. Redis持久化RDB与AOF
- 6. redis持久化RDB与AOF
- 7. Redis持久化（RDB与AOF）
- 8. redis持久化 rdb与aof
- 9. redis持久化 RDB与AOF
- 10. ＜Redis＞AOF持久化