1. 概述

Pika Raft 模式是 Pika 新增的分布式一致性功能，基于 braft 库实现了完整的 Raft 共识协议。相较于原有的主从复制模式，Raft 模式提供了强一致性保证、自动故障转移和成员动态管理等能力，适用于对数据一致性要求较高的生产场景。

1.1 设计目标

• 强一致性：通过 Raft 共识协议确保所有节点数据一致

• 高可用性：支持自动 Leader 选举，单节点故障不影响集群服务

• 易用性：兼容 Redis 协议，提供简洁的集群管理命令

• 可扩展性：支持动态添加和移除集群节点

1.2 与原有主从模式对比

2. 架构设计

2.1 整体架构

2.2 核心组件

2.2.1 RaftManager

RaftManager 是 Raft 功能的核心管理类，负责：

• 初始化和管理 Raft 节点

• 处理集群初始化和成员变更

• 日志复制和状态机应用

class RaftManager {
 public:
  // 初始化 Raft 管理器
  pstd::Status Init();
  
  // 启动 Raft 管理器
  pstd::Status Start();
  
  // 初始化 Raft 集群
  pstd::Status InitCluster(const std::string& db_name, 
                           const std::vector& peers);
  
  // 添加/移除节点
  pstd::Status AddNode(const std::string& db_name, const std::string& peer_addr);
  pstd::Status RemoveNode(const std::string& db_name, const std::string& peer_addr);
  
  // 追加日志（支持同步和异步模式）
  void AppendLog(const std::string& db_name, 
                 const ::pikiwidb::Binlog& log, 
                 std::promise&& promise,
                 storage::CommitCallback callback = nullptr);
};

2.2.2 PikaRaftNode

PikaRaftNode 封装了单个 Raft 节点的功能：

• 管理 braft::Node 生命周期

• 提供 Leader 选举和租约检查

• 处理成员变更操作

class PikaRaftNode {
 public:
  // 检查当前节点是否为 Leader
  bool IsLeader() const;
  
  // 获取当前 Leader 的 PeerId
  braft::PeerId GetLeaderId();
  
  // 成员变更
  pstd::Status AddPeer(const braft::PeerId& peer);
  pstd::Status RemovePeer(const braft::PeerId& peer);
  
  // 触发快照创建
  pstd::Status DoSnapshot(int64_t self_snapshot_index = 0, bool is_sync = true);
};

2.2.3 PikaStateMachine

PikaStateMachine 实现了 Raft 状态机接口：

class PikaStateMachine : public braft::StateMachine {
 public:
  // 应用已提交的日志条目
  void on_apply(braft::Iterator& iter) override;
  
  // 保存快照
  void on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done) override;
  
  // 加载快照
  int on_snapshot_load(braft::SnapshotReader* reader) override;
  
  // Leader 状态变更回调
  void on_leader_start(int64_t term) override;
  void on_leader_stop(const butil::Status& status) override;
};

2.3 数据流

2.3.1 写入流程

Client Write Request
        │
        ▼
┌───────────────────┐
│  Pika Command     │
│    Handler        │
└────────┬──────────┘
         │
         ▼
┌───────────────────┐
│  Storage Layer    │
│  (WriteBatch)     │
└────────┬──────────┘
         │
         ▼
┌───────────────────┐     Serialize
│  Binlog Creation  │──────────────┐
│  (Proto Format)   │              │
└────────┬──────────┘              ▼
         │                ┌───────────────────┐
         │                │  RaftManager      │
         │                │  AppendLog()      │
         │                └────────┬──────────┘
         │                         │
         │                         ▼
         │                ┌───────────────────┐
         │                │  braft::Node      │
         │                │  apply()          │
         │                └────────┬──────────┘
         │                         │
         │                         ▼ Raft Replication
         │                ┌───────────────────┐
         │                │  Follower Nodes   │
         │                │  Acknowledge      │
         │                └────────┬──────────┘
         │                         │
         │                         ▼ Committed
         │                ┌───────────────────┐
         │                │  StateMachine     │
         │                │  on_apply()       │
         │                └────────┬──────────┘
         │                         │
         ▼                         ▼
┌───────────────────┐     ┌───────────────────┐
│  Response to      │◄────│  Apply to         │
│  Client           │     │  RocksDB          │
└───────────────────┘     └───────────────────┘

2.3.2 Binlog 格式

Raft 模式使用 Protocol Buffers 格式的 Binlog 进行日志复制：

// 操作类型
enum OperateType {
  kNoOperate = 0;
  kPut = 1;
  kDelete = 2;
}

// 数据类型
enum DataType {
  kAll = 0;
  kStrings = 1;
  kHashes = 2;
  kLists = 3;
  kZSets = 4;
  kSets = 5;
  kStreams = 6;
}

// Binlog 条目
message BinlogEntry {
  DataType data_type = 1;
  uint32 cf_idx = 2;
  OperateType op_type = 3;
  bytes key = 4;
  optional bytes value = 5;
}

// Binlog（对应一次 Raft 日志提交）
message Binlog {
  uint32 db_id = 1;
  uint32 slot_idx = 2;
  repeated BinlogEntry entries = 3;
}

2.4 快照机制

Raft 快照用于：

• 压缩日志，节省存储空间

• 加速新节点的数据同步

• 支持节点故障恢复

PPosixFileSystemAdaptor 提供了自定义的快照文件系统适配器，基于 RocksDB 的 Checkpoint 功能实现：

class PPosixFileSystemAdaptor : public braft::PosixFileSystemAdaptor {
 public:
  // 打开文件时触发快照生成
  braft::FileAdaptor* open(const std::string& path, int oflag,
                           const ::google::protobuf::Message* file_meta,
                           butil::File::Error* e) override;
  
  // 递归添加所有快照文件
  void AddAllFiles(const std::string& dir, 
                   braft::LocalSnapshotMetaTable* snapshot_meta_memtable,
                   const std::string& base_path);
};

3. 配置说明

3.1 配置项

在 pika.conf 中配置 Raft 相关参数：

###################
## Raft Configuration
###################

# 启用 Raft 共识协议
# 启用后，Pika 将使用 Raft 确保节点间数据一致性
# 默认值：no
raft-enabled : yes

# Raft 组标识符
# 用于识别 Raft 集群，同一集群中所有节点应使用相同的 group-id
raft-group-id : pika_raft_group

# Raft 选举超时时间（毫秒）
# 如果在此时间内没有收到 Leader 心跳，Follower 将发起新的选举
# 较大的值可减少不必要的选举，但会增加故障转移时间
# 默认值：1000ms
raft-election-timeout-ms : 1000

# Raft 快照间隔时间（秒）
# 决定 Raft 创建状态机快照的频率
# 快照用于压缩日志和加速节点恢复
# 默认值：3600 秒（1小时）
raft-snapshot-interval-s : 3600

3.2 端口说明

Raft 模式使用额外的端口进行节点间通信：

4. 使用指南

4.1 集群部署

4.1.1 单节点部署（测试环境）

1. 修改配置文件启用 Raft：

raft-enabled : yes
raft-group-id : test_cluster

1. 启动 Pika：

./pika -c pika.conf

1. 初始化单节点集群：

redis-cli -p 9221 RAFT.CLUSTER INIT 127.0.0.1:12221

4.1.2 三节点集群部署（生产环境）

假设三个节点的配置如下：

1. 在每个节点配置相同的 raft-group-id：

raft-enabled : yes
raft-group-id : production_cluster
raft-election-timeout-ms : 1000
raft-snapshot-interval-s : 3600

1. 启动所有节点

2. 在所有节点执行集群初始化：

redis-cli -p 9221 RAFT.CLUSTER INIT 192.168.1.1:12221,192.168.1.2:12221,192.168.1.3:12221

4.2 集群管理命令

4.2.1 RAFT.CLUSTER INIT

初始化 Raft 集群。

语法：

RAFT.CLUSTER INIT [peer1:port1,peer2:port2,...] [db_name]

参数：

• peers：集群成员列表，格式为 ip:raft_port，多个成员用逗号分隔

• db_name：数据库名称，默认为 db0

示例：

# 初始化三节点集群
RAFT.CLUSTER INIT 192.168.1.1:12221,192.168.1.2:12221,192.168.1.3:12221

# 为新节点准备加入集群（不指定 peers）
RAFT.CLUSTER INIT

4.2.2 RAFT.CLUSTER INFO

获取集群状态信息。

语法：

RAFT.CLUSTER INFO [db_name]

返回信息：

Group: pika_raft_group_db0
PeerId: 192.168.1.1:12221:0
State: LEADER
Leader: 192.168.1.1:12221:0
Term: 5
Cluster Members (3): 192.168.1.1:12221:0, 192.168.1.2:12221:0, 192.168.1.3:12221:0
Committed Index: 1234
Known Applied Index: 1234
Pending Index: 0
Pending Queue Size: 0
Applying Index: 0
First Index: 100
Last Index: 1234

4.2.3 RAFT.NODE ADD

向集群添加新节点。

语法：

RAFT.NODE ADD peer_address [db_name]

参数：

• peer_address：新节点的 Raft 地址，格式为 ip:raft_port

• db_name：数据库名称，默认为 db0

注意：此命令必须在 Leader 节点上执行。

示例：

# 添加新节点
RAFT.NODE ADD 192.168.1.4:12221

4.2.4 RAFT.NODE REMOVE

从集群移除节点。

语法：

RAFT.NODE REMOVE peer_address [db_name]

参数：

• peer_address：要移除节点的 Raft 地址

• db_name：数据库名称，默认为 db0

注意：此命令必须在 Leader 节点上执行。

示例：

# 移除节点
RAFT.NODE REMOVE 192.168.1.4:12221

4.3 动态扩容流程

4.3.1 添加新节点

1. 在新节点上配置并启动 Pika（使用相同的 raft-group-id）

2. 在新节点上初始化（准备加入模式）：

redis-cli -h 192.168.1.4 -p 9221 RAFT.CLUSTER INIT

1. 在现有集群的 Leader 节点上添加新节点：

redis-cli -h  -p 9221 RAFT.NODE ADD 192.168.1.4:12221

1. 新节点会自动从 Leader 同步数据（通过快照或日志复制）

4.3.2 移除节点

1. 在 Leader 节点上执行移除命令：

redis-cli -h  -p 9221 RAFT.NODE REMOVE 192.168.1.4:12221

1. 停止被移除节点的 Pika 服务

4.4 故障恢复

4.4.1 单节点故障

• Follower 故障：集群继续正常服务，故障节点恢复后自动追赶日志

• Leader 故障：剩余节点自动选举新 Leader，选举超时后恢复服务

4.4.2 节点恢复流程

1. 节点重启后会检查本地 Raft 元数据

2. 如果元数据存在，自动从持久化配置恢复

3. 通过日志复制或快照安装追赶最新状态

5. 版本兼容性

6. 参考资料

• Raft 论文

• braft 官方文档