<p>这篇文章主要是从源码的级别来看 Raft 算法的实现。在网上找到了一个简化版： <a href="/misc/goto?guid=4959754432427820978" rel="nofollow,noindex">源码</a> .</p>    <p>一个 Server 结构代表 Raft 网络中的一个 节点 。节点会创建一个 Server ，并且通过 端(peers) 接口的方式暴露给其他节点。</p>    <p>传输层采用 http 包装， 端对端 通信通过 rest http 方式。</p>    <pre>  <code class="language-go">|http transport| ---> |peers| ---> |server|  </code></pre>    <h2>项目简介</h2>    <h2>节点的增加和删除</h2>    <p>支持动态增删节点，采用一个简单的 共识 算法(节点更新时，接受配置更新的节点需要超过1/2，即新集群要大于旧集群)。</p>    <h2>roadmap</h2>    <ul>     <li>leader选举</li>     <li>日志复制</li>     <li>单元测试</li>     <li>http 传输层</li>     <li>配置变更</li>    </ul>    <p>除此之外，还有一些是未完成的</p>    <ul>     <li>net/rpc 传输层或者其他类型的传输层</li>     <li>日志压缩</li>     <li>快照安装以及恢复</li>     <li>完整的 demo 应用</li>     <li>一些比较复杂的测试用例<br> 具体细节，看下面的代码分析。 <h2>源码分析</h2> <h2>源码目录结构</h2> <pre>  <code class="language-go">├── JOINT-CONSENSUS.md  ├── LICENSE  ├── README.md  ├── configuration.go // 配置  ├── example_test.go // demo  ├── log.go // 日志  ├── log_test.go // 日志测试模块  ├── peers.go // 端  ├── peers_test.go // 端模块  ├── rpc.go // rpc 对象模块  ├── server.go //  server模块  ├── server_internals_test.go // server内部测试模块  ├── server_test.go //  server测试模块  ├── transport.go // 传输层  └── transport_test.go // 传输层模块  </code></pre> </li>    </ul>    <h2>主要的数据结构</h2>    <h3>rpc.go</h3>    <pre>  <code class="language-go">// 日志追加  type appendEntriesTuple struct {          // 日志追加请求          Request  appendEntries           // 应答通道          Response chan appendEntriesResponse   }  // 投票选举  type requestVoteTuple struct {          // 选举内容          Request  requestVote           // 选举结构应答          Response chan requestVoteResponse  }    // appendEntries represents an appendEntries RPC.  // 日志追加-实体  type appendEntries struct {          // 任期号          Term         uint64     `json:"term"`           // leader 标识          LeaderID     uint64     `json:"leader_id"`           // 前一个日志索引          PrevLogIndex uint64     `json:"prev_log_index"`           // 前一个日志任期号          PrevLogTerm  uint64     `json:"prev_log_term"`           // 要追加的实体数组-支持批量追加          Entries      []logEntry `json:"entries"`           // 已经committed的缩影          CommitIndex  uint64     `json:"commit_index"`   }    // appendEntriesResponse represents the response to an appendEntries RPC.  // 日志追加应答  type appendEntriesResponse struct {          // 应答节点的任期号          Term    uint64 `json:"term"`           // 是否追加成功          Success bool   `json:"success"`           // 失败的原因          reason  string   }    // requestVote represents a requestVote RPC.  // 投票请求实体  type requestVote struct {           // 发起者的任期号          Term         uint64 `json:"term"`          // 发起者的id          CandidateID  uint64 `json:"candidate_id"`          // 发起者的最新条目          LastLogIndex uint64 `json:"last_log_index"`          // 发起者的最新任期号          LastLogTerm  uint64 `json:"last_log_term"`  }    // requestVoteResponse represents the response to a requestVote RPC.  // 投票应答  type requestVoteResponse struct {          // 应答者任期号          Term        uint64 `json:"term"`          // 应答结果，true赞同，false反对          VoteGranted bool   `json:"vote_granted"`          // 反对原因          reason      string  }  </code></pre>    <h3>log.go</h3>    <p><img src="https://simg.open-open.com/show/f2b9a5aa2ed1d1fbf99ff7b7cd04fd2e.png"></p>    <pre>  <code class="language-go">var (          // 任期号太小          errTermTooSmall    = errors.New("term too small")          // 日志索引太小          errIndexTooSmall   = errors.New("index too small")          // 日志缩影太大          errIndexTooBig     = errors.New("commit index too big")          // 日志条目内容已损坏          errInvalidChecksum = errors.New("invalid checksum")      // 无效的命令          errNoCommand       = errors.New("no command")          // 错误的日志索引          errBadIndex        = errors.New("bad index")          // 错误任期号          errBadTerm         = errors.New("bad term")  )  // 日志结构  type raftLog struct {          // 日志读写锁          sync.RWMutex          // 日志存储接口          store     io.Writer          // 日志镜像，现在存储于内存          entries   []logEntry          // 下一条日志commit索引          commitPos int          // "操作"的回调函数，这个函数比较重要，可以"操作集合"镜像，          // 在快照时，只需要将"结果"快到存储层即可          apply     func(uint64, []byte)[]byte  }    func newRaftLog(store io.ReadWriter, applyfunc(uint64, []byte) []byte) *raftLog {          l := &raftLog{           store:     store,           entries:   []logEntry{},           commitPos: -1, // no commits to begin with           apply:     apply,          }          l.recover(store)          return l  }    // recover reads from the log's store, to populate the log with log entries  // from persistent storage. It should be called once, at log instantiation.  // 日志恢复，当服务重启时，重建日志条目(一般重建都是居于于快照和日志的，可是这里没有实现快照，所以从日志中重建即可)  // 1、这里的日志时commited之后的日志，所以重建时，commitPos也会更新  // 2、重建日志条目，会调用apply函数，对日志进行处理，这个函数相当于"状态机"功能；如果有快照(相当于Redis 的RDB)，先将安装快照，再恢复日志(相当于Redis 的aof)  func (l *raftLog)recover(r io.Reader)error {          for {           var entry logEntry           switch err := entry.decode(r); err {           case io.EOF:            return nil // successful completion           case nil:            if err := l.appendEntry(entry); err != nil {             return err            }            l.commitPos++            l.apply(entry.Index, entry.Command)           default:            return err // unsuccessful completion           }          }  }    // entriesAfter returns a slice of log entries after (i.e. not including) the  // passed index, and the term of the log entry specified by index, as a  // convenience to the caller. (This function is only used by a leader attempting  // to flush log entries to its followers.)  //  // This function is called to populate an AppendEntries RPC. That implies they  // are destined for a follower, which implies the application of the commands  // should have the response thrown away, which implies we shouldn't pass a  // commandResponse channel (see: commitTo implementation). In the normal case,  // the raftLogEntries we return here will get serialized as they pass thru their  // transport, and lose their commandResponse channel anyway. But in the case of  // a LocalPeer (or equivalent) this doesn't happen. So, we must make sure to  // proactively strip commandResponse channels.  // 检索index之后的日志条目  func (l *raftLog)entriesAfter(indexuint64)([]logEntry,uint64) {          l.RLock()          defer l.RUnlock()        // 1.检索出index对应term以及在实体集合entries中的位置Pos          pos := 0          lastTerm := uint64(0)          for ; pos < len(l.entries); pos++ {           if l.entries[pos].Index > index {            break           }           lastTerm = l.entries[pos].Term          }            a := l.entries[pos:]          if len(a) == 0 {           return []logEntry{}, lastTerm          }    // 除去command Response channel          return stripResponseChannels(a), lastTerm  }    func stripResponseChannels(a []logEntry)[]logEntry {          stripped := make([]logEntry, len(a))          for i, entry := range a {           stripped[i] = logEntry{            Index:           entry.Index,            Term:            entry.Term,            Command:         entry.Command,            commandResponse: nil,           }          }          return stripped  }    // contains returns true if a log entry with the given index and term exists in  // the log.  // 判断是够包含{term, index}条目  func (l *raftLog)contains(index, termuint64)bool {          l.RLock()          defer l.RUnlock()            // It's not necessarily true that l.entries[i] has index == i.          for _, entry := range l.entries {           if entry.Index == index && entry.Term == term {            return true           }           if entry.Index > index || entry.Term > term {            break           }          }          return false  }    // 判断{term, index}是否为最新的日志条目，如果是,则将则将在其之后的日志清理掉,  // 这个条目应该在[commit_index, last_index]范围内  func (l *raftLog)ensureLastIs(index, termuint64)error {          l.Lock()          defer l.Unlock()            // Taken loosely from benbjohnson's impl            if index < l.getCommitIndexWithLock() {           return errIndexTooSmall          }            if index > l.lastIndexWithLock() {           return errIndexTooBig          }            // It's possible that the passed index is 0. It means the leader has come to          // decide we need a complete log rebuild. Of course, that's only valid if we          // haven't committed anything, so this check comes after that one.      // 全部重建，前提是没有commited过任何的条目          if index == 0 {           for pos := 0; pos < len(l.entries); pos++ {            if l.entries[pos].commandResponse != nil {             close(l.entries[pos].commandResponse)             l.entries[pos].commandResponse = nil            }            if l.entries[pos].committed != nil {             l.entries[pos].committed <- false             close(l.entries[pos].committed)             l.entries[pos].committed = nil            }           }           l.entries = []logEntry{}           return nil          }            // Normal case: find the position of the matching log entry.          pos := 0          for ; pos < len(l.entries); pos++ {           if l.entries[pos].Index < index {            continue // didn't find it yet           }           if l.entries[pos].Index > index {            return errBadIndex // somehow went past it           }           if l.entries[pos].Index != index {            panic("not <, not >, but somehow !=")           }           if l.entries[pos].Term != term {            return errBadTerm           }           break // good          }            // Sanity check.          // ? 怎么可能出现这种情况？          if pos < l.commitPos {           panic("index >= commitIndex, but pos < commitPos")          }            // `pos` is the position of log entry matching index and term.          // We want to truncate everything after that.      // 应为{term, index}是最新的了，所以将在其之后的所有条目给cut掉          truncateFrom := pos + 1          if truncateFrom >= len(l.entries) {           return nil // nothing to truncate          }            // If we blow away log entries that haven't yet sent responses to clients,          // signal the clients to stop waiting, by closing the channel without a          // response value.          for pos = truncateFrom; pos < len(l.entries); pos++ {           if l.entries[pos].commandResponse != nil {            close(l.entries[pos].commandResponse)            l.entries[pos].commandResponse = nil           }           if l.entries[pos].committed != nil {            l.entries[pos].committed <- false            close(l.entries[pos].committed)            l.entries[pos].committed = nil           }          }            // Truncate the log.          l.entries = l.entries[:truncateFrom]            // Done.          return nil  }    // getCommitIndex returns the commit index of the log. That is, the index of the  // last log entry which can be considered committed.  // 获取最新的commited日志条目  func (l *raftLog)getCommitIndex()uint64 {          l.RLock()          defer l.RUnlock()          return l.getCommitIndexWithLock()  }    // 获取最新的日志条目  func (l *raftLog)getCommitIndexWithLock()uint64 {          if l.commitPos < 0 {           return 0          }          if l.commitPos >= len(l.entries) {           panic(fmt.Sprintf("commitPos %d > len(l.entries) %d; bad bookkeeping in raftLog", l.commitPos, len(l.entries)))          }          return l.entries[l.commitPos].Index  }    // lastIndex returns the index of the most recent log entry.  func (l *raftLog)lastIndex()uint64 {          l.RLock()          defer l.RUnlock()          return l.lastIndexWithLock()  }    func (l *raftLog)lastIndexWithLock()uint64 {          if len(l.entries) <= 0 {           return 0          }          return l.entries[len(l.entries)-1].Index  }    // lastTerm returns the term of the most recent log entry.  func (l *raftLog)lastTerm()uint64 {          l.RLock()          defer l.RUnlock()          return l.lastTermWithLock()  }    func (l *raftLog)lastTermWithLock()uint64 {          if len(l.entries) <= 0 {           return 0          }          return l.entries[len(l.entries)-1].Term  }    // appendEntry appends the passed log entry to the log. It will return an error  // if the entry's term is smaller than the log's most recent term, or if the  // entry's index is too small relative to the log's most recent entry.  // 追加日志，注意此时还没有commit该条目  func (l *raftLog)appendEntry(entry logEntry)error {          l.Lock()          defer l.Unlock()      // 判定{entry.term, entry.index} > {last_term, last_index}          if len(l.entries) > 0 {           lastTerm := l.lastTermWithLock()           if entry.Term < lastTerm {            return errTermTooSmall           }           lastIndex := l.lastIndexWithLock()           if entry.Term == lastTerm && entry.Index <= lastIndex {            return errIndexTooSmall           }          }            l.entries = append(l.entries, entry)          return nil  }    // commitTo commits all log entries up to and including the passed commitIndex.  // Commit means: synchronize the log entry to persistent storage, and call the  // state machine apply function for the log entry's command.  // 注意:  // 1、commit是一个后端任务，再此并没有"1/2"确认的概念(实际上是不是这样呢，这得去参考raft的论文了)  // 2、apply函数是在commit过程中调用，而不是在append的时候调用  // 3、apply相当于状态机函数，一般用户会将这些操作结果保存起来，用于快照    // 比如，想实现一个kv存储，那么用户只要将kv相关的逻辑植入这个函数即可    // committed <= commitIndex <= last_index  func (l *raftLog)commitTo(commitIndexuint64)error {          if commitIndex == 0 {           panic("commitTo(0)")          }            l.Lock()          defer l.Unlock()            // Reject old commit indexes          if commitIndex < l.getCommitIndexWithLock() {           return errIndexTooSmall          }            // Reject new commit indexes          if commitIndex > l.lastIndexWithLock() {           return errIndexTooBig          }            // If we've already committed to the commitIndex, great!          if commitIndex == l.getCommitIndexWithLock() {           return nil          }            // We should start committing at precisely the last commitPos + 1          pos := l.commitPos + 1          if pos < 0 {           panic("pending commit pos < 0")          }            // Commit entries between our existing commit index and the passed index.          // Remember to include the passed index.          for {           // Sanity checks. TODO replace with plain `for` when this is stable.           if pos >= len(l.entries) {            panic(fmt.Sprintf("commitTo pos=%d advanced past all log entries (%d)", pos, len(l.entries)))           }           if l.entries[pos].Index > commitIndex {            panic("commitTo advanced past the desired commitIndex")           }             // Encode the entry to persistent storage.           if err := l.entries[pos].encode(l.store); err != nil {            return err           }             // Forward non-configuration commands to the state machine.           // Send the responses to the waiting client, if applicable.           // 如果不是配置类型的Log，则调用apply function           // 配置类型的Log，在其他地方处理           if !l.entries[pos].isConfiguration {            resp := l.apply(l.entries[pos].Index, l.entries[pos].Command)            if l.entries[pos].commandResponse != nil {             select {             case l.entries[pos].commandResponse <- resp:              break          // 问什么选取这个时间？？？             case <-time.After(maximumElectionTimeout()): // << ElectionInterval              panic("uncoöperative command response receiver")             }             close(l.entries[pos].commandResponse)             l.entries[pos].commandResponse = nil            }           }             // Signal the entry has been committed, if applicable.           if l.entries[pos].committed != nil {            l.entries[pos].committed <- true            close(l.entries[pos].committed)            l.entries[pos].committed = nil           }             // Mark our commit position cursor.           l.commitPos = pos             // If that was the last one, we're done.           if l.entries[pos].Index == commitIndex {            break           }           if l.entries[pos].Index > commitIndex {            panic(fmt.Sprintf(             "current entry Index %d is beyond our desired commitIndex %d",             l.entries[pos].Index,             commitIndex,            ))           }             // Otherwise, advance!           pos++          }            // Done.          return nil  }    // logEntry is the atomic unit being managed by the distributed log. A log entry  // always has an index (monotonically increasing), a term in which the Raft  // network leader first sees the entry, and a command. The command is what gets  // executed against the node state machine when the log entry is successfully  // replicated.  type logEntry struct {      // 日志索引号          Index           uint64        `json:"index"`          // 任期号      Term            uint64        `json:"term"` // when received by leader      // 日志内容          Command         []byte        `json:"command,omitempty"`      // commited 通道          committed       chan bool     `json:"-"`      // 命令应答 通道          commandResponse chan<- []byte `json:"-"` // only non-nil on receiver's log      // 日志类型标示          isConfiguration bool          `json:"-"` // for configuration change entries  }    // encode serializes the log entry to the passed io.Writer.  //  // Entries are serialized in a simple binary format:  //  // ---------------------------------------------  // | uint32 | uint64 | uint64 | uint32 | []byte |  // ---------------------------------------------  // | CRC | TERM | INDEX | SIZE | COMMAND |  // ---------------------------------------------  //    // 序列化，大端  func (e *logEntry)encode(w io.Writer)error {          if len(e.Command) <= 0 {           return errNoCommand          }          if e.Index <= 0 {           return errBadIndex          }          if e.Term <= 0 {           return errBadTerm          }            commandSize := len(e.Command)          buf := make([]byte, 24+commandSize)            binary.LittleEndian.PutUint64(buf[4:12], e.Term)          binary.LittleEndian.PutUint64(buf[12:20], e.Index)          binary.LittleEndian.PutUint32(buf[20:24], uint32(commandSize))            copy(buf[24:], e.Command)            binary.LittleEndian.PutUint32(           buf[0:4],           crc32.ChecksumIEEE(buf[4:]),          )            _, err := w.Write(buf)          return err  }    // 反序列化  // decode deserializes one log entry from the passed io.Reader.  func (e *logEntry)decode(r io.Reader)error {          header := make([]byte, 24)            if _, err := r.Read(header); err != nil {           return err          }            command := make([]byte, binary.LittleEndian.Uint32(header[20:24]))            if _, err := r.Read(command); err != nil {           return err          }            crc := binary.LittleEndian.Uint32(header[:4])            check := crc32.NewIEEE()          check.Write(header[4:])          check.Write(command)            if crc != check.Sum32() {           return errInvalidChecksum          }            e.Term = binary.LittleEndian.Uint64(header[4:12])          e.Index = binary.LittleEndian.Uint64(header[12:20])          e.Command = command            return nil  }  </code></pre>    <h3>Peers.go</h3>    <pre>  <code class="language-go">var (   errTimeout = errors.New("timeout")  )  // peers为节点的一个抽象，对外提供了一些访问接口，  // 需要注意的地方是peers的序列化  type Peer interface {     // 返回server标示   id() uint64     // 日志追加接口   callAppendEntries(appendEntries) appendEntriesResponse     // 投票选举接口   callRequestVote(requestVote) requestVoteResponse     // 命令调用   callCommand([]byte, chan<- []byte) error     // 集群配置变化接口   callSetConfiguration(...Peer) error  }    // localPeer is the simplest kind of peer, mapped to a server in the  // same process-space. Useful for testing and demonstration; not so  // useful for networks of independent processes.  // 本地local peers，用于测试，不用经过网络  type localPeer struct {   server *Server  }    func newLocalPeer(server *Server)*localPeer { return &localPeer{server} }    func (p *localPeer)id()uint64 { return p.server.id }    // 追加日志  func (p *localPeer)callAppendEntries(ae appendEntries)appendEntriesResponse {   return p.server.appendEntries(ae)  }    // 投票选举  func (p *localPeer)callRequestVote(rv requestVote)requestVoteResponse {   return p.server.requestVote(rv)  }    // 命令  // 实际调用为Leader  func (p *localPeer)callCommand(cmd []byte, responsechan<- []byte)error {   return p.server.Command(cmd, response)  }    // 设置配置  func (p *localPeer)callSetConfiguration(peers ...Peer)error {   return p.server.SetConfiguration(peers...)  }    // requestVoteTimeout issues the requestVote to the given peer.  // If no response is received before timeout, an error is returned.  // 投票  func requestVoteTimeout(p Peer, rv requestVote, timeout time.Duration)(requestVoteResponse, error) {   c := make(chan requestVoteResponse, 1)   go func() { c <- p.callRequestVote(rv) }()     select {   case resp := <-c:    return resp, nil   case <-time.After(timeout):    return requestVoteResponse{}, errTimeout   }  }    // peerMap is a collection of Peer interfaces. It provides some convenience  // functions for actions that should apply to multiple Peers.  type peerMap map[uint64]Peer    // makePeerMap constructs a peerMap from a list of peers.  func makePeerMap(peers ...Peer)peerMap {   pm := peerMap{}   for _, peer := range peers {    pm[peer.id()] = peer   }   return pm  }    // explodePeerMap converts a peerMap into a slice of peers.  func explodePeerMap(pm peerMap)[]Peer {   a := []Peer{}   for _, peer := range pm {    a = append(a, peer)   }   return a  }    func (pm peerMap)except(iduint64)peerMap {   except := peerMap{}   for id0, peer := range pm {    if id0 == id {     continue    }    except[id0] = peer   }   return except  }    func (pm peerMap)count()int { return len(pm) }    // 法定人数  func (pm peerMap)quorum()int {   switch n := len(pm); n {   case 0, 1:    return 1   default:    return (n / 2) + 1   }  }    // requestVotes sends the passed requestVote RPC to every peer in Peers. It  // forwards responses along the returned requestVoteResponse channel. It makes  // the RPCs with a timeout of BroadcastInterval * 2 (chosen arbitrarily). Peers  // that don't respond within the timeout are retried forever. The retry loop  // stops only when all peers have responded, or a Cancel signal is sent via the  // returned canceler.  func (pm peerMap)requestVotes(r requestVote)(chanvoteResponseTuple, canceler) {   // "[A server entering the candidate stage] issues requestVote RPCs in   // parallel to each of the other servers in the cluster. If the candidate   // receives no response for an RPC, it reissues the RPC repeatedly until a   // response arrives or the election concludes."     // construct the channels we'll return   abortChan := make(chan struct{})   tupleChan := make(chan voteResponseTuple)     go func() {    // We loop until all Peers have given us a response.    // Track which Peers have responded.    respondedAlready := peerMap{} // none yet      for {     notYetResponded := disjoint(pm, respondedAlready)     if len(notYetResponded) <= 0 {      return // done     }       // scatter     tupleChan0 := make(chan voteResponseTuple, len(notYetResponded))     for id, peer := range notYetResponded {      go func(iduint64, peer Peer) {       resp, err := requestVoteTimeout(peer, r, 2*maximumElectionTimeout())       tupleChan0 <- voteResponseTuple{id, resp, err}      }(id, peer)     }       // gather     for i := 0; i < cap(tupleChan0); i++ {      select {      case t := <-tupleChan0:       if t.err != nil {        continue // will need to retry       }       respondedAlready[t.id] = nil // set membership semantics       tupleChan <- t        case <-abortChan:       return // give up      }     }    }   }()     return tupleChan, cancel(abortChan)  }    // 选举应答  type voteResponseTuple struct {   id       uint64   response requestVoteResponse   err      error  }    type canceler interface {   Cancel()  }    type cancel chan struct{}    func (c cancel)Cancel() { close(c) }    // 过滤peers  func disjoint(all, except peerMap)peerMap {   d := peerMap{}   for id, peer := range all {    if _, ok := except[id]; ok {     continue    }    d[id] = peer   }   return d  }  </code></pre>    <h3>server.go</h3>    <p>这是最重要的一个逻辑</p>    <p><img src="https://simg.open-open.com/show/0ce67b467bfb0c5540095525dbfa59ec.png"></p>    <p>几点配置变更</p>    <p><img src="https://simg.open-open.com/show/19442432774f8731e27646368e69c044.png"></p>    <pre>  <code class="language-go">// 角色分类   const (   follower  = "Follower"   candidate = "Candidate"   leader    = "Leader"  )    const (   unknownLeader = 0   noVote        = 0  )    // 选举时间随机范围[MinimumElectionTimeoutMS, maximumElectionTimeoutMS]  var (   MinimumElectionTimeoutMS int32 = 250     maximumElectionTimeoutMS = 2 * MinimumElectionTimeoutMS  )    var (   errNotLeader             = errors.New("not the leader")   errUnknownLeader         = errors.New("unknown leader")   errDeposed               = errors.New("deposed during replication")   errAppendE#008000ntriesRejected = errors.New("appendEntries RPC rejected")   errReplicationFailed     = errors.New("command replication failed (but will keep retrying)")   errOutOfSync             = errors.New("out of sync")   errAlreadyRunning        = errors.New("already running")  )    // 重置选举时间  func resetElectionTimeoutMS(newMin, newMaxint)(int,int) {   oldMin := atomic.LoadInt32(&MinimumElectionTimeoutMS)   oldMax := atomic.LoadInt32(&maximumElectionTimeoutMS)   atomic.StoreInt32(&MinimumElectionTimeoutMS, int32(newMin))   atomic.StoreInt32(&maximumElectionTimeoutMS, int32(newMax))   return int(oldMin), int(oldMax)  }    // minimumElectionTimeout returns the current minimum election timeout.  func minimumElectionTimeout()time.Duration {   return time.Duration(MinimumElectionTimeoutMS) * time.Millisecond  }    // maximumElectionTimeout returns the current maximum election time.  func maximumElectionTimeout()time.Duration {   return time.Duration(maximumElectionTimeoutMS) * time.Millisecond  }    // 选举时间随机函数  func electionTimeout()time.Duration {   n := rand.Intn(int(maximumElectionTimeoutMS - MinimumElectionTimeoutMS))   d := int(MinimumElectionTimeoutMS) + n   return time.Duration(d) * time.Millisecond  }    // broadcastInterval returns the interval between heartbeats (AppendEntry RPCs)  // broadcast from the leader. It is the minimum election timeout / 10, as  // dictated by the spec: BroadcastInterval << ElectionTimeout << MTBF.  // 广播时间，用于Leader发送心跳广播，这个时间应小于选举时间；否则，非Leader节点会产生选举操作  func broadcastInterval()time.Duration {   d := MinimumElectionTimeoutMS / 10   return time.Duration(d) * time.Millisecond  }    // protectedString is just a string protected by a mutex.  type protectedString struct {   sync.RWMutex   value string  }    func (s *protectedString)Get()string {   s.RLock()   defer s.RUnlock()   return s.value  }    func (s *protectedString)Set(valuestring) {   s.Lock()   defer s.Unlock()   s.value = value  }    // protectedBool is just a bool protected by a mutex.  type protectedBool struct {   sync.RWMutex   value bool  }    func (s *protectedBool)Get()bool {   s.RLock()   defer s.RUnlock()   return s.value  }    func (s *protectedBool)Set(valuebool) {   s.Lock()   defer s.Unlock()   s.value = value  }    // Server is the agent that performs all of the Raft protocol logic.  // In a typical application, each running process that wants to be part of  // the distributed state machine will contain a server component.  type Server struct {   id      uint64 // id of this server      // 节点状态   state   *protectedString      // 节点运行状态   running *protectedBool      // Leader节点标示   leader  uint64       // 当前节点任期号   term    uint64 // "current term number, which increases monotonically"      // 0表示，当前节点还有投出自己的票;非零表示节点已经投票了，值是获票者的标示ID   vote    uint64 // who we voted for this term, if applicable   log     *raftLog   config  *configuration        // 追加日志信道   appendEntriesChan chan appendEntriesTuple     // 投票信道   requestVoteChan   chan requestVoteTuple     // 命令信道   commandChan       chan commandTuple     // 配置修改信道   configurationChan chan configurationTuple       // 选举信道   electionTick <-chan time.Time     // 退出信道   quit         chan chan struct{}  }    // 状态机函数  // 该函数不可并发执行，否则就达不到一致性状态机的效果(执行时间不要超过选举时间)    // 正常来说，只有"共识"达成的时候，才会调用该函数，然后返回给客户端  // 但是，在这里为了简化实现，"共识“算法是放在后台任务操作的，客户端发送命令单Leader时，Leader马上  // 应答客户端，并没有等”共识算法“的共识结果  type ApplyFunc func(commitIndexuint64, cmd []byte)[]byte    // 初始化节点  // 1. 构建日志 2.初始化为"follower"角色 3.leader为"unknown"  func NewServer(iduint64, store io.ReadWriter, a ApplyFunc) *Server {   if id <= 0 {    panic("server id must be > 0")   }     // 5.2 Leader election: "the latest term this server has seen is persisted,   // and is initialized to 0 on first boot."   log := newRaftLog(store, a)   latestTerm := log.lastTerm()     s := &Server{    id:      id,    state:   &protectedString{value: follower}, // "when servers start up they begin as followers"    running: &protectedBool{value: false},    leader:  unknownLeader, // unknown at startup    log:     log,    term:    latestTerm,    config:  newConfiguration(peerMap{}),      appendEntriesChan: make(chan appendEntriesTuple),    requestVoteChan:   make(chan requestVoteTuple),    commandChan:       make(chan commandTuple),    configurationChan: make(chan configurationTuple),      electionTick: nil,    quit:         make(chan chan struct{}),   }   s.resetElectionTimeout()   return s  }    type configurationTuple struct {   Peers []Peer   Err   chan error  }    // 设置配置  // 1. 服务启动时，先设置配置  // 2. 集群变更时，设置配置  func (s *Server)SetConfiguration(peers ...Peer)error {      // 节点刚启动   if !s.running.Get() {    s.config.directSet(makePeerMap(peers...))    return nil   }     err := make(chan error)      // 节点已经启动了   s.configurationChan <- configurationTuple{peers, err}   return <-err  }    // Start triggers the server to begin communicating with its peers.  func (s *Server)Start() {   go s.loop()  }    // Stop terminates the server. Stopped servers should not be restarted.  func (s *Server)Stop() {   q := make(chan struct{})   s.quit <- q   <-q   s.logGeneric("server stopped")  }    // 命令元组  type commandTuple struct {     // 命令内容   Command         []byte     // 命令信道   CommandResponse chan<- []byte   Err             chan error  }    // 命令接口  func (s *Server)Command(cmd []byte, responsechan<- []byte)error {   err := make(chan error)   s.commandChan <- commandTuple{cmd, response, err}   return <-err  }    // 日志追加  func (s *Server)appendEntries(ae appendEntries)appendEntriesResponse {   t := appendEntriesTuple{    Request:  ae,    Response: make(chan appendEntriesResponse),   }   s.appendEntriesChan <- t   return <-t.Response  }    // 投票  func (s *Server)requestVote(rv requestVote)requestVoteResponse {   t := requestVoteTuple{    Request:  rv,    Response: make(chan requestVoteResponse),   }   s.requestVoteChan <- t   return <-t.Response  }    // times out,  // new election  // | .-----.  // | | |  // v times out, | v receives votes from  // +----------+ starts election +-----------+ majority of servers +--------+  // | Follower |------------------>| Candidate |---------------------->| Leader |  // +----------+ +-----------+ +--------+  // ^ ^ | |  // | | discovers current leader | |  // | | or new term | |  // | '------------------------------' |  // | |  // | discovers server with higher term |  // '------------------------------------------------------------------'  //  //    func (s *Server)loop() {   s.running.Set(true)   for s.running.Get() {    switch state := s.state.Get(); state {    case follower:     s.followerSelect()    case candidate:     s.candidateSelect()    case leader:     s.leaderSelect()    default:     panic(fmt.Sprintf("unknown Server State '%s'", state))    }   }  }    func (s *Server)resetElectionTimeout() {   s.electionTick = time.NewTimer(electionTimeout()).C  }    func (s *Server)logGeneric(formatstring, args ...interface{}) {   prefix := fmt.Sprintf("id=%d term=%d state=%s: ", s.id, s.term, s.state.Get())   log.Printf(prefix+format, args...)  }    func (s *Server)logAppendEntriesResponse(req appendEntries, resp appendEntriesResponse, stepDownbool) {   s.logGeneric(    "got appendEntries, sz=%d leader=%d prevIndex/Term=%d/%d commitIndex=%d: responded with success=%v (reason='%s') stepDown=%v",    len(req.Entries),    req.LeaderID,    req.PrevLogIndex,    req.PrevLogTerm,    req.CommitIndex,    resp.Success,    resp.reason,    stepDown,   )  }    func (s *Server)logRequestVoteResponse(req requestVote, resp requestVoteResponse, stepDownbool) {   s.logGeneric(    "got RequestVote, candidate=%d: responded with granted=%v (reason='%s') stepDown=%v",    req.CandidateID,    resp.VoteGranted,    resp.reason,    stepDown,   )  }    func (s *Server)handleQuit(qchan struct{}) {   s.logGeneric("got quit signal")   s.running.Set(false)   close(q)  }    // 命令转发  // 如果当前节点不是Leader节点，并且已存在Leader节点，则其会以"代理“的角色，将命令转发至Leader节点  func (s *Server)forwardCommand(t commandTuple) {   switch s.leader {   case unknownLeader:    s.logGeneric("got command, but don't know leader")    t.Err <- errUnknownLeader     case s.id: // I am the leader    panic("impossible state in forwardCommand")     default:    leader, ok := s.config.get(s.leader)    if !ok {     panic("invalid state in peers")    }    s.logGeneric("got command, forwarding to leader (%d)", s.leader)    // We're blocking our {follower,candidate}Select function in the    // receive-command branch. If we continue to block while forwarding    // the command, the leader won't be able to get a response from us!    go func() { t.Err <- leader.callCommand(t.Command, t.CommandResponse) }()   }  }    // 配置变更  // 转发规则和命令转发一样  func (s *Server)forwardConfiguration(t configurationTuple) {   switch s.leader {   case unknownLeader:    s.logGeneric("got configuration, but don't know leader")    t.Err <- errUnknownLeader     case s.id: // I am the leader    panic("impossible state in forwardConfiguration")     default:    leader, ok := s.config.get(s.leader)    if !ok {     panic("invalid state in peers")    }    s.logGeneric("got configuration, forwarding to leader (%d)", s.leader)    go func() { t.Err <- leader.callSetConfiguration(t.Peers...) }()   }  }    // follower 节点逻辑  func (s *Server)followerSelect() {   for {    select {    case q := <-s.quit:     s.handleQuit(q)     return    // 命令转发    case t := <-s.commandChan:     s.forwardCommand(t)    // 集群变更转发    case t := <-s.configurationChan:     s.forwardConfiguration(t)    // Leader选举    case <-s.electionTick:     // 5.2 Leader election: "A follower increments its current term and     // transitions to candidate state."     if s.config == nil {      s.logGeneric("election timeout, but no configuration: ignoring")      s.resetElectionTimeout()      continue     }     s.logGeneric("election timeout, becoming candidate")             // 提高自己的任期号     s.term++             // 投票置为空     s.vote = noVote             // Leader     s.leader = unknownLeader             // 设置节点角色为"候选人"     s.state.Set(candidate)             // 重置选举时间，防止马上再次出发选举     s.resetElectionTimeout()     return          // 日志追加(除了客户端请求，leader的心跳也会出发这个行为)    case t := <-s.appendEntriesChan:     if s.leader == unknownLeader {      s.leader = t.Request.LeaderID      s.logGeneric("discovered Leader %d", s.leader)     }             // 处理日志最佳操作     resp, stepDown := s.handleAppendEntries(t.Request)     s.logAppendEntriesResponse(t.Request, resp, stepDown)     t.Response <- resp             // 如果节点已经脱离了当前的集群，需要跟新Leader地址     if stepDown {      // stepDown as a Follower means just to reset the leader      if s.leader != unknownLeader {       s.logGeneric("abandoning old leader=%d", s.leader)      }      s.logGeneric("following new leader=%d", t.Request.LeaderID)      s.leader = t.Request.LeaderID     }    // 选举    case t := <-s.requestVoteChan:             // 选举处理     resp, stepDown := s.handleRequestVote(t.Request)     s.logRequestVoteResponse(t.Request, resp, stepDown)     t.Response <- resp             // 如果落后于当前节点了，把当前的Leader修改为"unkownleader"，等待讯据成功后，进行切换     if stepDown {      // stepDown as a Follower means just to reset the leader      if s.leader != unknownLeader {       s.logGeneric("abandoning old leader=%d", s.leader)      }      s.logGeneric("new leader unknown")      s.leader = unknownLeader     }    }   }  }    // 候选状态  func (s *Server)candidateSelect() {   if s.leader != unknownLeader {    panic("known leader when entering candidateSelect")   }   if s.vote != 0 {    panic("existing vote when entering candidateSelect")   }     // "[A server entering the candidate stage] issues requestVote RPCs in   // parallel to each of the other servers in the cluster. If the candidate   // receives no response for an RPC, it reissues the RPC repeatedly until a   // response arrives or the election concludes."   // 发起选举RPC   requestVoteResponses, canceler := s.config.allPeers().except(s.id).requestVotes(requestVote{    Term:         s.term,    CandidateID:  s.id,    LastLogIndex: s.log.lastIndex(),    LastLogTerm:  s.log.lastTerm(),   })   defer canceler.Cancel()     // Set up vote tallies (plus, vote for myself)   votes := map[uint64]bool{s.id: true}   s.vote = s.id   s.logGeneric("term=%d election started (configuration state %s)", s.term, s.config.state)     // 如果已经达到了选举“共识”，则成功选举   if s.config.pass(votes) {    s.logGeneric("I immediately won the election")    s.leader = s.id    s.state.Set(leader)    s.vote = noVote    return   }     // "A candidate continues in this state until one of three things happens:   // (a) it wins the election, (b) another server establishes itself as   // leader, or (c) a period of time goes by with no winner."   for {    select {    case q := <-s.quit:     s.handleQuit(q)     return    // 命令转发    case t := <-s.commandChan:     s.forwardCommand(t)    // 配置更新转发，注意和Leader的不同    case t := <-s.configurationChan:     s.forwardConfiguration(t)    // 收到选举的应答    case t := <-requestVoteResponses:     s.logGeneric("got vote: id=%d term=%d granted=%v", t.id, t.response.Term, t.response.VoteGranted)     // "A candidate wins the election if it receives votes from a     // majority of servers in the full cluster for the same term."             // 本节点落后于其他几点     if t.response.Term > s.term {      s.logGeneric("got vote from future term (%d>%d); abandoning election", t.response.Term, s.term)      s.leader = unknownLeader      s.state.Set(follower)      s.vote = noVote      return // lose     }             // 收到了"落后"当前节点的应答，忽略掉它     if t.response.Term < s.term {      s.logGeneric("got vote from past term (%d<%d); ignoring", t.response.Term, s.term)      break     }                          // 收到赞同票     if t.response.VoteGranted {      s.logGeneric("%d voted for me", t.id)      votes[t.id] = true     }     // "Once a candidate wins an election, it becomes leader."             // “共识”达成     if s.config.pass(votes) {      s.logGeneric("I won the election")      s.leader = s.id      s.state.Set(leader)      s.vote = noVote      return // win     }            // 收到日志追加(在这里，心跳也当做日志追加的方式发送)    case t := <-s.appendEntriesChan:     // "While waiting for votes, a candidate may receive an     // appendEntries RPC from another server claiming to be leader.     // If the leader's term (included in its RPC) is at least as     // large as the candidate's current term, then the candidate     // recognizes the leader as legitimate and steps down, meaning     // that it returns to follower state."              // 处理日志     resp, stepDown := s.handleAppendEntries(t.Request)     s.logAppendEntriesResponse(t.Request, resp, stepDown)     t.Response <- resp             // candidate节点落后于Leader节点     if stepDown {      s.logGeneric("after an appendEntries, stepping down to Follower (leader=%d)", t.Request.LeaderID)      s.leader = t.Request.LeaderID      s.state.Set(follower)      return // lose     }            // 虽然当前节点是candidate节点，但集群中此时可能存在多个candidate节点    case t := <-s.requestVoteChan:     // We can also be defeated by a more recent candidate     resp, stepDown := s.handleRequestVote(t.Request)     s.logRequestVoteResponse(t.Request, resp, stepDown)     t.Response <- resp     if stepDown {                 // 当前candidate节点落后于集群中已存在的candidate节点，将自己的角色变为follower，                 // 并且也会投赞同票      s.logGeneric("after a requestVote, stepping down to Follower (leader unknown)")      s.leader = unknownLeader      s.state.Set(follower)      return // lose     }              // 选举    case <-s.electionTick:     // "The third possible outcome is that a candidate neither wins nor     // loses the election: if many followers become candidates at the     // same time, votes could be split so that no candidate obtains a     // majority. When this happens, each candidate will start a new     // election by incrementing its term and initiating another round of     // requestVote RPCs."     s.logGeneric("election ended with no winner; incrementing term and trying again")     s.resetElectionTimeout()     s.term++     s.vote = noVote     return // draw    }   }  }    // Leader 保存的Follower节点的所有最新同步条目  type nextIndex struct {   sync.RWMutex   m map[uint64]uint64 // followerId: nextIndex  }    func newNextIndex(pm peerMap, defaultNextIndexuint64)*nextIndex {   ni := &nextIndex{    m: map[uint64]uint64{},   }   for id := range pm {    ni.m[id] = defaultNextIndex   }   return ni  }    // 找出已经同步Follower的最小日志  func (ni *nextIndex)bestIndex()uint64 {   ni.RLock()   defer ni.RUnlock()     if len(ni.m) <= 0 {    return 0   }     i := uint64(math.MaxUint64)   for _, nextIndex := range ni.m {    if nextIndex < i {     i = nextIndex    }   }   return i  }    // 返回节点(id)最新的同步日志  func (ni *nextIndex)prevLogIndex(iduint64)uint64 {   ni.RLock()   defer ni.RUnlock()   if _, ok := ni.m[id]; !ok {    panic(fmt.Sprintf("peer %d not found", id))   }   return ni.m[id]  }    // 自减节点(id)的最新同步日志，用于同步失败时的回滚  func (ni *nextIndex)decrement(iduint64, prevuint64)(uint64, error) {   ni.Lock()   defer ni.Unlock()     i, ok := ni.m[id]   if !ok {    panic(fmt.Sprintf("peer %d not found", id))   }     if i != prev {    return i, errOutOfSync   }     if i > 0 {    ni.m[id]--   }   return ni.m[id], nil  }    // 更新节点(id)的同步日志  func (ni *nextIndex)set(id, index, prevuint64)(uint64, error) {   ni.Lock()   defer ni.Unlock()     i, ok := ni.m[id]   if !ok {    panic(fmt.Sprintf("peer %d not found", id))   }   if i != prev {    return i, errOutOfSync   }     ni.m[id] = index   return index, nil  }    // 心跳、复制命令都会用到该函数，flush是同步的，如果对端节点不可达，则阻塞  func (s *Server)flush(peer Peer, ni *nextIndex)error {   peerID := peer.id()   // Leader的任期号   currentTerm := s.term   // 节点(peer)的最新同步索引   prevLogIndex := ni.prevLogIndex(peerID)   // 检索出peers节点落后于Leader几点的日志条目，然后进行同步   entries, prevLogTerm := s.log.entriesAfter(prevLogIndex)   // 获取Leader committed的最新索引   commitIndex := s.log.getCommitIndex()   s.logGeneric("flush to %d: term=%d leaderId=%d prevLogIndex/Term=%d/%d sz=%d commitIndex=%d", peerID, currentTerm, s.id, prevLogIndex, prevLogTerm, len(entries), commitIndex)      // 日志追加RPC   resp := peer.callAppendEntries(appendEntries{    Term:         currentTerm,    LeaderID:     s.id,    PrevLogIndex: prevLogIndex,    PrevLogTerm:  prevLogTerm,    Entries:      entries,    CommitIndex:  commitIndex,   })     if resp.Term > currentTerm {    // 应答的节点比当前节点的任期号大，当前的Leader被罢免    s.logGeneric("flush to %d: responseTerm=%d > currentTerm=%d: deposed", peerID, resp.Term, currentTerm)    return errDeposed   }        if !resp.Success {    // 应答失败，可能是leader RPC等待超时，或者出现了网络错误(包括脑裂)，回滚    newPrevLogIndex, err := ni.decrement(peerID, prevLogIndex)    if err != nil {     s.logGeneric("flush to %d: while decrementing prevLogIndex: %s", peerID, err)     return err    }    s.logGeneric("flush to %d: rejected; prevLogIndex(%d) becomes %d", peerID, peerID, newPrevLogIndex)    return errAppendEntriesRejected   }     if len(entries) > 0 {    // 复制成功，更新同步状态    newPrevLogIndex, err := ni.set(peer.id(), entries[len(entries)-1].Index, prevLogIndex)    if err != nil {     s.logGeneric("flush to %d: while moving prevLogIndex forward: %s", peerID, err)     return err    }    s.logGeneric("flush to %d: accepted; prevLogIndex(%d) becomes %d", peerID, peerID, newPrevLogIndex)    return nil   }     s.logGeneric("flush to %d: accepted; prevLogIndex(%d) remains %d", peerID, peerID, ni.prevLogIndex(peerID))   return nil  }    // Leader并发同步日志  func (s *Server)concurrentFlush(pm peerMap, ni *nextIndex, timeout time.Duration)(int,bool) {   type tuple struct {    id  uint64    err error   }   responses := make(chan tuple, len(pm))   for _, peer := range pm {    go func(peer Peer) {     errChan := make(chan error, 1)     go func() { errChan <- s.flush(peer, ni) }()     go func() { time.Sleep(timeout); errChan <- errTimeout }()     responses <- tuple{peer.id(), <-errChan} // first responder wins    }(peer)   }     successes, stepDown := 0, false   for i := 0; i < cap(responses); i++ {    switch t := <-responses; t.err {    case nil:     s.logGeneric("concurrentFlush: peer %d: OK (prevLogIndex(%d)=%d)", t.id, t.id, ni.prevLogIndex(t.id))     successes++    case errDeposed:     // 当前的Leder节点落后于其他节点     s.logGeneric("concurrentFlush: peer %d: deposed!", t.id)     stepDown = true    default:     s.logGeneric("concurrentFlush: peer %d: %s (prevLogIndex(%d)=%d)", t.id, t.err, t.id, ni.prevLogIndex(t.id))     // nothing to do but log and continue    }   }   return successes, stepDown  }    // 作为Leader角色运行  func (s *Server)leaderSelect() {   if s.leader != s.id {    panic(fmt.Sprintf("leader (%d) not me (%d) when entering leaderSelect", s.leader, s.id))   }   if s.vote != 0 {    panic(fmt.Sprintf("vote (%d) not zero when entering leaderSelect", s.leader))   }     // 5.3 Log replication: "The leader maintains a nextIndex for each follower,   // which is the index of the next log entry the leader will send to that   // follower. When a leader first comes to power it initializes all nextIndex   // values to the index just after the last one in its log."   //   // I changed this from lastIndex+1 to simply lastIndex. Every initial   // communication from leader to follower was being rejected and we were   // doing the decrement. This was just annoying, except if you manage to   // sneak in a command before the first heartbeat. Then, it will never get   // properly replicated (it seemed).      // Leader为每个Follower保存了最新的同步日志索引   ni := newNextIndex(s.config.allPeers().except(s.id), s.log.lastIndex()) // +1)     flush := make(chan struct{})   heartbeat := time.NewTicker(broadcastInterval())   defer heartbeat.Stop()   go func() {         // 发送心跳，除了检测心跳外，还有防止Follower发送选举    for _ = range heartbeat.C {     flush <- struct{}{}    }   }()     for {    select {    case q := <-s.quit:     s.handleQuit(q)     return    // 收到命令    case t := <-s.commandChan:     // Append the command to our (leader) log     s.logGeneric("got command, appending")     currentTerm := s.term     entry := logEntry{      Index:           s.log.lastIndex() + 1,      Term:            currentTerm,      Command:         t.Command,      commandResponse: t.CommandResponse,     }             // 追加日志     if err := s.log.appendEntry(entry); err != nil {      t.Err <- err      continue     }     s.logGeneric(      "after append, commitIndex=%d lastIndex=%d lastTerm=%d",      s.log.getCommitIndex(),      s.log.lastIndex(),      s.log.lastTerm(),     )       // Now that the entry is in the log, we can fall back to the     // normal flushing mechanism to attempt to replicate the entry     // and advance the commit index. We trigger a manual flush as a     // convenience, so our caller might get a response a bit sooner.             // 这里将日志同步放到了同步队列就返回给客户端了，正常来说，需要"共识"达成才返回给客户端     go func() { flush <- struct{}{} }()     t.Err <- nil          // 收到配置变更    case t := <-s.configurationChan:     // Attempt to change our local configuration     if err := s.config.changeTo(makePeerMap(t.Peers...)); err != nil {      t.Err <- err      continue     }       // Serialize the local (C_old,new) configuration     encodedConfiguration, err := s.config.encode()     if err != nil {      t.Err <- err      continue     }       // We're gonna write+replicate that config via log mechanisms.     // Prepare the on-commit callback.     entry := logEntry{      Index:           s.log.lastIndex() + 1,      Term:            s.term,      Command:         encodedConfiguration,      isConfiguration: true,      committed:       make(chan bool),     }     go func() {                 // 当日志被commited时，committed将被回调      committed := <-entry.committed      if !committed {       s.config.changeAborted()       return      }                // 日志被committed了，说明其他节点都应用了最新的配置，所以当前的节点配置也需要更新      s.config.changeCommitted()      if _, ok := s.config.allPeers()[s.id]; !ok {                     // 当前节点已被新集群剔除       s.logGeneric("leader expelled; shutting down")       q := make(chan struct{})       s.quit <- q                     // 节点已退出       <-q      }     }()             // 日志追加     if err := s.log.appendEntry(entry); err != nil {      t.Err <- err      continue     }      case <-flush:             // 获取需要同步的节点     recipients := s.config.allPeers().except(s.id)       // Special case: network of 1     if len(recipients) <= 0 {      ourLastIndex := s.log.lastIndex()      if ourLastIndex > 0 {       if err := s.log.commitTo(ourLastIndex); err != nil {        s.logGeneric("commitTo(%d): %s", ourLastIndex, err)        continue       }       s.logGeneric("after commitTo(%d), commitIndex=%d", ourLastIndex, s.log.getCommitIndex())      }      continue     }       // Normal case: network of at-least-2             // 并发同步日志     successes, stepDown := s.concurrentFlush(recipients, ni, 2*broadcastInterval())     if stepDown {                 // 节点已被卸任      s.logGeneric("deposed during flush")      s.state.Set(follower)      s.leader = unknownLeader      return     }       // Only when we know all followers accepted the flush can we     // consider incrementing commitIndex and pushing out another     // round of flushes.     if successes == len(recipients) {                 // 最小被同步的Index      peersBestIndex := ni.bestIndex()      ourLastIndex := s.log.lastIndex()      ourCommitIndex := s.log.getCommitIndex()      if peersBestIndex > ourLastIndex {       // safety check: we've probably been deposed       s.logGeneric("peers' best index %d > our lastIndex %d", peersBestIndex, ourLastIndex)       s.logGeneric("this is crazy, I'm gonna become a follower")       s.leader = unknownLeader       s.vote = noVote       s.state.Set(follower)       return      }      if peersBestIndex > ourCommitIndex {       // committed Leader Index                     if err := s.log.commitTo(peersBestIndex); err != nil {        s.logGeneric("commitTo(%d): %s", peersBestIndex, err)                         // 比如某个Follower在同步Index时失败了，        continue // oh well, next time?       }              if s.log.getCommitIndex() > ourCommitIndex {        // 继续同步日志        s.logGeneric("after commitTo(%d), commitIndex=%d -- queueing another flush", peersBestIndex, s.log.getCommitIndex())        go func() { flush <- struct{}{} }()       }      }     }    // 追加日志， 正常来说，Leader节点是不会受到该命令的，出现这种的可能是集群存在一个新的Leader节点，这命令就是该Leader发送过来的    case t := <-s.appendEntriesChan:     resp, stepDown := s.handleAppendEntries(t.Request)     s.logAppendEntriesResponse(t.Request, resp, stepDown)     t.Response <- resp     if stepDown {      s.logGeneric("after an appendEntries, deposed to Follower (leader=%d)", t.Request.LeaderID)      s.leader = t.Request.LeaderID      s.state.Set(follower)      return // deposed     }    // 受到投票请求    case t := <-s.requestVoteChan:     resp, stepDown := s.handleRequestVote(t.Request)     s.logRequestVoteResponse(t.Request, resp, stepDown)     t.Response <- resp     if stepDown {      s.logGeneric("after a requestVote, deposed to Follower (leader unknown)")      s.leader = unknownLeader      s.state.Set(follower)      return // deposed     }    }   }  }    // handleRequestVote will modify s.term and s.vote, but nothing else.  // stepDown means you need to: s.leader=unknownLeader, s.state.Set(Follower).  // 处理投票  // 可能会修改s.term和s.vote 的值; stepDown意味着需要设置s.leader = unkownLeader, s.state.Set(Follower)  func (s *Server)handleRequestVote(rv requestVote)(requestVoteResponse,bool) {   // Spec is ambiguous here; basing this (loosely!) on benbjohnson's impl     // If the request is from an old term, reject   if rv.Term < s.term {    return requestVoteResponse{     Term:        s.term,     VoteGranted: false,     reason:      fmt.Sprintf("Term %d < %d", rv.Term, s.term),    }, false   }     // If the request is from a newer term, reset our state   stepDown := false   if rv.Term > s.term {    // 本地节点落后于集群的其他节点，需要更新一下自己的任期号    s.logGeneric("requestVote from newer term (%d): we defer", rv.Term)    s.term = rv.Term    s.vote = noVote    s.leader = unknownLeader    stepDown = true   }     // Special case: if we're the leader, and we haven't been deposed by a more   // recent term, then we should always deny the vote   if s.state.Get() == leader && !stepDown {    // 如果本地节点是Leader，并且又不落后于req 节点，则投反对票    return requestVoteResponse{     Term:        s.term,     VoteGranted: false,     reason:      "already the leader",    }, stepDown   }     // If we've already voted for someone else this term, reject   // 如果已经投过票，则投失败票   if s.vote != 0 && s.vote != rv.CandidateID {    if stepDown {     panic("impossible state in handleRequestVote")    }    return requestVoteResponse{     Term:        s.term,     VoteGranted: false,     reason:      fmt.Sprintf("already cast vote for %d", s.vote),    }, stepDown   }     // If the candidate log isn't at least as recent as ours, reject   if s.log.lastIndex() > rv.LastLogIndex || s.log.lastTerm() > rv.LastLogTerm {    return requestVoteResponse{     Term:        s.term,     VoteGranted: false,     reason: fmt.Sprintf(      "our index/term %d/%d > %d/%d",      s.log.lastIndex(),      s.log.lastTerm(),      rv.LastLogIndex,      rv.LastLogTerm,     ),    }, stepDown   }     // We passed all the tests: cast vote in favor   s.vote = rv.CandidateID   s.resetElectionTimeout()   return requestVoteResponse{    Term:        s.term,    VoteGranted: true,   }, stepDown  }    // handleAppendEntries will modify s.term and s.vote, but nothing else.  // stepDown means you need to: s.leader=r.LeaderID, s.state.Set(Follower).  // 追加日志，需要注意的是，handleAppendEntries也会修改s.term和s.vote  // stepDown也会修改s.Leader, s,state  // 需要注意的是，本地节点的state不同时，其行为也是不用的  func (s *Server)handleAppendEntries(r appendEntries)(appendEntriesResponse,bool) {   // Spec is ambiguous here; basing this on benbjohnson's impl     // Maybe a nicer way to handle this is to define explicit handler functions   // for each Server state. Then, we won't try to hide too much logic (i.e.   // too many protocol rules) in one code path.     // If the request is from an old term, reject   if r.Term < s.term {    return appendEntriesResponse{     Term:    s.term,     Success: false,     reason:  fmt.Sprintf("Term %d < %d", r.Term, s.term),    }, false   }     // If the request is from a newer term, reset our state   stepDown := false   if r.Term > s.term {    s.term = r.Term    s.vote = noVote    stepDown = true   }     // Special case for candidates: "While waiting for votes, a candidate may   // receive an appendEntries RPC from another server claiming to be leader.   // If the leader’s term (included in its RPC) is at least as large as the   // candidate’s current term, then the candidate recognizes the leader as   // legitimate and steps down, meaning that it returns to follower state."   if s.state.Get() == candidate && r.LeaderID != s.leader && r.Term >= s.term {    s.term = r.Term    s.vote = noVote    stepDown = true   }     // In any case, reset our election timeout   s.resetElectionTimeout()     // Reject if log doesn't contain a matching previous entry   // 如果{PreLogIndex, PreLogTerm} 不是最新的条目，则失败   // [{1, 2},{1, 3}, {1,4},{1,5},{1,6}] => {1,5} => [{1, 2},{1, 3}, {1,4},{1,5}]   if err := s.log.ensureLastIs(r.PrevLogIndex, r.PrevLogTerm); err != nil {    return appendEntriesResponse{     Term:    s.term,     Success: false,     reason: fmt.Sprintf(      "while ensuring last log entry had index=%d term=%d: error: %s",      r.PrevLogIndex,      r.PrevLogTerm,      err,     ),    }, stepDown   }     // Process the entries   for i, entry := range r.Entries {    // Configuration changes requre special preprocessing    var pm peerMap    // 处理配置    if entry.isConfiguration {     commandBuf := bytes.NewBuffer(entry.Command)     if err := gob.NewDecoder(commandBuf).Decode(±); err != nil {      panic("gob decode of peers failed")     }       if s.state.Get() == leader {      // TODO should we instead just ignore this entry?      return appendEntriesResponse{       Term:    s.term,       Success: false,       reason: fmt.Sprintf(        "AppendEntry %d/%d failed (configuration): %s",        i+1,        len(r.Entries),        "Leader shouldn't receive configurations via appendEntries",       ),      }, stepDown     }       // Expulsion recognition     if _, ok := pm[s.id]; !ok {      entry.committed = make(chan bool)      go func() {       if <-entry.committed {        s.logGeneric("non-leader expelled; shutting down")        q := make(chan struct{})        s.quit <- q        <-q       }      }()     }    }      // Append entry to the log    if err := s.log.appendEntry(entry); err != nil {     return appendEntriesResponse{      Term:    s.term,      Success: false,      reason: fmt.Sprintf(       "AppendEntry %d/%d failed: %s",       i+1,       len(r.Entries),       err,      ),     }, stepDown    }      // "Once a given server adds the new configuration entry to its log, it    // uses that configuration for all future decisions (it does not wait    // for the entry to become committed)."    if entry.isConfiguration {     if err := s.config.directSet(pm); err != nil {      return appendEntriesResponse{       Term:    s.term,       Success: false,       reason: fmt.Sprintf(        "AppendEntry %d/%d failed (configuration): %s",        i+1,        len(r.Entries),        err,       ),      }, stepDown     }    }   }     // Commit up to the commit index.   //   // < ptrb> ongardie: if the new leader sends a 0-entry appendEntries   // with lastIndex=5 commitIndex=4, to a follower that has lastIndex=5   // commitIndex=5 -- in my impl, this fails, because commitIndex is too   // small. shouldn't be?   // <@ongardie> ptrb: i don't think that should fail   // <@ongardie> there are 4 ways an appendEntries request can fail: (1)   // network drops packet (2) caller has stale term (3) would leave gap in   // the recipient's log (4) term of entry preceding the new entries doesn't   // match the term at the same index on the recipient   //   // 出现这种情况的原因可能是本地节点运行到committed逻辑的时候出现了问题，或者说应答给Leader时，网络出现了问题等等。   // 这些情况都会造成数据不同步的情况，也就是本地节点的commiitted情况和Leader节点保存的Follower(本地节点)不一致   if r.CommitIndex > 0 && r.CommitIndex > s.log.getCommitIndex() {    if err := s.log.commitTo(r.CommitIndex); err != nil {     return appendEntriesResponse{      Term:    s.term,      Success: false,      reason:  fmt.Sprintf("CommitTo(%d) failed: %s", r.CommitIndex, err),     }, stepDown    }   }     // all good   return appendEntriesResponse{    Term:    s.term,    Success: true,   }, stepDown  }  </code></pre>    <h3>configuration.go</h3>    <pre>  <code class="language-go">var (          errConfigurationAlreadyChanging = errors.New("configuration already changing")  )    const (          cOld    = "C_old"          cOldNew = "C_old,new"  )    // configuration represents the sets of peers and behaviors required to  // implement joint-consensus.  type configuration struct {          sync.RWMutex          state     string          // 老配置          cOldPeers peerMap          // 新配置-》用于过度          cNewPeers peerMap  }    // newConfiguration returns a new configuration in stable (C_old) state based  // on the passed peers.  func newConfiguration(pm peerMap)*configuration {          return &configuration{           state:     cOld, // start in a stable state,           cOldPeers: pm,   // with only C_old          }  }    // directSet is used when bootstrapping, and when receiving a replicated  // configuration from a leader. It directly sets the configuration to the  // passed peers. It's assumed this is called on a non-leader, and therefore  // requires no consistency dance.  // 配置变更  func (c *configuration)directSet(pm peerMap)error {          c.Lock()          defer c.Unlock()            c.cOldPeers = pm          c.cNewPeers = peerMap{}          c.state = cOld          return nil  }    func (c *configuration)get(iduint64)(Peer,bool) {          c.RLock()          defer c.RUnlock()            if peer, ok := c.cOldPeers[id]; ok {           return peer, true          }          if peer, ok := c.cNewPeers[id]; ok {           return peer, true          }          return nil, false  }    func (c *configuration)encode()([]byte, error) {          buf := &bytes.Buffer{}          if err := gob.NewEncoder(buf).Encode(c.allPeers()); err != nil {           return []byte{}, err          }          return buf.Bytes(), nil  }    // allPeers returns the union set of all peers in the configuration.  func (c *configuration)allPeers()peerMap {          c.RLock()          defer c.RUnlock()            union := peerMap{}          for id, peer := range c.cOldPeers {           union[id] = peer          }          for id, peer := range c.cNewPeers {           union[id] = peer          }          return union  }    // pass returns true if the votes represented by the votes map are sufficient  // to constitute a quorum. pass respects C_old,new requirements, which dictate  // that any request must receive a majority from both C_old and C_new to pass.  // 共识判断  func (c *configuration)pass(votesmap[uint64]bool)bool {          c.RLock()          defer c.RUnlock()            // Count the votes          cOldHave, cOldRequired := 0, c.cOldPeers.quorum()          for id := range c.cOldPeers {           if votes[id] {            cOldHave++           }           if cOldHave >= cOldRequired {            break           }          }            // If we've already failed, we can stop here          if cOldHave < cOldRequired {           return false          }            // C_old passes: if we're in C_old, we pass          if c.state == cOld {           return true          }            // Not in C_old, so make sure we have some peers in C_new          if len(c.cNewPeers) <= 0 {           panic(fmt.Sprintf("configuration state '%s', but no C_new peers", c.state))          }            // Since we're in C_old,new, we need to also pass C_new to pass overall.          // It's important that we range through C_new and check our votes map, and          // not the other way around: if a server casts a vote but doesn't exist in          // a particular configuration, that vote should not be counted.          cNewHave, cNewRequired := 0, c.cNewPeers.quorum()          for id := range c.cNewPeers {           if votes[id] {            cNewHave++           }           if cNewHave >= cNewRequired {            break           }          }            return cNewHave >= cNewRequired  }    // 配置变更准备, prepare-change  func (c *configuration)changeTo(pm peerMap)error {          c.Lock()          defer c.Unlock()            if c.state != cOld {           return errConfigurationAlreadyChanging          }            if len(c.cNewPeers) > 0 {           panic(fmt.Sprintf("configuration ChangeTo in state '%s', but have C_new peers already", c.state))          }            c.cNewPeers = pm          c.state = cOldNew          return nil  }    // 提交变更逻辑  func (c *configuration)changeCommitted() {          c.Lock()          defer c.Unlock()            if c.state != cOldNew {           panic("configuration ChangeCommitted, but not in C_old,new")          }            if len(c.cNewPeers) <= 0 {           panic("configuration ChangeCommitted, but C_new peers are empty")          }            c.cOldPeers = c.cNewPeers          c.cNewPeers = peerMap{}          c.state = cOld  }    // 中断变更  func (c *configuration)changeAborted() {          c.Lock()          defer c.Unlock()            if c.state != cOldNew {           panic("configuration ChangeAborted, but not in C_old,new")          }            c.cNewPeers = peerMap{}          c.state = cOld  }  </code></pre>    <h2>Demo</h2>    <pre>  <code class="language-go">package main    import (          "bytes"          "fmt"          "hash/fnv"          "net/http"          "net/url"          "time"            "github.com/peterbourgon/raft"  )    func main() {          a := func(idxuint64, cmd []byte)[]byte {           fmt.Printf("%d, apply function: %s\n", idx, cmd)           return cmd          }            mustParseURL := func(rawURLstring)*url.URL {           u, _ := url.Parse(rawURL)           u.Path = ""           return u          }          mustNewHTTPPeer := func(u *url.URL)raft.Peer {           p, err := raft.NewHTTPPeer(u)           if err != nil {            panic(err)           }           return p          }          peersAddr := []string{           "127.0.0.1:7090",           "127.0.0.1:7091",           "127.0.0.1:7092",           "127.0.0.1:7093",           "127.0.0.1:7094"}          var ss []*raft.Server          for _, addr := range peersAddr {           hash := fnv.New64()           hash.Write([]byte(addr))           id := hash.Sum64()           hash.Reset()           s := raft.NewServer(id, &bytes.Buffer{}, a)           mux := http.NewServeMux()           raft.HTTPTransport(mux, s)           go func(addrstring) {            if err := http.ListenAndServe(addr, mux); err != nil {             panic(err)            }           }(addr)           ss = append(ss, s)          }          time.Sleep(time.Second)          for _, s := range ss {           s.SetConfiguration(            mustNewHTTPPeer(mustParseURL("http://127.0.0.1:7090")),            mustNewHTTPPeer(mustParseURL("http://127.0.0.1:7091")),            mustNewHTTPPeer(mustParseURL("http://127.0.0.1:7092")),            mustNewHTTPPeer(mustParseURL("http://127.0.0.1:7093")),            mustNewHTTPPeer(mustParseURL("http://127.0.0.1:7094")),           )           s.Start()          }            for {           cmd := []byte(time.Now().String())           cmdChan := make(chan []byte)           go ss[0].Command(cmd, cmdChan)           <-cmdChan           time.Sleep(time.Millisecond * 500)          }            time.Sleep(time.Hour)  }  </code></pre>    <p>Run</p>    <pre>  <code class="language-go">» go run raft-server.go 2>/dev/null       1, apply function: 2017-09-11 11:41:13.668460404 +0800 CST  1, apply function: 2017-09-11 11:41:13.668460404 +0800 CST  1, apply function: 2017-09-11 11:41:13.668460404 +0800 CST  1, apply function: 2017-09-11 11:41:13.668460404 +0800 CST  1, apply function: 2017-09-11 11:41:13.668460404 +0800 CST  2, apply function: 2017-09-11 11:41:14.169165702 +0800 CST  2, apply function: 2017-09-11 11:41:14.169165702 +0800 CST  2, apply function: 2017-09-11 11:41:14.169165702 +0800 CST  2, apply function: 2017-09-11 11:41:14.169165702 +0800 CST  2, apply function: 2017-09-11 11:41:14.169165702 +0800 CST  3, apply function: 2017-09-11 11:41:14.670873193 +0800 CST  3, apply function: 2017-09-11 11:41:14.670873193 +0800 CST  3, apply function: 2017-09-11 11:41:14.670873193 +0800 CST  3, apply function: 2017-09-11 11:41:14.670873193 +0800 CST  3, apply function: 2017-09-11 11:41:14.670873193 +0800 CST  4, apply function: 2017-09-11 11:41:15.171741805 +0800 CST  4, apply function: 2017-09-11 11:41:15.171741805 +0800 CST  4, apply function: 2017-09-11 11:41:15.171741805 +0800 CST  4, apply function: 2017-09-11 11:41:15.171741805 +0800 CST  4, apply function: 2017-09-11 11:41:15.171741805 +0800 CST  5, apply function: 2017-09-11 11:41:15.673498401 +0800 CST  5, apply function: 2017-09-11 11:41:15.673498401 +0800 CST  5, apply function: 2017-09-11 11:41:15.673498401 +0800 CST  5, apply function: 2017-09-11 11:41:15.673498401 +0800 CST  5, apply function: 2017-09-11 11:41:15.673498401 +0800 CST  6, apply function: 2017-09-11 11:41:16.175658603 +0800 CST  6, apply function: 2017-09-11 11:41:16.175658603 +0800 CST  6, apply function: 2017-09-11 11:41:16.175658603 +0800 CST  6, apply function: 2017-09-11 11:41:16.175658603 +0800 CST  6, apply function: 2017-09-11 11:41:16.175658603 +0800 CST  7, apply function: 2017-09-11 11:41:16.677758823 +0800 CST  </code></pre>    <p> </p>    <p>来自：http://laohanlinux.github.io/2017/09/11/raft源码分析/</p>    <p> </p>