Raft解决什么问题

解决数据一致性的问题, 允许一组机器像一台整体一样工作, 只要写入成功, 那么就不用担心这一组机器中一定数量机器的故障导致数据丢失或者不一致的情况.因为这一组机器上面存储着完全一致的用户数据,为了叙述方便, 把这一组机器叫做一个Raft集群.一个Raft集群最少需要3台机器.

但是瓶颈也是显而易见的, 这一组机器的吞吐, 比这一组机器中任意一台机器的吞吐都还要低, 这是因为请求进入到集群中需要进行分发,确认等协调工作. 所以在大型系统中, 我们需要把用户数据按照用户UID进行切片, 调整到适合一个Raft集群存储的大小. 如果有需要Raft集群之间的协调, 事务工作, 需要解耦, 把事务包装成事务消息, 再通过消息中间件的方式完成.

Raft简介

Raft集群中角色有三种, 分为Leader, Candidate,Follower. 其中Candidate是一个中间状态, 只发生在发生领导人选举的过程中.

对于写请求来说

• Leader负责接受用户的请求(如 set key = value),给请求包装成一条日志, 并且分配一个单调递增编号Index, 并且分发这个请求日志到集群中所有其他机器上, 等到集群中的大多数机器确认收到了这个请求后,Raft就会调整自己的commitIndex(代表可以提交到状态机中的序号), 异步向本地状态机内提交这个请求(执行set key = value),并且向集群中其他机器发送commitIndex, 要求其他机器提交这个请求.
• 需要注意的是, 这里并不是一个完整的两阶段提交.

对于读请求来说

• Leader在收到读请求后, 记录下此时的commitIndex, 并且向集群中其他机器发送心跳广播, 如果确认多数派相应成功, 证明自己还是Leader, 再确定已经执行的Index(applyIndex) >= commitIndex, 然后在从状态机中取出结果返回给客户端.
• 如果是Follower收到了读请求, 需要向Leader请求当前的commitIndex(Leader收到请求后, 也会和上述行为一样,发送心跳确定自己的Leader地位), 并且保证Follower上的applyIndex > commitIndex.再返回读请求.
• 这仅仅是ReadIndex的一种读实现, 你可以看到, 步骤很繁琐, 但是是一致性最高的读实现. 另外还有基于时间的LeaseRead实现, 效率更高, 但是依赖时间.

如果集群中的一台机器在一段时间内没有收到心跳, 就会把自己的Term加一, 并且向其他机器发送选举请求.所以任期的增加来自于选主.如果没有发生选主, 那么集群内并不会发生任期升级.

Raft协议能够保证在一个Term内, 只有一个Leader, 手段就是.

• 成为Leader需要经过集群中大多数机器赞成
• 任何一台机器, 在一个Term内, 只能给一台机器投票

Raft实现

首先我们来考虑如何实现选主.我们先以时间序列来看,选主的过程中发生了什么.

1. 在新集群启动的时候, 所有机器A, B, C的默认状态都是Follower, 所有机器地址endpoint作为初始化参数传入进程.
2. 如果收到心跳, 就开始作为Follower搬砖,选主结束, 如果在经过一段随机(大于心跳数倍)时间后, 开始发起Election.随机的目的是为了保证不要同时发起Election,当然, 并不能保证一定不会同时Election, 但是绝大多数情况下不会同时发生.
3. 集群初始化时是没有Leader, 所以没有机器收到心跳,机器A开始Election, 首先, 机器A把自己的身份置为Candidate, 把Term自增一位, 并且把当前Term的票投给自己.
4. 随后, 机器A会向其他机器发起vote请求, 要求投票给自己.

- 对于接受到这个vote请求的机器来说, 如果机器发现Candidate的Term小于自己, 就否决,并且返回自己的Term
- 如果机器发现Candidate的Term大于自己, 就更新term, 并且重置votedFor
- 如果机器发现在这个term已经给别人vote过了, 就否决
- 如果Candidate的日志没有自己新, 就否决(这是后面实现的逻辑, 本次暂不实现)
- 赞成
复制代码

1. 机器A在发送vote请求后, 如果发现集群中的大多数赞成, 就跳到步骤7, 成为Leader

2. 如果发现返回的Term大于自己, 就放弃本次Election, 更新自己的term到返回的term, 重置自己状态为Follower,回到步骤3

3. 如果机器A在随机一段时间后(数倍心跳时间), 没有收到大多数赞成,就回到步骤3

4. 机器A选主成功, 调整自身状态为Leader, 并且立即发送一次心跳(避免其他机器超时进入选主状态)

5. 周期性发送心跳, 保证Leader地位

代码实现

首选来定义一下一个Raft Peer

type Raft struct {mu        sync.RWMutex        // Lock to protect shared access to this peer's statepeers     []*labrpc.ClientEnd // RPC end points of all peerspersister *Persister          // Object to hold this peer's persisted stateme        int                 // this peer's index into peers[]nextIndex       []int         //    matchIndex      []inttimeoutDuration time.Duration // 基础超时时间currentTerm     int           // 当前任期votedFor        int           // 投给谁了logEntries      []RLog        // log entrisrole            Role          // Leader Candidate Followerleader int                    // 谁是LeadercommitIndex      int          lastAppliedIndex intheartbeatTime time.Duration  // 心跳周期recvPeerMsgCh chan interface{}// 收到集群内请求的chan, 和收到外部请求的chan区分开}
复制代码

说实话, 第一次看到这些互相依赖的超时, 并行处理, 还是有点懵逼的..

不过仔细想了下, 需要把这些行为抽象成几个线程.

第一个线程, 主要负责成为Leader之后,调整自身身份,并且周期性发送心跳

func (rf *Raft) stepLeader() {rf.mu.Lock()rf.leader = rf.merf.role = Leaderrf.logWithoutLockSprint("now i am the leader")rf.mu.Unlock()// 立即发送心跳go rf.broadcast()for {_, isLeader := rf.GetState()if isLeader {select {case <-time.Tick(rf.heartbeatTime):// broadcast heartbeatgo rf.broadcast()case logEntry := <-rf.reqCh:switch e := logEntry.(type) {case RLog:rf.mu.Lock()rf.reqBufNum --rf.logEntries = append(rf.logEntries, e)rf.mu.Unlock()go rf.broadcast()}}} }
}
复制代码

第二个线程, 来负责处理集群间的RPC, 统计超时情况, 以及选主.

这里看到, 在选主的时候, 是不会接受其他的投票以及追加日志以及心跳请求的.这是我的实现, 这样处理更清晰.

即便造成了VoteRequest以及AppendEntry的超时也没关系.VoteRequest超时是可以接受的,因为即使Term不对劲的话, 会让Candidate通过投票返回的结果放弃选主.

因为Candidate本身就不会给其他人投票.AppendEntry超时就更没关系了, 不应该让AppendEntry打断选主的流程

func (rf *Raft) loop() {for {select {case msg := <-rf.recvPeerMsgCh:switch m := msg.(type) {case VoteRequest:rf.VoteRequestHandle(m)case AppendEntriesRequest:rf.AppendEntriesHandle(m)}case <-time.After(rf.jitterTimeoutDuration()):role, _ := rf.GetRoleAndTerm()if role == Follower {success := rf.raiseElection()if success {go rf.stepLeader()}}}}
}
复制代码

主要的线程就是上面的两个, 第一个线程也是第二个线程派生出来的, 但是可能会同时运行.

下面看看发起选举的实现,尽可能对变量加锁, 并且避免锁中的外部调用. 可以的话, 用原子操作代替锁的使用

func (rf *Raft) raiseElection() bool {rf.logWithRLock("Start to raise election")rf.mu.Lock()rf.role = Candidaterf.votedFor = rf.merf.currentTerm += 1rf.logWithoutLock(fmt.Sprintf("Term upgrade to %d due to raise election",rf.currentTerm))rf.mu.Unlock()_, currentTerm := rf.GetRoleAndTerm()lastLogIndex := rf.getLastLogIndex()lastLogTerm := rf.getLastLogTerm()peerNum := rf.getPeersNum()// 自带Candidate一张选票winThreshold := int64(peerNum/2 + 1)counter := int64(1)// 只有写没有读的ch会被GC, 所以不用担心resultCh := make(chan bool)arg := &RequestVoteArgs{Term:         currentTerm,CandidateId:  rf.me,LastLogIndex: lastLogIndex,LastLogTerm:  lastLogTerm,}wg := sync.WaitGroup{}wg.Add(peerNum - 1)go func() {for i := 0; i < len(rf.peers); i++ {i := iif i == rf.me {continue}go func() {defer wg.Done()reply := &RequestVoteReply{}ok := rf.sendRequestVote(i, arg, reply)if ok {if reply.VoteGranted {atomic.AddInt64(&counter, 1)if atomic.LoadInt64(&counter) >= winThreshold {resultCh <- true}} else {if reply.Term > currentTerm {// 这轮已经输了rf.mu.Lock()rf.upgradeTermWithoutLock(reply.Term)rf.mu.Unlock()resultCh <- false}}}}()}wg.Wait()if atomic.LoadInt64(&counter) < winThreshold {resultCh <- false}}()// 实现选举超时select {case result := <-resultCh:return resultcase <-time.After(rf.jitterTimeoutDuration()):rf.mu.Lock()rf.role = Followerrf.logWithoutLockSprint("Lost this term election because of timeout")rf.mu.Unlock()return false}
}
复制代码

同步转异步

外部RPC调用都是同步调用, 而我们在mainLoop中, 希望异步的来处理这个流程, 所以这个地方做了一点小小的改造.

在处理请求时,把请求带上一个done chan, 包装成一个wrap request, 然后把wrap request发送给mainLoop, 最后监听done chan是否返回.

在真正进行请求处理是, 通过defer函数来向done chan发送结束消息

另外一点是, 由于处理请求完全都是内部变量, 状态的操作, 并且和其他线程抢占的时间很少(因为我们都是监听chan 顺序处理请求, 并且不会和发起选举这类请求同时处理), 我可以直接进入函数就Lock上锁, 最后defer Unlock.

func (rf *Raft) sendAppendEntries(server int, entries *AppendEntries, reply *AppendEntriesReply) bool {ok := rf.peers[server].Call("Raft.AppendEntries", entries, reply)return ok
}// append entries
func (rf *Raft) AppendEntries(entries *AppendEntries, reply *AppendEntriesReply) {done := make(chan struct{})entriesReq := AppendEntriesRequest{entries, reply, done}rf.recvPeerMsgCh <- entriesReq<-done
}// append entry handler
func (rf *Raft) AppendEntriesHandle(request AppendEntriesRequest) {entries := request.entriesreply := request.replydefer func() {rf.mu.Unlock()request.done <- struct{}{}close(request.done)}()rf.mu.Lock()//...
}
复制代码

最后, 执行一次go -test run 2A就ok了

➜  raft git:(master) go test -run 2A
复制代码

记得多执行几次, 因为这种多线程的bug不是必然复现的, 提高执行次数有助于更早的发现bug