RDB 持久化机制

学习目标

深入理解 RDB 持久化的工作原理
掌握 RDB 文件格式和编码方式
实现完整的 RDB 生成和加载功能
理解 RDB 的优缺点和适用场景

RDB 持久化概述

RDB（Redis Database）是 Redis 的默认持久化方式，它通过创建数据快照来保存数据库状态。

RDB 的特点

紧凑性：RDB 文件是压缩的二进制文件，占用空间小
快速恢复：启动时加载 RDB 文件速度快
完整性：RDB 文件包含完整的数据快照
可移植性：RDB 文件可以在不同 Redis 版本间迁移

RDB 的触发方式

手动触发：SAVE 和 BGSAVE 命令
自动触发：根据配置的保存条件
主从同步：主服务器创建 RDB 发送给从服务器
关闭服务器：正常关闭时自动创建 RDB

RDB 文件格式

文件结构

┌─────────────────────────────────────────────────────────────┐
│                    RDB 文件结构                              │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │   文件头    │  │   数据库0   │  │   数据库N   │   ...   │
│  │  (9字节)    │  │  (数据)     │  │  (数据)     │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │   数据库N   │  │   数据库N   │  │   文件尾    │         │
│  │  (数据)     │  │  (数据)     │  │  (1字节)    │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
└─────────────────────────────────────────────────────────────┘

文件头格式

┌─────────────┬─────────────┬─────────────┐
│  REDIS(5)   │  版本号(4)  │  校验和(8)  │
└─────────────┴─────────────┴─────────────┘

️ RDB 实现

1. 基础结构定义

// rdb/types.go
package rdb

import (
    "bufio"
    "bytes"
    "encoding/binary"
    "fmt"
    "io"
    "time"
)

// RDB 常量
const (
    RDB_VERSION = 6
    RDB_MAGIC   = "REDIS"
)

// RDB 操作码
const (
    RDB_OPCODE_EOF           = 0xFF
    RDB_OPCODE_SELECTDB      = 0xFE
    RDB_OPCODE_EXPIRETIME_MS = 0xFC
    RDB_OPCODE_EXPIRETIME    = 0xFD
    RDB_OPCODE_RESIZEDB      = 0xFB
    RDB_OPCODE_AUX           = 0xFA
)

// 数据类型
const (
    RDB_TYPE_STRING  = 0
    RDB_TYPE_LIST    = 1
    RDB_TYPE_SET     = 2
    RDB_TYPE_ZSET    = 3
    RDB_TYPE_HASH    = 4
    RDB_TYPE_HASH_ZIPMAP = 9
    RDB_TYPE_LIST_ZIPLIST = 10
    RDB_TYPE_SET_INTSET   = 11
    RDB_TYPE_ZSET_ZIPLIST = 12
    RDB_TYPE_HASH_ZIPLIST = 13
    RDB_TYPE_LIST_QUICKLIST = 14
)

// RDB 编码类型
const (
    RDB_ENCODING_INT8  = 0
    RDB_ENCODING_INT16 = 1
    RDB_ENCODING_INT32 = 2
    RDB_ENCODING_LZF   = 3
)

// 数据库条目
type DBEntry struct {
    Key        string
    Value      interface{}
    ExpireTime int64
    Type       int
}

// RDB 写入器
type RDBWriter struct {
    writer *bufio.Writer
    buffer *bytes.Buffer
}

// RDB 读取器
type RDBReader struct {
    reader *bufio.Reader
}

2. RDB 写入器实现

// rdb/writer.go
package rdb

import (
    "bufio"
    "bytes"
    "compress/lzf"
    "encoding/binary"
    "fmt"
    "io"
    "time"
)

// 创建 RDB 写入器
func NewRDBWriter(w io.Writer) *RDBWriter {
    return &RDBWriter{
        writer: bufio.NewWriter(w),
        buffer: &bytes.Buffer{},
    }
}

// 写入 RDB 文件
func (w *RDBWriter) WriteRDB(data map[string]*DBEntry) error {
    // 写入文件头
    if err := w.writeHeader(); err != nil {
        return err
    }
    
    // 写入数据库 0
    if err := w.writeDatabase(0, data); err != nil {
        return err
    }
    
    // 写入 EOF
    if err := w.writeEOF(); err != nil {
        return err
    }
    
    // 写入校验和
    if err := w.writeChecksum(); err != nil {
        return err
    }
    
    return w.writer.Flush()
}

// 写入文件头
func (w *RDBWriter) writeHeader() error {
    // 写入魔数 "REDIS"
    if _, err := w.writer.WriteString(RDB_MAGIC); err != nil {
        return err
    }
    
    // 写入版本号
    version := fmt.Sprintf("%04d", RDB_VERSION)
    if _, err := w.writer.WriteString(version); err != nil {
        return err
    }
    
    return nil
}

// 写入数据库
func (w *RDBWriter) writeDatabase(dbNum int, data map[string]*DBEntry) error {
    // 写入 SELECTDB 操作码
    if err := w.writeOpcode(RDB_OPCODE_SELECTDB); err != nil {
        return err
    }
    
    // 写入数据库编号
    if err := w.writeLength(dbNum); err != nil {
        return err
    }
    
    // 写入 RESIZEDB 操作码
    if err := w.writeOpcode(RDB_OPCODE_RESIZEDB); err != nil {
        return err
    }
    
    // 写入哈希表大小
    if err := w.writeLength(len(data)); err != nil {
        return err
    }
    
    // 写入过期键数量（简化实现，设为 0）
    if err := w.writeLength(0); err != nil {
        return err
    }
    
    // 写入键值对
    for _, entry := range data {
        if err := w.writeKeyValue(entry); err != nil {
            return err
        }
    }
    
    return nil
}

// 写入键值对
func (w *RDBWriter) writeKeyValue(entry *DBEntry) error {
    // 写入过期时间（如果有）
    if entry.ExpireTime > 0 {
        if err := w.writeExpireTime(entry.ExpireTime); err != nil {
            return err
        }
    }
    
    // 写入数据类型
    if err := w.writeOpcode(entry.Type); err != nil {
        return err
    }
    
    // 写入键
    if err := w.writeString(entry.Key); err != nil {
        return err
    }
    
    // 写入值
    return w.writeValue(entry.Value, entry.Type)
}

// 写入过期时间
func (w *RDBWriter) writeExpireTime(expireTime int64) error {
    now := time.Now().Unix()
    if expireTime > now {
        // 使用秒级过期时间
        if err := w.writeOpcode(RDB_OPCODE_EXPIRETIME); err != nil {
            return err
        }
        return w.writeUint32(uint32(expireTime))
    }
    return nil
}

// 写入值
func (w *RDBWriter) writeValue(value interface{}, valueType int) error {
    switch valueType {
    case RDB_TYPE_STRING:
        return w.writeStringValue(value.(string))
    case RDB_TYPE_LIST:
        return w.writeListValue(value.([]string))
    case RDB_TYPE_SET:
        return w.writeSetValue(value.(map[string]bool))
    case RDB_TYPE_HASH:
        return w.writeHashValue(value.(map[string]string))
    case RDB_TYPE_ZSET:
        return w.writeZSetValue(value.(map[string]float64))
    default:
        return fmt.Errorf("unsupported value type: %d", valueType)
    }
}

// 写入字符串值
func (w *RDBWriter) writeStringValue(value string) error {
    return w.writeString(value)
}

// 写入列表值
func (w *RDBWriter) writeListValue(value []string) error {
    // 写入列表长度
    if err := w.writeLength(len(value)); err != nil {
        return err
    }
    
    // 写入列表元素
    for _, item := range value {
        if err := w.writeString(item); err != nil {
            return err
        }
    }
    
    return nil
}

// 写入集合值
func (w *RDBWriter) writeSetValue(value map[string]bool) error {
    // 写入集合大小
    if err := w.writeLength(len(value)); err != nil {
        return err
    }
    
    // 写入集合元素
    for member := range value {
        if err := w.writeString(member); err != nil {
            return err
        }
    }
    
    return nil
}

// 写入哈希值
func (w *RDBWriter) writeHashValue(value map[string]string) error {
    // 写入哈希大小
    if err := w.writeLength(len(value)); err != nil {
        return err
    }
    
    // 写入哈希字段
    for field, val := range value {
        if err := w.writeString(field); err != nil {
            return err
        }
        if err := w.writeString(val); err != nil {
            return err
        }
    }
    
    return nil
}

// 写入有序集合值
func (w *RDBWriter) writeZSetValue(value map[string]float64) error {
    // 写入有序集合大小
    if err := w.writeLength(len(value)); err != nil {
        return err
    }
    
    // 写入有序集合元素
    for member, score := range value {
        if err := w.writeString(member); err != nil {
            return err
        }
        if err := w.writeFloat64(score); err != nil {
            return err
        }
    }
    
    return nil
}

// 写入字符串
func (w *RDBWriter) writeString(s string) error {
    // 写入长度
    if err := w.writeLength(len(s)); err != nil {
        return err
    }
    
    // 写入字符串内容
    _, err := w.writer.WriteString(s)
    return err
}

// 写入长度
func (w *RDBWriter) writeLength(length int) error {
    if length < 0x40 {
        // 6位长度
        return w.writeByte(byte(length))
    } else if length < 0x4000 {
        // 14位长度
        return w.writeUint16(uint16(length) | 0x4000)
    } else {
        // 32位长度
        return w.writeUint32(uint32(length) | 0x80000000)
    }
}

// 写入操作码
func (w *RDBWriter) writeOpcode(opcode int) error {
    return w.writeByte(byte(opcode))
}

// 写入字节
func (w *RDBWriter) writeByte(b byte) error {
    return w.writer.WriteByte(b)
}

// 写入 16 位整数
func (w *RDBWriter) writeUint16(value uint16) error {
    return binary.Write(w.writer, binary.LittleEndian, value)
}

// 写入 32 位整数
func (w *RDBWriter) writeUint32(value uint32) error {
    return binary.Write(w.writer, binary.LittleEndian, value)
}

// 写入 64 位浮点数
func (w *RDBWriter) writeFloat64(value float64) error {
    return binary.Write(w.writer, binary.LittleEndian, value)
}

// 写入 EOF
func (w *RDBWriter) writeEOF() error {
    return w.writeOpcode(RDB_OPCODE_EOF)
}

// 写入校验和
func (w *RDBWriter) writeChecksum() error {
    // 简化实现，写入 8 字节 0
    for i := 0; i < 8; i++ {
        if err := w.writeByte(0); err != nil {
            return err
        }
    }
    return nil
}

3. RDB 读取器实现

// rdb/reader.go
package rdb

import (
    "bufio"
    "encoding/binary"
    "fmt"
    "io"
    "time"
)

// 创建 RDB 读取器
func NewRDBReader(r io.Reader) *RDBReader {
    return &RDBReader{
        reader: bufio.NewReader(r),
    }
}

// 读取 RDB 文件
func (r *RDBReader) ReadRDB() (map[string]*DBEntry, error) {
    // 读取文件头
    if err := r.readHeader(); err != nil {
        return nil, err
    }
    
    // 读取数据库
    data := make(map[string]*DBEntry)
    if err := r.readDatabase(data); err != nil {
        return nil, err
    }
    
    return data, nil
}

// 读取文件头
func (r *RDBReader) readHeader() error {
    // 读取魔数
    magic := make([]byte, 5)
    if _, err := io.ReadFull(r.reader, magic); err != nil {
        return err
    }
    
    if string(magic) != RDB_MAGIC {
        return fmt.Errorf("invalid RDB magic: %s", string(magic))
    }
    
    // 读取版本号
    version := make([]byte, 4)
    if _, err := io.ReadFull(r.reader, version); err != nil {
        return err
    }
    
    // 验证版本号
    var ver int
    if _, err := fmt.Sscanf(string(version), "%d", &ver); err != nil {
        return err
    }
    
    if ver > RDB_VERSION {
        return fmt.Errorf("unsupported RDB version: %d", ver)
    }
    
    return nil
}

// 读取数据库
func (r *RDBReader) readDatabase(data map[string]*DBEntry) error {
    for {
        // 读取操作码
        opcode, err := r.readByte()
        if err != nil {
            return err
        }
        
        switch opcode {
        case RDB_OPCODE_EOF:
            return nil
        case RDB_OPCODE_SELECTDB:
            // 读取数据库编号
            dbNum, err := r.readLength()
            if err != nil {
                return err
            }
            if dbNum != 0 {
                return fmt.Errorf("unsupported database number: %d", dbNum)
            }
        case RDB_OPCODE_RESIZEDB:
            // 读取哈希表大小
            _, err := r.readLength()
            if err != nil {
                return err
            }
            // 读取过期键数量
            _, err = r.readLength()
            if err != nil {
                return err
            }
        case RDB_OPCODE_EXPIRETIME:
            // 读取过期时间
            expireTime, err := r.readUint32()
            if err != nil {
                return err
            }
            // 读取键值对
            entry, err := r.readKeyValue()
            if err != nil {
                return err
            }
            entry.ExpireTime = int64(expireTime)
            data[entry.Key] = entry
        default:
            // 读取键值对
            entry, err := r.readKeyValue()
            if err != nil {
                return err
            }
            data[entry.Key] = entry
        }
    }
}

// 读取键值对
func (r *RDBReader) readKeyValue() (*DBEntry, error) {
    // 读取数据类型
    valueType, err := r.readByte()
    if err != nil {
        return nil, err
    }
    
    // 读取键
    key, err := r.readString()
    if err != nil {
        return nil, err
    }
    
    // 读取值
    value, err := r.readValue(valueType)
    if err != nil {
        return nil, err
    }
    
    return &DBEntry{
        Key:   key,
        Value: value,
        Type:  int(valueType),
    }, nil
}

// 读取值
func (r *RDBReader) readValue(valueType int) (interface{}, error) {
    switch valueType {
    case RDB_TYPE_STRING:
        return r.readString()
    case RDB_TYPE_LIST:
        return r.readList()
    case RDB_TYPE_SET:
        return r.readSet()
    case RDB_TYPE_HASH:
        return r.readHash()
    case RDB_TYPE_ZSET:
        return r.readZSet()
    default:
        return nil, fmt.Errorf("unsupported value type: %d", valueType)
    }
}

// 读取字符串
func (r *RDBReader) readString() (string, error) {
    length, err := r.readLength()
    if err != nil {
        return "", err
    }
    
    data := make([]byte, length)
    if _, err := io.ReadFull(r.reader, data); err != nil {
        return "", err
    }
    
    return string(data), nil
}

// 读取列表
func (r *RDBReader) readList() ([]string, error) {
    length, err := r.readLength()
    if err != nil {
        return nil, err
    }
    
    list := make([]string, 0, length)
    for i := 0; i < length; i++ {
        item, err := r.readString()
        if err != nil {
            return nil, err
        }
        list = append(list, item)
    }
    
    return list, nil
}

// 读取集合
func (r *RDBReader) readSet() (map[string]bool, error) {
    length, err := r.readLength()
    if err != nil {
        return nil, err
    }
    
    set := make(map[string]bool)
    for i := 0; i < length; i++ {
        member, err := r.readString()
        if err != nil {
            return nil, err
        }
        set[member] = true
    }
    
    return set, nil
}

// 读取哈希
func (r *RDBReader) readHash() (map[string]string, error) {
    length, err := r.readLength()
    if err != nil {
        return nil, err
    }
    
    hash := make(map[string]string)
    for i := 0; i < length; i++ {
        field, err := r.readString()
        if err != nil {
            return nil, err
        }
        value, err := r.readString()
        if err != nil {
            return nil, err
        }
        hash[field] = value
    }
    
    return hash, nil
}

// 读取有序集合
func (r *RDBReader) readZSet() (map[string]float64, error) {
    length, err := r.readLength()
    if err != nil {
        return nil, err
    }
    
    zset := make(map[string]float64)
    for i := 0; i < length; i++ {
        member, err := r.readString()
        if err != nil {
            return nil, err
        }
        score, err := r.readFloat64()
        if err != nil {
            return nil, err
        }
        zset[member] = score
    }
    
    return zset, nil
}

// 读取长度
func (r *RDBReader) readLength() (int, error) {
    b, err := r.readByte()
    if err != nil {
        return 0, err
    }
    
    if (b & 0x80) == 0 {
        // 6位长度
        return int(b), nil
    } else if (b & 0x40) == 0 {
        // 14位长度
        b2, err := r.readByte()
        if err != nil {
            return 0, err
        }
        return int((uint16(b)&0x3F)<<8 | uint16(b2)), nil
    } else {
        // 32位长度
        return r.readUint32()
    }
}

// 读取字节
func (r *RDBReader) readByte() (byte, error) {
    return r.reader.ReadByte()
}

// 读取 32 位整数
func (r *RDBReader) readUint32() (int, error) {
    var value uint32
    if err := binary.Read(r.reader, binary.LittleEndian, &value); err != nil {
        return 0, err
    }
    return int(value), nil
}

// 读取 64 位浮点数
func (r *RDBReader) readFloat64() (float64, error) {
    var value float64
    if err := binary.Read(r.reader, binary.LittleEndian, &value); err != nil {
        return 0, err
    }
    return value, nil
}

测试验证

1. 单元测试

// rdb/rdb_test.go
package rdb

import (
    "bytes"
    "testing"
    "time"
)

func TestRDBWriteAndRead(t *testing.T) {
    // 准备测试数据
    data := map[string]*DBEntry{
        "string_key": {
            Key:   "string_key",
            Value: "string_value",
            Type:  RDB_TYPE_STRING,
        },
        "list_key": {
            Key:   "list_key",
            Value: []string{"item1", "item2", "item3"},
            Type:  RDB_TYPE_LIST,
        },
        "set_key": {
            Key:   "set_key",
            Value: map[string]bool{"member1": true, "member2": true},
            Type:  RDB_TYPE_SET,
        },
        "hash_key": {
            Key:   "hash_key",
            Value: map[string]string{"field1": "value1", "field2": "value2"},
            Type:  RDB_TYPE_HASH,
        },
        "zset_key": {
            Key:   "zset_key",
            Value: map[string]float64{"member1": 1.0, "member2": 2.0},
            Type:  RDB_TYPE_ZSET,
        },
        "expired_key": {
            Key:        "expired_key",
            Value:      "expired_value",
            Type:       RDB_TYPE_STRING,
            ExpireTime: time.Now().Unix() + 3600, // 1小时后过期
        },
    }
    
    // 写入 RDB
    var buf bytes.Buffer
    writer := NewRDBWriter(&buf)
    if err := writer.WriteRDB(data); err != nil {
        t.Fatalf("WriteRDB failed: %v", err)
    }
    
    // 读取 RDB
    reader := NewRDBReader(&buf)
    loadedData, err := reader.ReadRDB()
    if err != nil {
        t.Fatalf("ReadRDB failed: %v", err)
    }
    
    // 验证数据
    if len(loadedData) != len(data) {
        t.Errorf("Expected %d entries, got %d", len(data), len(loadedData))
    }
    
    for key, expectedEntry := range data {
        loadedEntry, exists := loadedData[key]
        if !exists {
            t.Errorf("Key %s not found in loaded data", key)
            continue
        }
        
        if loadedEntry.Key != expectedEntry.Key {
            t.Errorf("Key mismatch: expected %s, got %s", expectedEntry.Key, loadedEntry.Key)
        }
        
        if loadedEntry.Type != expectedEntry.Type {
            t.Errorf("Type mismatch for key %s: expected %d, got %d", key, expectedEntry.Type, loadedEntry.Type)
        }
        
        if loadedEntry.ExpireTime != expectedEntry.ExpireTime {
            t.Errorf("ExpireTime mismatch for key %s: expected %d, got %d", key, expectedEntry.ExpireTime, loadedEntry.ExpireTime)
        }
    }
}

2. 性能测试

// rdb/benchmark_test.go
package rdb

import (
    "bytes"
    "testing"
    "time"
)

func BenchmarkRDBWrite(b *testing.B) {
    // 准备测试数据
    data := make(map[string]*DBEntry)
    for i := 0; i < 1000; i++ {
        key := fmt.Sprintf("key_%d", i)
        data[key] = &DBEntry{
            Key:   key,
            Value: fmt.Sprintf("value_%d", i),
            Type:  RDB_TYPE_STRING,
        }
    }
    
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        var buf bytes.Buffer
        writer := NewRDBWriter(&buf)
        writer.WriteRDB(data)
    }
}

func BenchmarkRDBRead(b *testing.B) {
    // 准备测试数据
    data := make(map[string]*DBEntry)
    for i := 0; i < 1000; i++ {
        key := fmt.Sprintf("key_%d", i)
        data[key] = &DBEntry{
            Key:   key,
            Value: fmt.Sprintf("value_%d", i),
            Type:  RDB_TYPE_STRING,
        }
    }
    
    // 写入 RDB
    var buf bytes.Buffer
    writer := NewRDBWriter(&buf)
    writer.WriteRDB(data)
    
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        reader := NewRDBReader(&buf)
        reader.ReadRDB()
    }
}