【分布式系统设计】实战Raft：主节点选举

Techdemic 2020-04-05

234

在之前的文章中，笔者介绍了一下分布式系统中的共识问题和Raft算法（分布式共识与Raft协议）。因为共识问题在分布式系统中属于最难的一类问题之一，所以就算Raft协议以作为最易懂的共识协议而闻名，想在短时间内理解并掌握它也是一件比较困难的事情，笔者收到了很多朋友的反馈说看不太懂上一篇文章，因此决定用多篇文章并附上代码来逐一攻破各个难点，第一篇文章将手把手教读者实现Raft的主节点选举。

本篇文章所有代码都用Go实现，如果读者想熟悉Go的编程模式可以通过这篇文章熟悉Go的channel和select原语，因为本篇文章将大量使用它们。

回顾

如上图所示，Raft每个节点有三个状态，分别是：Leader
， Candidate
和 Follower
。当集群启动时，每个节点的状态都是Follower，因为集群中没有Leader，所以在一个随机的Election Timeout之后，某一节点转换成Candidate状态并发起投票，如果获得多数票，则此节点成为Leader。这里“随机”的Election Timeout非常重要，因为如果每个节点都使用一个固定的Timeout的话，他们都会几乎在同一时刻发起自己的Election，导致谁也当不上Leader，这种情况称之为 SplitVote
。当Candidate成功当选为Leader时，发起会定期给所有Follower发送Hearbeat信号以刷新它们的Election Timer。当Leader通过收到的RPC请求或者回复发现有Term比自己高的Leader时，马上转变成Follower。

有很多人不理解Term在共识协议中的重要性，这里笔者再解释一下。Term可以理解为任期，用来防止集群中出现多个Leader。假设没有Term，当一个Follower错误地认为目前的Leader已经宕机时（由于network partition导致），它将发起投票并成为新的Leader，导致集群中有两个节点认为自己是Leader，但是由于没有任期，其他节点无法识别谁才是最新的合法leader。当有了Term之后，各个节点只需承认Term最大的节点为Leader就好。就比如川普是美国第45任总统，但是由于疫情在家隔离他并不知道自己已经被弹劾了，继续向国防部发号施令说我是第45任总统你们要听我的话，但是国防部回复他说Rick Sanchez现在已经是第46届总统了，你的话已经不管用了。于是川普变成了一个普通美国公民，为了重新当上总统，他需要发起第47届总统选举。

目标

上图是笔者从论文中截取的与Leader Election相关的描述，在本文中我们将一字不漏地将它们实现出来。

定义Raft节点

首先我们定义一个 Raft
类，除了论文中提到的几个状态，笔者还定义了几个信号量， killSig
用来通知各个goroutine节点已经被kill掉了， hearbeatSig
用来刷新Follower的Election Timer， stopListingHeartbeatSig
用来通知专属Follower的goroutine退出， stopSendingHeartbeatSig
用来通知专属Leader的goroutine退出。

type Raft struct {
  mu        sync.Mutex          // Lock to protect shared access to this peer's state
  peers     []*labrpc.ClientEnd // RPC end points of all peers
  me        int                 // this peer's Index into peers[]


  // Paper-defined Raft states
  currentTerm int
  votedFor    int
  log         []*LogEntry


  commitIndex int
  lastApplied int


  nextIndex  []int
  matchIndex []int


  // Self-defined Raft states
  isLeader bool


  // None raft related
  killSig                   chan struct{}
  heartbeatSig              chan struct{}
  stopListeningHeartbeatSig chan struct{}
  stopSendingHeartbeatSig   chan struct{}
}

节点启动

在以下代码中， Make
函数将启动并返回一个Raft对象。在初始化节点状态时， rf.isLeader
为 false
，对应了状态图中所有节点初始状态为 Follower
的情况。在函数的最后启动了 rf.listenHearbeatLoop
来监听来自可能的Leader的心跳信号，同时在Election Timeout之内没有收到心跳的话，节点将发起投票。

Make
函数有一个参数是 peers
，可以看成其他所有节点的地址，用来发送和接受RPC请求。

func Make(peers []*labrpc.ClientEnd, me int,
  persister *Persister, applyCh chan ApplyMsg) *Raft {
  rf := &Raft{}
  rf.peers = peers
  rf.me = me


  // Your initialization code here (2A, 2B, 2C).


  // Initialize Raft states
  rf.currentTerm = -1
  rf.votedFor = -1
  rf.log = make([]*LogEntry, 0)
  rf.commitIndex = -1
  rf.lastApplied = -1
  rf.nextIndex = make([]int, len(peers))
  rf.matchIndex = make([]int, len(peers))


  rf.isLeader = false


  // Initialize signal channels
  rf.killSig = make(chan struct{}, 1)
  rf.heartbeatSig = make(chan struct{}, 1)
  rf.stopListeningHeartbeatSig = make(chan struct{}, 1)
  rf.stopSendingHeartbeatSig = make(chan struct{}, 1)


  go rf.listenHeartbeatLoop()


  return rf
}

发起投票

在 listenHeartbeatLoop
中，Follower节点在每个循环中创建一个随机的timer来实现Election Timeout，然后同时监听来自 rf.hearbeatSig
, timer.C
, rf.killSig
和 rf.stopListenHeartbeatSig
的信号，当在Election Timeout之内(timer.C 触发)没有收到心跳信号，节点将通过 rf.startElection()
发起投票。

func (rf *Raft) listenHeartbeatLoop() {
  defer DPrintf("raft %d exiting listenHeartbeatLoop\n", rf.me)
  for {
    timer := time.NewTimer(time.Duration(rand.Intn(maxElectionTimeout-minElectionTimeout)+minElectionTimeout) * time.Millisecond)
    select {
    case <-rf.heartbeatSig:
      continue
    case <-timer.C:
      // Turn to candidate and request for votes
      DPrintf("raft %d turning Candidate\n", rf.me)
      go rf.startElection()
    case <-rf.stopListeningHeartbeatSig:
      return
    case <-rf.killSig:
      return
    }
  }
}

在 startElection
方法开始先锁住节点，然后将 rf.currentTerm
加一，更新 rf.votedFor
为自己，并记录下 electionTerm
。随后并发地向除自己外所有节点发送 rf.requestVote()
请求，然后等待所有请求完成并收集票数。当票数收集完毕时，节点首先判断自己的 rf.currentTerm
是否为当初的 electionTerm
，这一点非常重要，因为节点在等候票数的时候可能收到了其他节点的 RequestVote
请求并且投出了自己的一票，导致当前term比electionTerm要高。如果这种情况出现的话，本轮投票立即终止，否则节点将会通过错误的任期成为Leader。在本文中，所有RPC请求结束后都会做一次这种验证。最后当节点确认自己收到的票数过半时，通过 rf.turnLeader()
成为Leader。

rf.turnLeader()
将 rf.isLeader
设为True，并发送 rf.stopListeningHearbeatSig
关闭 rf.listenHeartbeatLoop
。最后启动 rf.sendHeartbeatLoop()
给所有folloer发送心跳。

这里有一个非常容易犯的错误，那就是在发送RPC的时候没有将 rf.mu
解锁，这会导致当两个节点同时投票时出现死锁的bug。

func (rf *Raft) startElection() {
  rf.mu.Lock()
  if rf.isLeader {
    rf.mu.Unlock()
    return
  }
  rf.currentTerm += 1
  rf.votedFor = rf.me
  electionTerm := rf.currentTerm
  rf.mu.Unlock()


  votes := make(chan bool, len(rf.peers)-1)
  for server := range rf.peers {
    if server == rf.me {
      continue
    }
    go rf.requestVote(electionTerm, server, votes)
  }


  numGranted := 1
  for i := 0; i < len(rf.peers)-1; i++ {
    if vote := <-votes; vote {
      numGranted++
    }
    if numGranted > len(rf.peers)/2 {
      break
    }
  }


  rf.mu.Lock()
  defer rf.mu.Unlock()


  if electionTerm != rf.currentTerm {
    return
  }


  if numGranted > len(rf.peers)/2 {
    if !rf.isLeader {
      rf.turnLeader()
    }
  }
  return
}


func (rf *Raft) turnLeader() {
  DPrintf("raft %d turning to leader\n", rf.me)
  rf.isLeader = true
  select {
  case rf.stopListeningHeartbeatSig <- struct{}{}:
  default:
  }
  go rf.sendHeartbeatLoop()
}

rf.requestVote
对单个其他节点发送求票请求。一开始在请求中设置term与本节点的id。LastLogIndex与LastLogTerm暂时忽略，因为这属于第二篇文章(Log replication)的范畴。当收到回复时，节点首先判断回复的Term是否大于 rf.currentTerm
，一旦条件为真，说明了比自己term高的Leader已经产生了，此时需要更新自己的 rf.currentTerm
并且放弃本轮票选。如果没有发现比自己term高的Leader，节点判断 reply.VoteGranted
是否为真，如果为真，说明得到了对面节点的投票。

func (rf *Raft) requestVote(electionTerm int, server int, votes chan bool) {
  args := &RequestVoteArgs{
    Term:        electionTerm,
    CandidateId: rf.me,
    // TODO: fill these two
    LastLogIndex: 0,
    LastLogTerm:  0,
  }
  reply := &RequestVoteReply{}


  DPrintf("raft %d sending RequestVote %v to raft %d \n", rf.me, args, server)
  if success := rf.sendRequestVote(server, args, reply); !success {
    DPrintf("raft %d sendRequestVote RPC %v to raft %d failure\n", rf.me, args, server)
    votes <- false
    return
  }
  DPrintf("raft %d got RequestVote reply %v from raft %d\n", rf.me, reply, server)


  rf.mu.Lock()
  defer rf.mu.Unlock()


  if args.Term != rf.currentTerm {
    votes <- false
    return
  }


  if reply.Term > rf.currentTerm {
    if rf.isLeader {
      rf.turnFollower()
    }
    rf.currentTerm = reply.Term
    rf.votedFor = -1
    votes <- false
    return
  }


  if reply.VoteGranted {
    votes <- true
    return
  }
  votes <- false
  return
}

处理投票请求

rf.RequestVote
方法定义了一个节点收到其他节点RequestVote RPC时的行为。当节点收到其他节点的求票请求时，首先判断对方的Term是否小于自己的 rf.currentTerm
，如果小于，那么告知对方最新的Term并且拒绝请求。如果对方的Term等于自己的Term，并且对方的id不等于自己的 rf.VotedFor
，说明节点在这个Term已经给其他节点（可以包括自己）投票了，也拒绝请求。否则将票投给对方并更新自己的Term。在方法最后最好刷新一下节点的hearbeat信号，防止在投票之后和收到新Leader心跳信号之前自己发起新一轮投票。

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
  DPrintf("raft %d received RequestVoteRequest %v from raft %d\n", rf.me, args, args.CandidateId)
  defer DPrintf("raft %d returning RequestVoteResponse %v to raft %d\n", rf.me, reply, args.CandidateId)
  // Your code here (2A, 2B).
  rf.mu.Lock()
  defer rf.mu.Unlock()
  
  if args.Term < rf.currentTerm {
    reply.Term = rf.currentTerm
    reply.VoteGranted = false
    return
  }
  
  if args.Term == rf.currentTerm && rf.votedFor >= 0 && rf.votedFor != args.CandidateId {
    reply.Term = args.Term
    reply.VoteGranted = false
    return
  }
  // TODO: implement log checking below


  if rf.isLeader {
    rf.turnFollower()
  }
  // update current Term and grant vote
  rf.currentTerm = args.Term
  rf.votedFor = args.CandidateId
  reply.Term = args.Term
  reply.VoteGranted = true
  rf.refreshHeartbeat()
}

心跳发送

在 发起投票
部分中我们提到了当节点成功当选Leader后会启动 rf.sendHearbeatLoop()
来持续地给所有节点发送心跳以维持自己Leader的地位。这个循环会每隔一段时间调用 rf.sendHeatbeats()
来广播一次心跳。rf.sendHeartbeats()
会并发地向所有Follower节点调用 rf.sendHeartbeat()
以提高性能。rf.sendHeartbeat()
通过 AppendEntriesRPC
向对方节点发送自己的Term与id，当收到回复后，判断回复的term是否大于自己的 rf.currentTerm
，如果大于，更新自己的term并马上step down成为follower。

func (rf *Raft) sendHeartbeatLoop() {
  rf.sendHeartbeats()
  defer DPrintf("raft %d exiting sendHeartbeatLoop\n", rf.me)
  ticker := time.NewTicker(heartbeatPeriod * time.Millisecond)
  for {
    select {
    case <-ticker.C:
      go rf.sendHeartbeats()
    case <-rf.stopSendingHeartbeatSig:
      return
    case <-rf.killSig:
      return
    }
  }
}


func (rf *Raft) sendHeartbeats() {
  rf.mu.Lock()
  if !rf.isLeader {
    rf.mu.Unlock()
    return
  }
  term := rf.currentTerm
  rf.mu.Unlock()


  replies := make(chan bool, len(rf.peers)-1)
  for server := range rf.peers {
    if server == rf.me {
      continue
    }
    go rf.sendHeartbeat(term, server, replies)
  }
  numSuccess := 1
  for i := 0; i < len(rf.peers)-1; i++ {
    if reply := <-replies; reply {
      numSuccess++
    }
  }
  if numSuccess <= len(rf.peers)/2 {
    rf.mu.Lock()
    defer rf.mu.Unlock()
    if rf.isLeader {
      rf.turnFollower()
    }
  }
}


func (rf *Raft) sendHeartbeat(term int, server int, replies chan bool) {
  args := &AppendEntriesArgs{
    Term:     term,
    LeaderId: rf.me,
    // TODO: fill the field below
    PrevLogIndex: 0,
    PrevLogTerm:  0,
    Entries:      nil,
    LeaderCommit: 0,
  }
  reply := &AppendEntriesReply{}


  if success := rf.sendAppendEntries(server, args, reply); !success {
    DPrintf("raft %d sendAppendEntries RPC %v to raft %d failure\n", rf.me, args, server)
    replies <- false
    return
  }


  rf.mu.Lock()
  defer rf.mu.Unlock()


  if args.Term != rf.currentTerm {
    replies <- false
    return
  }


  if !reply.Success {
    if reply.Term > rf.currentTerm {
      if rf.isLeader {
        rf.turnFollower()
      }
      rf.currentTerm = reply.Term
      rf.votedFor = -1
    }
    replies <- false
    return
  }
  replies <- true
}

心跳接收

当一个节点收到另外一个节点的 AppendEntries
(心跳)请求时，首先判断对方的term是否小于自己的 rf.currentTerm
，如果小于，那么告知对方最新的term并拒绝请求。否则接受请求并刷新自己的heartbeat信号。

值得注意的是本文中所有信号的刷新都是non-blocking的，这是因为channel大小有限，防止没有必要的阻塞。

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
  DPrintf("raft %d received AppendEntriesRequest %v from raft %d\n", rf.me, args, args.LeaderId)
  defer DPrintf("raft %d return AppendEntiresResponse %v to raft %d\n", rf.me, reply, args.LeaderId)
  rf.mu.Lock()
  defer rf.mu.Unlock()


  if args.Term < rf.currentTerm {
    reply.Term = rf.currentTerm
    reply.Success = false
    return
  }


  reply.Term = args.Term
  reply.Success = true


  rf.currentTerm = args.Term
  if rf.isLeader {
    rf.votedFor = -1
    rf.turnFollower()
  }


  rf.refreshHeartbeat()
  // TODO: implement append entries


  return


}


func (rf *Raft) refreshHeartbeat() {
  select {
  case rf.heartbeatSig <- struct{}{}:
  default:
  }
}

总结

自此我们已经完全实现了Raft的 LeaderElection
，代码量也就三百来行，但是其对并发编程和Raft论文的的理解都有着比较高的要求。相信通过细读这篇文章，读者能够对Raft有第一步的了解，在第二篇文章中笔者将带大家实现一遍 Logreplication
。

读者如果想读一读完整代码并运行单元测试，可以访问笔者的GitHub的这个文件夹链接：

https://github.com/Lancerchiang/6.824/tree/master/src/raft

将代码clone下来之后需要执行一下

exportGOPATH=<some_path_prefix>/6.824

来设置GOPATH，并执行

git checkout9b5e2f5f2b7128c3360c957c60f8d006ccc83382

以切换到Leader Election的commit节点，否则将会看到包含了Log Replication的完整实现。

往期精彩回顾

分布式共识与Raft协议
实战Apache Kafka
CSP范式编程

我就知道你“在看”

数据库

文章转载自Techdemic，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。