利用Go語言實現(xiàn)Raft日志同步
在raft中，選取成功后集群就可以正常工作，一次正常的客戶端提案過程如下：
- 客戶端連接到leader，發(fā)起提案
- leader收到提案后將提案，包裝為一條日志
- leader將日志緩存到待提交日志
- leader發(fā)送日志到集群其他節(jié)點
- follower收到日志后，將日志緩存到待提交日志，并響應leader請求
- leader收到follower響應，檢查是否集群大多數(shù)節(jié)點已響應
- 集群中大多數(shù)節(jié)點已緩存日志后，leader提交日志，發(fā)送空日志要求follower提交日志
- leader響應客戶端提案已接收
raft中日志規(guī)則如下：
1.如果在不同節(jié)點日志中的兩條日志記錄有相同的任期、編號，則這兩條日志記錄具有相同的內容
- leader在單個任期中最多創(chuàng)建一個給定編號的日志記錄
- leader不會改變日志記錄所在的位置（leader不會覆蓋、刪除日志記錄）
2.如果在不同節(jié)點日志中的兩條日志記錄有相同的任期、編號，則這兩條日志之前的所有日志相同
leader發(fā)送日志記錄到follower會包含上次日志記錄編號、任期
follower在收到追加日志時請求時會進行一致性檢查，如檢查失敗，follower會拒絕追加請求，保證了日志與leader一致，一致性檢查失敗時leader會強迫follower接受leader日志，從而保證日志一致性。一致性檢查失敗分兩種情況，
a.follower日志存在缺失，收到失敗響應后leader會將發(fā)送日志編號減一（follower檢查失敗時也可以返回當前最新日志編號），發(fā)送follower缺失的日志
當節(jié)點下線過一段時間/網絡異常消息丟失會出現(xiàn)日志缺失
b.follower有多余/不一致的日志，收到失敗響應后leader會將發(fā)送日志編號減一（follower檢查失敗時也可以返回當前最后提交的日志編號），找到兩份日志中最新的一致記錄編號，刪除follower中不一致的部分
節(jié)點當選leader收到消息只追加到本地尚未同步到集群便異常下線，后續(xù)上線會出現(xiàn)內存中有不一致的日志，已提交日志不會出現(xiàn)不一致
日志在節(jié)點間的同步分兩步完成，
1.leader通過rpc將日志復制到其他節(jié)點
2.當leader確認日志已被復制到集群中大多數(shù)節(jié)點后，將日志持久化到磁盤，leader追蹤已提交的日志的最大編號，通過日志rpc（含心跳）發(fā)送該編號告知follower需提交日志
日志提交時會持久化當前編號及該編號之前的未提交日志
每條日志到包含客戶端的實際提案內容、leader任期、日志編號，定義日志記錄為如下結構：
- type 日志類型，當前只有一種日志，客戶端的提案
- term 日志產生的任期，編號和任期相同代表為同一條日志
- index 日志編號，單調增加
- data 提案的實際內容
enum EntryType {
NORMAL = 0;
}
message LogEntry {
EntryType type = 1;
uint64 term = 2;
uint64 index = 3;
bytes data = 4;
}定義日志整體結構
未提交日志，保存在內存切片中
日志提交時，取出切片待提交部分，進行持久化
type WaitApply struct {
done bool
index uint64
ch chan struct{}
}
type RaftLog struct {
logEnties []*pb.LogEntry // 未提交日志
storage Storage // 已提交日志存儲
commitIndex uint64 // 提交進度
lastAppliedIndex uint64 // 最后提交日志
lastAppliedTerm uint64 // 最后提交日志任期
lastAppendIndex uint64 // 最后追加日志
logger *zap.SugaredLogger
}定義日志持久化接口，實際存儲實現(xiàn)由外部提供
type Storage interface {
Append(entries []*pb.LogEntry)
GetEntries(startIndex, endIndex uint64) []*pb.LogEntry
GetTerm(index uint64) uint64
GetLastLogIndexAndTerm() (uint64, uint64)
Close()
}實現(xiàn)一致性檢查
已持久化的日志必然與leader一致，檢查一致時只需檢查內存中日志切片，存在以下幾種情況：
1.節(jié)點日志中有找到leader上次追加日志
I.為節(jié)點追加的最后一條日志
II.為節(jié)點內存切片中的某條日志
- 節(jié)點網絡波動，導致未響應leader，leader重發(fā)了記錄，清除重復日志記錄
- 節(jié)點作為leader期間有部分日志未同步到其他節(jié)點就失效，集群重新選舉，導致后續(xù)日志不一致，清除沖突日志(內存中后續(xù)日志)
III.為節(jié)點最后提交日志
如內存中存在日志記錄，則內存中的記錄皆不一致，清除內存日志記錄
2.節(jié)點未找到leader上次追加日志
I.存在相同日志編號記錄，任不相同
節(jié)點作為leader期間有部分日志未同步到其他節(jié)點就失效，集群重新選舉，導致使用了相同日志編號，清除沖突日志(相同任期的日志)，從節(jié)點未沖突部分開始重發(fā)
II.沒有相同日志編號記錄
日志缺失，需從最后提交開始重發(fā)
func (l *RaftLog) HasPrevLog(lastIndex, lastTerm uint64) bool {
if lastIndex == 0 {
return true
}
var term uint64
size := len(l.logEnties)
if size > 0 {
lastlog := l.logEnties[size-1]
if lastlog.Index == lastIndex {
term = lastlog.Term
} else if lastlog.Index > lastIndex {
// 檢查最后提交
if lastIndex == l.lastAppliedIndex { // 已提交日志必然一致
l.logEnties = l.logEnties[:0]
return true
} else if lastIndex > l.lastAppliedIndex {
// 檢查未提交日志
for i, entry := range l.logEnties[:size] {
if entry.Index == lastIndex {
term = entry.Term
// 將leader上次追加后日志清理
// 網絡異常未收到響應導致leader重發(fā)日志/leader重選舉使舊leader未同步數(shù)據(jù)失效
l.logEnties = l.logEnties[:i+1]
break
}
}
}
}
} else if lastIndex == l.lastAppliedIndex {
return true
}
b := term == lastTerm
if !b {
l.logger.Debugf("最新日志: %d, 任期: %d ,本地記錄任期: %d", lastIndex, lastTerm, term)
if term != 0 { // 當日志與leader不一致，刪除內存中不一致數(shù)據(jù)同任期日志記錄
for i, entry := range l.logEnties {
if entry.Term == term {
l.logEnties = l.logEnties[:i]
break
}
}
}
}
return b
}實現(xiàn)日志追加，將新的日志添加到內存切片，更新最后追加日志編號
func (l *RaftLog) AppendEntry(entry []*pb.LogEntry) {
??????? size := len(entry)
if size == 0 {
return
}
l.logEnties = append(l.logEnties, entry...)
l.lastAppendIndex = entry[size-1].Index
}實現(xiàn)日志提交
- follower可能未同步全部日志，同步時如節(jié)點日志已同步全部待提交日志，則提交待提交日志，否則提交索引已追加日志
- 取出日志中待提交部分，添加到持久化存儲，更新提交進度、內存切片
func (l *RaftLog) Apply(lastCommit, lastLogIndex uint64) {
// 更新可提交索引
if lastCommit > l.commitIndex {
if lastLogIndex > lastCommit {
l.commitIndex = lastCommit
} else {
l.commitIndex = lastLogIndex
}
}
// 提交索引
if l.commitIndex > l.lastAppliedIndex {
n := 0
for i, entry := range l.logEnties {
if l.commitIndex >= entry.Index {
n = i
} else {
break
}
}
entries := l.logEnties[:n+1]
l.storage.Append(entries)
l.lastAppliedIndex = l.logEnties[n].Index
l.lastAppliedTerm = l.logEnties[n].Term
l.logEnties = l.logEnties[n+1:]
l.NotifyReadIndex()
}
}定義新建函數(shù)，創(chuàng)建實例時需提供存儲實現(xiàn)
func NewRaftLog(storage Storage, logger *zap.SugaredLogger) *RaftLog {
lastIndex, lastTerm := storage.GetLastLogIndexAndTerm()
return &RaftLog{
logEnties: make([]*pb.LogEntry, 0),
storage: storage,
commitIndex: lastIndex,
lastAppliedIndex: lastIndex,
lastAppliedTerm: lastTerm,
lastAppendIndex: lastIndex,
logger: logger,
}
}實現(xiàn)了日志的一致性檢查、追加、提交，接下實現(xiàn)raft中日志處理邏輯，首先我們需要在leader節(jié)點中保存集群其他節(jié)點的日志同步進度
節(jié)點在切換為leader時會將進度重置
- 投票響應中會返回節(jié)點最新日志信息
- 未收到投票響應的，使用leader最新日志，在一致性檢查后動態(tài)更新
在集群使用中通過第一條消息確認網絡可用，后續(xù)假設網絡正常，消息發(fā)送即成功，不等待節(jié)點響應消息，直到出現(xiàn)同步失敗
- prevResp 記錄上次發(fā)送結果，初始時為flase
- pending 中記錄未發(fā)送完成的日志編號
- 消息發(fā)送時如 !prevResp && len(pending) 為true，表示上次發(fā)送未完成，延遲后續(xù)信息發(fā)送
- 一次消息發(fā)送成功后，prevResp標記為true，后續(xù)有待發(fā)送日志都直接發(fā)送
type ReplicaProgress struct {
MatchIndex uint64 // 已接收日志
NextIndex uint64 // 下次發(fā)送日志
pending []uint64 // 未發(fā)送完成日志
prevResp bool // 上次日志發(fā)送結果
maybeLostIndex uint64 // 可能丟失的日志,記上次發(fā)送未完以重發(fā)
}leader將日志記錄追加到本地，再廣播到集群
func (r *Raft) BroadcastAppendEntries() {
r.cluster.Foreach(func(id uint64, _ *ReplicaProgress) {
if id == r.id {
return
}
r.SendAppendEntries(id)
})
}
func (r *Raft) SendAppendEntries(to uint64) {
p := r.cluster.progress[to]
if p == nil || p.IsPause() {
return
}
nextIndex := r.cluster.GetNextIndex(to)
lastLogIndex := nextIndex - 1
lastLogTerm := r.raftlog.GetTerm(lastLogIndex)
maxSize := MAX_LOG_ENTRY_SEND
if !p.prevResp {
maxSize = 1
}
// var entries []*pb.LogEntry
entries := r.raftlog.GetEntries(nextIndex, maxSize)
size := len(entries)
if size > 0 {
r.cluster.AppendEntry(to, entries[size-1].Index)
}
r.send(&pb.RaftMessage{
MsgType: pb.MessageType_APPEND_ENTRY,
Term: r.currentTerm,
From: r.id,
To: to,
LastLogIndex: lastLogIndex,
LastLogTerm: lastLogTerm,
LastCommit: r.raftlog.commitIndex,
Entry: entries,
})
}- 從日志中取得最新日志編號，遍歷待追加日志，設置日志編號
- 追加日志到內存切片
- 更新leader追加進度
- 廣播日志到集群
func (r *Raft) AppendEntry(entries []*pb.LogEntry) {
lastLogIndex, _ := r.raftlog.GetLastLogIndexAndTerm()
for i, entry := range entries {
entry.Index = lastLogIndex + 1 + uint64(i)
entry.Term = r.currentTerm
}
r.raftlog.AppendEntry(entries)
r.cluster.UpdateLogIndex(r.id, entries[len(entries)-1].Index)
r.BroadcastAppendEntries()
}
func (c *Cluster) UpdateLogIndex(id uint64, lastIndex uint64) {
p := c.progress[id]
if p != nil {
p.NextIndex = lastIndex
p.MatchIndex = lastIndex + 1
}
}廣播日志與之前廣播心跳一致，遍歷集群信息發(fā)送到每個節(jié)點，發(fā)送按下述流程
檢查發(fā)送狀態(tài)，如上次發(fā)送未完成，暫緩發(fā)送
func (rp *ReplicaProgress) IsPause() bool {
return (!rp.prevResp && len(rp.pending) > 0)
}從節(jié)點同步進度中取得當前需發(fā)送日志編號
func (c *Cluster) GetNextIndex(id uint64) uint64 {
p := c.progress[id]
if p != nil {
return p.NextIndex
}
return 0
}從leader的日志中取到要發(fā)送的日志
func (l *RaftLog) GetEntries(index uint64, maxSize int) []*pb.LogEntry {
// 請求日志已提交，從存儲獲取
if index <= l.lastAppliedIndex {
endIndex := index + MAX_APPEND_ENTRY_SIZE
if endIndex >= l.lastAppliedIndex {
endIndex = l.lastAppliedIndex + 1
}
return l.storage.GetEntries(index, endIndex)
} else { // 請求日志未提交,從數(shù)組獲取
var entries []*pb.LogEntry
for i, entry := range l.logEnties {
if entry.Index == index {
if len(l.logEnties)-i > maxSize {
entries = l.logEnties[i : i+maxSize]
} else {
entries = l.logEnties[i:]
}
break
}
}
return entries
}
}更新節(jié)點發(fā)送進度，將節(jié)點待發(fā)送日志編號加一，將發(fā)送的日志編號加入未發(fā)送完成切片
上次發(fā)送成功時，假設本次也會成功，如發(fā)送失敗再回退發(fā)送進度
func (c *Cluster) AppendEntry(id uint64, lastIndex uint64) {
p := c.progress[id]
if p != nil {
p.AppendEntry(lastIndex)
}
}
func (rp *ReplicaProgress) AppendEntry(lastIndex uint64) {
rp.pending = append(rp.pending, lastIndex)
if rp.prevResp {
rp.NextIndex = lastIndex + 1
}
}日志發(fā)送后，是follower收到日志進行處理
進行一致性檢查
- 檢查成功，將日志追加到follower內存中，標記追加成功
- 檢查失敗，一致性檢查中已處理沖突日志，直接標記追加失敗
嘗試提交日志，每次日志消息都會包含leader提交進度，按leader提交進度，提交follower日志
響應leader本次追加結果
func (r *Raft) ReciveAppendEntries(mLeader, mTerm, mLastLogTerm, mLastLogIndex, mLastCommit uint64, mEntries []*pb.LogEntry) {
var accept bool
if !r.raftlog.HasPrevLog(mLastLogIndex, mLastLogTerm) { // 檢查節(jié)點日志是否與leader一致
r.logger.Infof("節(jié)點未含有上次追加日志: Index: %d, Term: %d ", mLastLogIndex, mLastLogTerm)
accept = false
} else {
r.raftlog.AppendEntry(mEntries)
accept = true
}
lastLogIndex, lastLogTerm := r.raftlog.GetLastLogIndexAndTerm()
r.raftlog.Apply(mLastCommit, lastLogIndex)
r.send(&pb.RaftMessage{
MsgType: pb.MessageType_APPEND_ENTRY_RESP,
Term: r.currentTerm,
From: r.id,
To: mLeader,
LastLogIndex: lastLogIndex,
LastLogTerm: lastLogTerm,
Success: accept,
})
}leader處理follower日志追加響應，響應分為日志追加成功、日志追加失敗
func (r *Raft) ReciveAppendEntriesResult(from, term, lastLogIndex uint64, success bool) {
leaderLastLogIndex, _ := r.raftlog.GetLastLogIndexAndTerm()
if success {
r.cluster.AppendEntryResp(from, lastLogIndex)
if lastLogIndex > r.raftlog.commitIndex {
// 取已同步索引更新到lastcommit
if r.cluster.CheckCommit(lastLogIndex) {
prevApplied := r.raftlog.lastAppliedIndex
r.raftlog.Apply(lastLogIndex, lastLogIndex)
r.BroadcastAppendEntries()
}
} else if len(r.raftlog.waitQueue) > 0 {
r.raftlog.NotifyReadIndex()
}
if r.cluster.GetNextIndex(from) <= leaderLastLogIndex {
r.SendAppendEntries(from)
}
} else {
r.logger.Infof("節(jié)點 %s 追加日志失敗, Leader記錄節(jié)點最新日志: %d ,節(jié)點最新日志: %d ", strconv.FormatUint(from, 16), r.cluster.GetNextIndex(from)-1, lastLogIndex)
??????? r.cluster.ResetLogIndex(from, lastLogIndex, leaderLastLogIndex)
r.SendAppendEntries(from)
}
}日志追加成功時
更新同步進度，更新節(jié)點已接收進度，從未發(fā)送完成切片中清除已發(fā)送部分，標記上次發(fā)送成功
func (c *Cluster) AppendEntryResp(id uint64, lastIndex uint64) {
p := c.progress[id]
if p != nil {
p.AppendEntryResp(lastIndex)
}
}
func (rp *ReplicaProgress) AppendEntryResp(lastIndex uint64) {
if rp.MatchIndex < lastIndex {
rp.MatchIndex = lastIndex
}
idx := -1
for i, v := range rp.pending {
if v == lastIndex {
idx = i
}
}
// 標記前次日志發(fā)送成功，更新下次發(fā)送
if !rp.prevResp {
rp.prevResp = true
rp.NextIndex = lastIndex + 1
}
if idx > -1 {
// 清除之前發(fā)送
rp.pending = rp.pending[idx+1:]
}
}檢查follower數(shù)據(jù)同步進度，判斷響應對應日志編號是否在集群中大多數(shù)節(jié)點已同步
func (c *Cluster) CheckCommit(index uint64) bool {
// 集群達到多數(shù)共識才允許提交
incomingLogged := 0
for id := range c.progress {
if index <= c.progress[id].MatchIndex {
incomingLogged++
}
}
incomingCommit := incomingLogged >= len(c.progress)/2+1
return incomingCommit
}集群達成多數(shù)共識時，提交日志，繼續(xù)廣播日志
當響應follower待發(fā)送日志編號小于leader最新日志時繼續(xù)發(fā)送日志
當日志追加失敗時
按follower響應的日志進度重置日志同步進度，標記上次發(fā)送失敗，以延緩發(fā)送起始日志編號與follower不一致的日志，直到日志正確追加
func (c *Cluster) ResetLogIndex(id uint64, lastIndex uint64, leaderLastIndex uint64) {
p := c.progress[id]
if p != nil {
p.ResetLogIndex(lastIndex, leaderLastIndex)
}
}
func (rp *ReplicaProgress) ResetLogIndex(lastLogIndex uint64, leaderLastLogIndex uint64) {
// 節(jié)點最后日志小于leader最新日志按節(jié)點更新進度，否則按leader更新進度
if lastLogIndex < leaderLastLogIndex {
rp.NextIndex = lastLogIndex + 1
rp.MatchIndex = lastLogIndex
} else {
rp.NextIndex = leaderLastLogIndex + 1
rp.MatchIndex = leaderLastLogIndex
}
if rp.prevResp {
rp.prevResp = false
rp.pending = nil
}
}按更新后同步進度重發(fā)日志
修改raft新建函數(shù)，參數(shù)中加入存儲接口
func NewRaft(id uint64, storage Storage, peers map[uint64]string, logger *zap.SugaredLogger) *Raft {
raftlog := NewRaftLog(storage, logger)
...
}raft日志同步邏輯基本實現(xiàn)，接下來實現(xiàn)raftNode中的提案方法以追加日志，在raftNode主循環(huán)中已實現(xiàn)讀取recv通道，調用raft消息處理方法，當為leader時會將提案追加到日志，當前只需要將提案消息加入recv通道
當前l(fā)eader將提案加入讀寫通道后視為寫入成功，暫不實現(xiàn)集群多數(shù)共識后響應客戶端
要實現(xiàn)多數(shù)響應后通知，可添加一個新的結構含RaftMessage和一個channel，在raftlog添加一個等待隊列，當raft處理追加消息時，將日志最后一條記錄的日志編號通過channel返回給提案方法，提案方法再將channel放入raftlog中等待隊列，提交日志檢查等待隊列待通知對象，通過channel通知提案方法指定日志編號已提交
func (n *RaftNode) Propose(ctx context.Context, entries []*pb.LogEntry) error {
msg := &pb.RaftMessage{
MsgType: pb.MessageType_PROPOSE,
Term: n.raft.currentTerm,
Entry: entries,
}
return n.Process(ctx, msg)
}修改raftNode新建函數(shù)添加存儲接口，存儲實現(xiàn)在下篇lsm中實現(xiàn)
func NewRaftNode(id uint64, storage Storage, peers map[uint64]string, logger *zap.SugaredLogger) *RaftNode {
node := &RaftNode{
raft: NewRaft(id, storage, peers, logger),
...
}
...
}修改raft server中批量發(fā)送消息方法，將多個日志記錄合并到一個raft meassage進行發(fā)送
func (p *Peer) SendBatch(msgs []*pb.RaftMessage) {
p.wg.Add(1)
var appEntryMsg *pb.RaftMessage
var propMsg *pb.RaftMessage
for _, msg := range msgs {
if msg.MsgType == pb.MessageType_APPEND_ENTRY {
if appEntryMsg == nil {
appEntryMsg = msg
} else {
size := len(appEntryMsg.Entry)
if size == 0 || len(msg.Entry) == 0 || appEntryMsg.Entry[size-1].Index+1 == msg.Entry[0].Index {
appEntryMsg.LastCommit = msg.LastCommit
appEntryMsg.Entry = append(appEntryMsg.Entry, msg.Entry...)
} else if appEntryMsg.Entry[0].Index >= msg.Entry[0].Index {
appEntryMsg = msg
}
}
} else if msg.MsgType == pb.MessageType_PROPOSE {
if propMsg == nil {
propMsg = msg
} else {
propMsg.Entry = append(propMsg.Entry, msg.Entry...)
}
} else {
p.send(msg)
}
}
if appEntryMsg != nil {
p.send(appEntryMsg)
}
if propMsg != nil {
p.send(propMsg)
}
p.wg.Done()
}通過上述代碼實現(xiàn)了提案到leader，leader包裝為日志，同步到集群的過程，后續(xù)將通過lsm實現(xiàn)日志落盤并將raft server作為一個簡單的kv數(shù)據(jù)庫。
參考：https://github.com/etcd-io/etcd
到此這篇關于利用Go語言實現(xiàn)Raft日志同步的文章就介紹到這了,更多相關Go Raft日志同步內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
本文將介紹Go語言中字符串的實戰(zhàn)操作,通過本文的學習,讀者將掌握Go語言中字符串的常用操作,為實際開發(fā)提供幫助,需要的朋友可以參考下2023-10-10
本文主要介紹了go module構建項目的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-03-03
Golang 處理浮點數(shù)遇到的精度問題(使用decimal)
本文主要介紹了Golang 處理浮點數(shù)遇到的精度問題，不使用decimal會出大問題，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-02-02
一文帶你掌握Go語言I/O操作中的io.Reader和io.Writer
在?Go?語言中,io.Reader?和?io.Writer?是兩個非常重要的接口,它們在許多標準庫中都扮演著關鍵角色,下面就跟隨小編一起學習一下它們的使用吧2025-01-01
這篇文章主要介紹了解決golang中container/list包中的坑，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-04-04
這篇文章主要介紹了Go語言生成素數(shù)的方法,實例分析了Go語言生成素數(shù)的技巧,需要的朋友可以參考下2015-03-03
fmt包實現(xiàn)了類似C語言printf和scanf的格式化I/O,主要分為向外輸出內容和獲取輸入內容兩大部分,本文就來介紹一下GoLang語法之標準庫fmt.Printf的使用,感興趣的可以了解下2023-10-10
這篇文章主要介紹了golang值接收者和指針接收者的區(qū)別,它和函數(shù)的區(qū)別在于方法有一個接收者，給一個函數(shù)添加一個接收者，那么它就變成了方法，接收者可以是值接收者，也可以是指針接收者，本文通過實例代碼介紹的非常詳細，需要的朋友可以參考下2022-08-08
golang中sync.Map并發(fā)創(chuàng)建、讀取問題實戰(zhàn)記錄
這篇文章主要給大家介紹了關于golang中sync.Map并發(fā)創(chuàng)建、讀取問題的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-07-07
在Go語言中，切片（Slice）和指針的切片（即切片中每個元素都是指向某種數(shù)據(jù)類型的指針）是兩個不同的概念，它們各自具有特定的用途和優(yōu)勢，這篇文章主要介紹了go切片和指針切片,需要的朋友可以參考下2024-04-04