1.go語(yǔ)言實(shí)現(xiàn)raft分發(fā)原理
main.go 文件代碼如下
package main
import (
"sync"
"math/rand"
"fmt"
"time"
"go/ast"
)
// 設(shè)置節(jié)點(diǎn)的個(gè)數(shù)為3
const raftCount =3
// 聲明leader
type AppendEntiresArgs struct {
Term int// 第幾任領(lǐng)導(dǎo)
leaderId int//設(shè)置領(lǐng)導(dǎo)編號(hào)
}
// 創(chuàng)建一個(gè)領(lǐng)導(dǎo)對(duì)象
var args =AppendEntiresArgs{0,-1}
// 聲明節(jié)點(diǎn)類(lèi)型
type Raft struct {
// 線程鎖
mu sync.Mutex
// 節(jié)點(diǎn)編號(hào)
me int
//當(dāng)前領(lǐng)導(dǎo)期數(shù)
currentTerm int
// 為哪個(gè)節(jié)點(diǎn)投票
votedFor int
//當(dāng)前節(jié)點(diǎn)的狀態(tài)
state int// 0 跟隨者 1候選人 2就是領(lǐng)導(dǎo)
// 厚厚一天發(fā)送消息時(shí)間
lastMessageTime int64
// 節(jié)點(diǎn)間發(fā)送消息的通道
message chan bool
// 選舉通道
eclectCh chan bool
// 心跳信號(hào)通道
heartBeat chan bool
// 返回心跳信號(hào)
heartBeatRe chan bool
// 超時(shí)時(shí)間
timeout int
// 設(shè)置當(dāng)前節(jié)點(diǎn)的領(lǐng)導(dǎo)
currentLeader int
}
// 產(chǎn)生隨機(jī)值
func randRange(min,max int64)int64 {
return rand.Int63n(max-min)+min
}
// 打印當(dāng)前時(shí)間
func printTime() {
fmt.Printf("%s",time.Now().String())
}
// 獲取當(dāng)前時(shí)間的ms
func millisecond()int64 {
return time.Now().UnixNano()/int64(time.Millisecond)
}
func main() {
// 創(chuàng)建三個(gè)節(jié)點(diǎn)
for i:=0;i<raftCount ;i++ {
// 創(chuàng)建三個(gè)raft
Make(i)
}
for{
}
}
// 通過(guò)make函數(shù)創(chuàng)建raft節(jié)點(diǎn)的對(duì)象
func Make(me int)*Raft {
rf :=&Raft{}
rf.me = me
rf.votedFor = -1
rf.state = 0
rf.timeout =0
rf.currentLeader = -1
rf.setTerm(0)
// 創(chuàng)建通道對(duì)象
rf.eclectCh = make(chan bool)
rf.message = make(chan bool)
rf.heartBeat = make(chan bool)
rf.heartBeatRe = make(chan bool)
rand.Seed(time.Now().UnixNano())
go rf.sendLeaderHeartBeat()
go rf.election()
return rf
}
// 創(chuàng)建raft對(duì)象的方法
func (rf *Raft) setTerm(term int) {
rf.currentTerm = term
}
// 設(shè)置節(jié)點(diǎn)發(fā)送心跳信號(hào)的方法
func (rf *Raft)sendLeaderHeartBeat() {
}
//設(shè)置節(jié)點(diǎn)投票的方法
func (rf *Raft)election () {
// 設(shè)置標(biāo)簽
var result bool
for {
// 設(shè)置超時(shí)時(shí)間
timeout := randRange(150,300)
// 設(shè)置每個(gè)節(jié)點(diǎn)的時(shí)間
rf.lastMessageTime = millisecond()
select {
// 延遲等待
case <-time.After(time.Duration(timeout)*time.Millisecond):
fmt.Println(rf.state)// 打印當(dāng)前節(jié)點(diǎn)的狀態(tài)
}
result= false
for !result{
// 選擇誰(shuí)作為leader
result = rf.election_one_round(&args)
}
}
}
// 選擇誰(shuí)為leader
func (rf *Raft)election_one_round(args *AppendEntiresArgs)bool {
var timeout int64
var done int
// 是否開(kāi)始心跳信號(hào)的產(chǎn)生
var triggerHeartbeat bool
last := millisecond()
timeout = 100
// 信號(hào)變量標(biāo)簽
success :=false
// 修改raft節(jié)點(diǎn)的狀態(tài)為candidate候選人狀態(tài)
rf.mu.Lock()
rf.becomeCandidate()
rf.mu.Unlock()
fmt.Println("start electing leader")
for {
for i:=0;i<raftCount ;i++ {
if i!= rf.me {
//拉選票
printTime()
go func() {
if args.leaderId <0 {
// 設(shè)置此節(jié)點(diǎn)開(kāi)始選舉
rf.eclectCh <- true
}
}()
}
}
// 設(shè)置投票數(shù)量
done =0
triggerHeartbeat = false
for i:=0;i<raftCount-1 ;i++ {
// 計(jì)算一下投票的數(shù)量
//raftCount-1 把自己去掉
select {
case ok:= <-rf.eclectCh:
if ok {
done ++
success = done>raftCount/2
if success && !triggerHeartbeat {
// 選舉成功了
triggerHeartbeat = true
rf.mu.Lock()
rf.brcomeLeader()
rf.mu.Unlock()
// 由leader向其他節(jié)點(diǎn)發(fā)送心跳信號(hào)
rf.heartBeat <- true
fmt.Println("leader發(fā)送心跳信號(hào)")
}
}
}
}
if(timeout+last <millisecond()||(done>raftCount/2 || rf.currentLeader>-1)) {
break
}else {
select {
// 延時(shí)操作
case <-time.After(time.Duration(10)*time.Millisecond):
}
}
}
return success
}
func(rf *Raft) becomeCandidate() {
rf.state = 1
rf.setTerm(rf.currentTerm +1)
rf.votedFor = rf.me
rf.currentLeader = -1
}
func (rf *Raft)brcomeLeader() {
rf.state = 2
rf.currentLeader = rf.me
}