Raft 是一种强一致性协议。注意是 consul 中 raft 指 servers 间的协议，与 client 无关。
当consul server 收到 consul client 转发的请求或者同步过来的信息时，consul servers 内部需要保持一致，这就是 raft 的作用。
Raft 的几个关键概念：
Log - 行为序列（ordered sequence of entries）。Entry 包括所有的集群行为：添加节点、注册服务、写kv对等等。
Peer set - 参与 log replication 的所有成员，即所有 servers。
Quorum - 法定人数。对于 N 来说，quorum = (N / 2) + 1，即刚刚超过半数的 servers 数量。
即当 servers 只有 2 个时，quorum 也为 2，则当其中一个不可用时，整个系统则无法达到 quorum 数，将无法更新集群状态，同时，考虑同步性能，推荐 server 数为 3 或者 5 个。为什么不是不推荐 4 个？因为 4 个和 3 个一样，都只能容错一台节点错误。
Committed Entry - 达到 quorum 数的 server 收到了某个 entry，则称这个 entry 为 committed entry。
Term - 任期。每个 Term 都是一个连续递增的编号，每一轮选举都是一个 Term 周期，在一个 Term 中只能产生一个 leader。
Raft 的节点一共有三种角色：leader、follower、candidate。所有的 server 一开始都是 follower 。
正常情况下 follower 从 leader 处接收 entry，如果超过一定时间没有接收到 entry，则暗示着 leader 有可能出问题，则该 follower 自动成为 candidate。注意，当 follower 接收到 client 的 RPC 请求，follower 不进行处理，而是转发给 leader 进行处理。
在candidate 状态下，可以请求其他的节点给自己投票。当票数达到 quorum 时，就转变为 leader 状态。
在leader 状态下，需要处理 log。如果是读请求，则 leader 根据当前状态返回所需结果，如果是写请求，则生成一个新的 log 并且开始同步，直到 这个 log 被 committed（达到 quorum 数节点收到） 并且 applied（写入状态机）则写完成。
一致性模式
default - 只允许 leader 节点响应请求，但是当 leader 节点变更时，可能会出现两个 leader ，在这种情况下，老 leader 响应的可能是脏数据。
consistent - 强一致性但是最慢，leader 在响应读请求时，需要遍历其他节点确保自己是 leader。
stale - 允许非 leader 节点响应读请求。速度最快，但是值有可能是过期的（通常不会超过50ms过期）。
关键过程
raft 主要包括 Leader Election 和 Log Replication 两个关键步骤。
Leader Election
心跳超时（heartbeat timeout) - follower 与 leader 进行通信的最大时间间隔。
选举超时（election timeout) - candidate 收到每张选票的最大时间间隔。
当follower 心跳超时，会成为 candidate 进行选举开始选举。candidate 会向每个节点请求投票（request votes），若出现选举超时，则重新开始请求投票；若收到法定人数的计票则成为 leader。注意 candidate 会在开始请求投票之前给自己投一票，即只需（法定人数 - 1）张票即可。
注意当candidate 和 leader 接收到新 term 的请求时，都会退化为 follower。
Log Replication
leader 需要向其他节点同步操作，即 log replication 过程，该过程通过 append entry 指令发出。当有法定数节点（包括 leader 自身）感知到状态更新后，即称为该 log 为 committed。
所以一个有 3 个和 4 个 servers 的 raft 集群，容错节点数都为 1 个节点。
一次完整 Log Replication 如下进行：
1、 client请求leader更新状态；
2、 leader向其他节点发出appendentry指令；
3、 其他节点相应指令并成功返回给leader；
4、 当leader发现log已committed，即返回给client；
实现
当agent 运行在 server mode 下时会初始化 raft 实例：
https://github.com/hashicorp/raft
func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps SnapshotStore, trans Transport) (*Raft, error) {
// ...
// 初始化为 follower
r.setState(Follower)
// Start as leader if specified. This should only be used
// for testing purposes.
if conf.StartAsLeader {
r.setState(Leader)
r.setLeader(r.localAddr)
}
// ...
// Start the background work.
r.goFunc(r.run)
r.goFunc(r.runFSM)
r.goFunc(r.runSnapshots)
return r, nil
}
重点关注 r.goFunc(r.run) 中的 run 方法：
// run is a long running goroutine that runs the Raft FSM.
func (r *Raft) run() {
for {
// Check if we are doing a shutdown
select {
case <-r.shutdownCh:
r.setLeader("")
return
default:
}
// 根据节点当前角色运行对应的处理函数
switch r.getState() {
case Follower:
r.runFollower()
case Candidate:
r.runCandidate()
case Leader:
r.runLeader()
}
}
}
对于follower 来说需要处理心跳超时：
// runFollower runs the FSM for a follower.
func (r *Raft) runFollower() {
didWarn := false
r.logger.Info("entering follower state", "follower", r, "leader", r.Leader())
metrics.IncrCounter([]string{
"raft", "state", "follower"}, 1)
heartbeatTimer := randomTimeout(r.conf.HeartbeatTimeout)
for r.getState() == Follower {
select {
case rpc := <-r.rpcCh:
r.processRPC(rpc) // setLastContact
// ...
case a := <-r.applyCh:
// Reject any operations since we are not the leader
a.respond(ErrNotLeader)
case <-heartbeatTimer: // 心跳超时
// 重置随机 timeout，避免同时进行选举
heartbeatTimer = randomTimeout(r.conf.HeartbeatTimeout)
// Check if we have had a successful contact
lastContact := r.LastContact() // 上一次和 leader 通信的时刻
if time.Now().Sub(lastContact) < r.conf.HeartbeatTimeout {
continue
}
r.setLeader("") // 至此， leader 通信失败，置空 leader
r.setState(Candidate) // 转为 candidate 角色，跳出循环
case <-r.shutdownCh:
return
}
}
}
其中的processRPC 方法会执行 append entry 等指令：
func (r *Raft) processRPC(rpc RPC) {
if err := r.checkRPCHeader(rpc); err != nil {
rpc.Respond(nil, err)
return
}
switch cmd := rpc.Command.(type) {
case *AppendEntriesRequest: // 适用于 follower 的 append entry
r.appendEntries(rpc, cmd)
case *RequestVoteRequest: // candidate 请求投票
r.requestVote(rpc, cmd)
case *InstallSnapshotRequest:
r.installSnapshot(rpc, cmd)
case *TimeoutNowRequest:
r.timeoutNow(rpc, cmd)
default:
r.logger.Error("got unexpected command",
"command", hclog.Fmt("%#v", rpc.Command))
rpc.Respond(nil, fmt.Errorf("unexpected command"))
}
}
针对candidate 来说需要处理选举超时：
// runCandidate runs the FSM for a candidate.
func (r *Raft) runCandidate() {
r.logger.Info("entering candidate state", "node", r, "term", r.getCurrentTerm()+1)
metrics.IncrCounter([]string{
"raft", "state", "candidate"}, 1)
// 开始请求其他节点给自己投票，注意自己会给自己投一票
voteCh := r.electSelf()
electionTimer := randomTimeout(r.conf.ElectionTimeout)
// Tally the votes, need a simple majority
grantedVotes := 0
votesNeeded := r.quorumSize() // 法定人数即超过半数节点数
r.logger.Debug("votes", "needed", votesNeeded)
for r.getState() == Candidate {
select {
case rpc := <-r.rpcCh:
r.processRPC(rpc)
case vote := <-voteCh:
// 如果发现有更新的 term 则退化为 follower
if vote.Term > r.getCurrentTerm() {
r.logger.Debug("newer term discovered, fallback to follower")
r.setState(Follower)
r.setCurrentTerm(vote.Term)
return
}
// Check if the vote is granted
if vote.Granted {
grantedVotes++
r.logger.Debug("vote granted", "from", vote.voterID, "term", vote.Term, "tally", grantedVotes)
}
// 成功当选为 leader
if grantedVotes >= votesNeeded {
r.logger.Info("election won", "tally", grantedVotes)
r.setState(Leader)
r.setLeader(r.localAddr)
return
}
case a := <-r.applyCh:
a.respond(ErrNotLeader)
case <-electionTimer: // 选举超时
r.logger.Warn("Election timeout reached, restarting election")
return
case <-r.shutdownCh:
return
}
}
}
参考：
- https://www.consul.io/docs/architecture/consensus
- http://thesecretlivesofdata.com/raft
- https://raft.github.io/
版权声明：「DDKK.COM 弟弟快看，程序员编程资料站」本站文章，版权归原作者所有
来源：https://blog.csdn.net/jikunk8/category_10622611.html