分布式存储之Raft协议应用详解

 func (s *server) processAppendEntriesRequest(req *AppendEntriesRequest){

      //如果AppendEntries请求的term小于该server当前term
      //返回失败
      if req.Term < s.currentTerm {
          s.debugln("server.ae.error: stale term")
          return false
      }

      if req.Term == s.currentTerm {

          // 当server处于candidate状态，则降为follower
          if s.state == Candidate {
              s.setState(Follower)
          }

      } else {
     
          // term 比 目前的大，则有新的leader产生
          // 更新 term 以及 leader
          s.updateCurrentTerm(req.Term, req.LeaderName)
      }

      // 将req.PrevLogIndex 编号后的log entry都删除了
      if err := s.log.truncate(req.PrevLogIndex, req.PrevLogTerm); err != nil {
          return false
      }

     // 加入该server节点的 log entries
      if err := s.log.appendEntries(req.Entries); err != nil {
          return false
      }

      // 提交至req.CommitIndex
      if err := s.log.setCommitIndex(req.CommitIndex); err != nil {
          s.debugln("server.ae.commit.error: ", err)
          return true
      }
      return true;
  }

func (s *server) processAppendEntriesResponse(resp *AppendEntriesResponse) {
 
      // 如果发现AppendEntries的返回中term较大，则角色变为follower
      // 这个判断非常重要，能够保证网络分区恢复后的一致性
      if resp.Term() > s.Term() {
          s.updateCurrentTerm(resp.Term(), "")
          return
      }

      // 对于回应返回成功的，更新map表
      if resp.append == true {
          s.syncedPeer[resp.peer] = true
      }

      if len(s.syncedPeer) < s.QuorumSize() {
          return
      }

      // 计算此次需要提交的编号
      var indices []uint64
      indices = append(indices, s.log.currentIndex())
      for _, peer := range s.peers {
          indices = append(indices, peer.getPrevLogIndex())
      }
      sort.Sort(sort.Reverse(uint64Slice(indices)))

      commitIndex := indices[s.QuorumSize()-1]
      committedIndex := s.log.commitIndex

      if commitIndex > committedIndex {
          s.log.sync()
          s.log.setCommitIndex(commitIndex)
      }
  }

func (s *server) processRequestVoteRequest(req *RequestVoteRequest){

      // term小于当前的，则返回false
      if req.Term < s.Term() {
          s.debugln("server.rv.deny.vote: cause stale term")
		  return false
      }

      // 如果请求的term大于该节点的term，则更新该节点的term
      // 如果term相等，且我们已经投给了其他候选节点(votedFor参数),则不投给该候选节点candidate
      // 即一个任期内只投给一个候选节点，但是可以投多次（可能存在网络异常，候选节点再次发出投票请求）
      if req.Term > s.Term() {
          s.updateCurrentTerm(req.Term, "")
      } else if s.votedFor != "" && s.votedFor != req.CandidateName {
          return false
      }

      // 如果candidate中最新日志项编号小于 当前server的最新日志项编号，则不投票
      // 这里满足了Raft的安全性：必须要比大部分其它候选者的log新，才有机会成为leader
      lastIndex, lastTerm := s.log.lastInfo()
      if lastIndex > req.LastLogIndex || lastTerm > req.LastLogTerm {
          return false
      }

      s.votedFor = req.CandidateName

      return true
  }

 func (s *server) candidateLoop() {
	  //处于 Candidate 状态，直到状态改变为leader或者follower
	  //否则超时后再次发起投票请求     
      for s.State() == Candidate {

          if doVote {
              //自增term
              s.currentTerm++
              s.votedFor = s.name

              // 向每个servers发送 RequestVote RPCs
              respChan = make(chan *RequestVoteResponse, len(s.peers))
              for _, peer := range s.peers {
                  s.routineGroup.Add(1)
                  go func(peer *Peer) {
                      defer s.routineGroup.Done()
                      peer.sendVoteRequest(newRequestVoteRequest(s.currentTerm, s.name, lastLogIndex, lastLogTerm), respChan)
                  }(peer)
              }

              //发起请求后的变量初始化
              votesGranted = 1
              timeoutChan = afterBetween(s.ElectionTimeout(), s.ElectionTimeout()*2)
              doVote = false
           }

           //如果收到超过半数以上的投票支持，则状态变为leader
           if votesGranted == s.QuorumSize() {
              s.setState(Leader)
              return
           }

          select {
          //处理收到RequestVote RPC回应
          case resp := <-respChan:
              if success := s.processVoteResponse(resp); success {
                  s.debugln("server.candidate.vote.granted: ", votesGranted)
                  votesGranted++
              }
          //如果选举超时，则继续选举    	
          case <-timeoutChan:
              doVote = true
       }