AOF 持久化是典型的异步任务，主协程(goroutine) 可以使用 channel 将数据发送到异步协程由异步协程执行持久化操作。
在 DB 中定义相关字段:
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
}
在进行持久化时需要注意两个细节:
我们在命令处理方法中返回 AOF 需要的额外信息：
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)
}
在异步协程中写入命令：
func (handler *Handler) handleAof() {
handler.currentDB = 0
for p := range handler.aofChan {
// 使用锁保证每次都会写入一条完整的命令
handler.pausingAof.RLock()
// 每个客户端都可以选择自己的数据库，所以 payload 中要保存客户端选择的数据库
// 选择的数据库与 aof 文件中最新的数据库不一致时写入一条 Select 命令
if p.dbIndex != handler.currentDB {
// select db
data := reply.MakeMultiBulkReply(utils.ToCmdLine("SELECT", strconv.Itoa(p.dbIndex))).ToBytes()
_, err := handler.aofFile.Write(data)
if err != nil {
logger.Warn(err)
continue // skip this command
}
handler.currentDB = p.dbIndex
}
// 写入命令内容
data := reply.MakeMultiBulkReply(p.cmdLine).ToBytes()
_, err := handler.aofFile.Write(data)
if err != nil {
logger.Warn(err)
}
handler.pausingAof.RUnlock()
}
// 关闭过程中主协程会先关闭 handler.aofChan，然后使用
func (db *DB) loadAof(maxBytes int) {
// delete aofChan to prevent write again
aofChan := db.aofChan
db.aofChan = nil
defer func(aofChan chan *reply.MultiBulkReply) {
db.aofChan = aofChan
}(aofChan)
file, err := os.Open(db.aofFilename)
if err != nil {
if _, ok := err.(*os.PathError); ok {
return
}
logger.Warn(err)
return
}
defer file.Close()
reader := utils.NewLimitedReader(file, maxBytes)
ch := parser.ParseStream(reader)
for p := range ch {
if p.Err != nil {
if p.Err == io.EOF {
break
}
logger.Error("parse error: " + p.Err.Error())
continue
}
if p.Data == nil {
logger.Error("empty payload")
continue
}
r, ok := p.Data.(*reply.MultiBulkReply)
if !ok {
logger.Error("require multi bulk reply")
continue
}
cmd := strings.ToLower(string(r.Args[0]))
command, ok := cmdTable[cmd]
if ok {
handler := command.executor
handler(db, r.Args[1:])
}
}
}
若我们对键a赋值100次会在AOF文件中产生100条指令但只有最后一条指令是有效的，为了减少持久化文件的大小需要进行AOF重写以删除无用的指令。
重写必须在固定不变的数据集上进行，不能直接使用内存中的数据。Redis 重写的实现方式是进行 fork 并在子进程中遍历数据库内的数据重新生成AOF文件。由于 golang 不支持 fork 操作，我们只能采用读取AOF文件生成副本的方式来代替fork。
在进行AOF重写操作时需要满足两个要求:
因此我们设计了一套比较复杂的流程：
首先准备开始重写操作:
func (handler *Handler) StartRewrite() (*rewriteCtx, error) {
// 暂停 aof 写入， 数据会在 aofChan 中暂时堆积
handler.pausingAof.Lock() // pausing aof
defer handler.pausingAof.Unlock()
// 调用 fsync 将缓冲区中的数据落盘，防止 aof 文件不完整造成错误
err := handler.aofFile.Sync()
if err != nil {
logger.Warn("fsync failed")
return nil, err
}
// 获得当前 aof 文件大小，用于判断哪些数据是 aof 重写过程中产生的
// handleAof 会保证每次写入完整的一条指令
fileInfo, _ := os.Stat(handler.aofFilename)
filesize := fileInfo.Size()
// 创建临时文件供重写使用
file, err := ioutil.TempFile("", "*.aof")
if err != nil {
logger.Warn("tmp file create failed")
return nil, err
}
return &rewriteCtx{
tmpFile: file,
fileSize: filesize,
dbIdx: handler.currentDB, // 重写开始时 aof 文件选中的数据库
}, nil
}
执行重写:
func (handler *Handler) DoRewrite(ctx *rewriteCtx) error {
tmpFile := ctx.tmpFile
// 将重写开始前的数据加载到内存
tmpAof := handler.newRewriteHandler()
tmpAof.LoadAof(int(ctx.fileSize))
// 将内存中的数据写入临时文件
for i := 0; i < config.Properties.Databases; i++ {
// select db
data := reply.MakeMultiBulkReply(utils.ToCmdLine("SELECT", strconv.Itoa(i))).ToBytes()
_, err := tmpFile.Write(data)
if err != nil {
return err
}
// dump db
tmpAof.db.ForEach(i, func(key string, entity *database.DataEntity, expiration *time.Time) bool {
cmd := EntityToCmd(key, entity)
if cmd != nil {
_, _ = tmpFile.Write(cmd.ToBytes())
}
if expiration != nil {
cmd := MakeExpireCmd(key, *expiration)
if cmd != nil {
_, _ = tmpFile.Write(cmd.ToBytes())
}
}
return true
})
}
return nil
}
结束重写的过程最为复杂:
func (handler *Handler) FinishRewrite(ctx *rewriteCtx) {
// 同样暂停 handleAof 的写入
handler.pausingAof.Lock()
defer handler.pausingAof.Unlock()
// 打开线上 aof 文件并 seek 到重写开始的位置
tmpFile := ctx.tmpFile
src, err := os.Open(handler.aofFilename)
if err != nil {
logger.Error("open aofFilename failed: " + err.Error())
return
}
defer func() {
_ = src.Close()
}()
_, err = src.Seek(ctx.fileSize, 0)
if err != nil {
logger.Error("seek failed: " + err.Error())
return
}
// 写入一条 Select 命令，使 tmpAof 选中重写开始时刻线上 aof 文件选中的数据库
data := reply.MakeMultiBulkReply(utils.ToCmdLine("SELECT", strconv.Itoa(ctx.dbIdx))).ToBytes()
_, err = tmpFile.Write(data)
if err != nil {
logger.Error("tmp file rewrite failed: " + err.Error())
return
}
// 对齐数据库后就可以把重写过程中产生的数据复制到 tmpAof 文件了
_, err = io.Copy(tmpFile, src)
if err != nil {
logger.Error("copy aof filed failed: " + err.Error())
return
}
// 使用 mv 命令用 tmpAof 代替线上 aof 文件
_ = handler.aofFile.Close()
_ = os.Rename(tmpFile.Name(), handler.aofFilename)
// 重新打开线上 aof
aofFile, err := os.OpenFile(handler.aofFilename, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600)
if err != nil {
panic(err)
}
handler.aofFile = aofFile
// 重新写入一次 select 指令保证 aof 中的数据库与 handler.currentDB 一致
data = reply.MakeMultiBulkReply(utils.ToCmdLine("SELECT", strconv.Itoa(handler.currentDB))).ToBytes()
_, err = handler.aofFile.Write(data)
if err != nil {
panic(err)
}
}
Original: https://www.cnblogs.com/Finley/p/12663636.html
Author: -Finley-
Title: Golang 实现 Redis(4): AOF 持久化与AOF重写
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/529396/
转载文章受原作者版权保护。转载请注明原作者出处！